กฎสำหรับ Publisher Interface ใน Java 9

Publisher เป็นผู้ให้ unbounded number of sequenced elements เผยแพร่ตาม demand ที่ได้รับจาก Subscribers Publisher<T> interface รับผิดชอบ publish elements ประเภท T และให้เมธอด subscribe() สำหรับ subscribers เชื่อมต่อ

public interface Publisher<T> {
    public void subscribe(Subscriber<? super T> s);
}

กฎสำหรับ Publisher Interface

1. จำนวน onNext() ทั้งหมดที่ Publisher ส่งให้ Subscriber ต้องน้อยกว่าหรือเท่ากับจำนวน elements ที่ Subscriber ขอผ่าน Subscription ตลอดเวลา
2. Publisher อาจส่ง onNext() น้อยกว่าที่ขอ และจบ Subscription โดยเรียก onComplete() หรือ onError()
3. onSubscribe(), onNext(), onError(), onComplete() ที่ส่งให้ Subscriber ต้องส่งแบบ serial (ลำดับชัดเจน)
4. ถ้า Publisher ล้มเหลว ต้องส่งสัญญาณ onError()
5. ถ้า Publisher สำเร็จ ต้องส่งสัญญาณ onComplete()
6. ถ้า Publisher ส่ง onError() หรือ onComplete() ให้ Subscriber แล้ว Subscription ของ Subscriber ต้องถูก cancel
7. เมื่อ terminal state ถูกส่งแล้ว (onError(), onComplete()) ห้ามมี signals เพิ่มเติม
8. ถ้า Subscription ถูก cancel Subscriber ต้องหยุดรับ signals
9. Publisher.subscribe() ต้องเรียก onSubscribe() บน Subscriber ที่ให้มา ก่อน signals อื่นๆ และคืนค่าปกติ ยกเว้น Subscriber เป็น null ต้องโยน NullPointerException
10. Publisher.subscribe() อาจเรียกได้หลายครั้งกับ Subscriber คนละตัวทุกครั้ง
11. Publisher อาจรองรับ multiple Subscribers และตัดสินใจว่าแต่ละ Subscription เป็น unicast หรือ multicast

Reactive Streams Specification

กฎเหล่านี้เป็นส่วนของ Reactive Streams Specification (java.util.concurrent.Flow) ที่มาตรฐาน化 asynchronous stream processing กับ non-blocking backpressure

Component Relationships

Publisher --subscribe()--> Subscriber
                |
                v
          Subscription (1:1 with Subscriber)
                |
                |-- request(n) --> Publisher produces n items
                |-- cancel() --> Stop production

Implementation Example

import java.util.concurrent.Flow.*;

public class SimplePublisher implements Publisher<Integer> {
    private final Iterator<Integer> iterator;
    
    SimplePublisher(int count) {
        this.iterator = IntStream.rangeClosed(1, count).iterator();
    }
    
    @Override
    public void subscribe(Subscriber<? super Integer> subscriber) {
        subscriber.onSubscribe(new Subscription() {
            @Override
            public void request(long n) {
                for (long i = 0; i < n && iterator.hasNext(); i++) {
                    subscriber.onNext(iterator.next());
                }
                if (!iterator.hasNext()) {
                    subscriber.onComplete();
                }
            }
            
            @Override
            public void cancel() {
                // Cleanup if needed
            }
        });
    }
}

Key Points

• Backpressure: Subscriber ควบคุม rate ผ่าน request(n)
• Asynchronous: ไม่บล็อก thread รอข้อมูล
• Non-blocking: Publisher ไม่รอ Subscriber process
• Error propagation: onError ส่งผ่าน chain
• Completion signal: onComplete แจ้งจบ stream