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

Subscription แชร์ระหว่าง Publisher และ Subscriber เพียงหนึ่งเดียว เพื่อ mediate data exchange นั่นคือเหตุผลที่ subscribe() method ไม่คืนค่า Subscription ที่สร้างขึ้น แตคืน void Subscription ถูกส่งให้ Subscriber ผ่าน onSubscribe() method callback เท่านั้น Subscription interface มี 2 เมธอด: request() และ cancel()

Syntax

public interface Subscription {
    public void request(long n);
    public void cancel();
}

กฎสำหรับ Subscription Interface

1. Subscription.request() และ Subscription.cancel() ต้องถูกเรียกเฉพาะภายใน context ของ Subscriber เท่านั้น
2. Subscription ต้องอนุญาตให้ Subscriber เรียก Subscription.request() แบบ synchronous จากภายใน onNext() หรือ onSubscribe()
3. Subscription.request() ต้องกำหนด upper bound สำหรับ synchronous recursion ระหว่าง Publisher และ Subscriber
4. Subscription.request() ควรเคารพ responsivity ของ caller โดยคืนค่าในเวลาที่เหมาะสม
5. Subscription.cancel() ต้องเคารพ responsivity ของ caller โดยคืนค่าในเวลาที่เหมาะสม และต้องเป็น thread-safe
6. ขณะ Subscription ไม่ถูก cancel Subscription.request(long n) ต้องจดทะเบียนจำนวน elements เพิ่มเติมที่จะผลิตให้ subscriber
7. ขณะ Subscription ไม่ถูก cancel Subscription.request(long n) ต้องโยน onError() พร้อม IllegalArgumentException ถ้า argument <= 0
8. ขณะ Subscription ไม่ถูก cancel Subscription.request(long n) อาจเรียก onNext() แบบ synchronous บน subscriber นี้หรืออื่นๆ
9. ขณะ Subscription ไม่ถูก cancel Subscription.request(long n) อาจเรียก onComplete() หรือ onError() แบบ synchronous บน subscriber นี้หรืออื่นๆ
10. ขณะ Subscription ไม่ถูก cancel Subscription.cancel() ต้องขอ Publisher หยุดส่ง signal ให้ Subscriber การดำเนินการอาจไม่มีผลทันที
11. ขณะ Subscription ไม่ถูก cancel Subscription.cancel() ต้องขอ Publisher drop references ไปยัง subscriber นั้น
12. การเรียก Subscription.cancel() และ Subscription.request() ต้องคืนค่าปกติ
13. Subscription ต้องรองรับ unbounded จำนวนการเรียก request และรองรับ demand ขึ้นถึง 2^63-1 ถ้า demand เท่าหรือมากกว่า 2^63-1 อาจถือว่า effectively unbounded โดย Publisher

Subscription Lifecycle

1. Publisher.subscribe(subscriber)
2. Publisher creates Subscription
3. Publisher calls subscriber.onSubscribe(subscription)
4. Subscriber calls subscription.request(n)  --> Demand signaled
5. Publisher calls subscriber.onNext(item)   --> Up to n times
6. Repeat 4-5 as needed
7. Terminal: subscriber.onComplete() OR subscriber.onError(throwable)
8. Subscription considered cancelled
9. Optional: subscriber calls subscription.cancel() explicitly

Backpressure Mechanism

• Subscriber ควบคุม flow rate ด้วย request(n)
• Publisher ต้องไม่ส่งมากกว่า n items ที่ request
• Subscriber สามารถ request เพิ่มได้ระหว่างการทำงาน
• Unbounded demand: request(Long.MAX_VALUE)

Common Implementation Patterns

// Simple synchronous subscription
subscription.request(n) {
    for (int i = 0; i < n && hasNext(); i++) {
        subscriber.onNext(next());
    }
    if (!hasNext()) subscriber.onComplete();
}

// Async subscription with buffer
subscription.request(n) {
    buffer.addDemand(n);
    drainBuffer(); // process in separate thread
}

Error Handling

• request(0) หรือ negative -> IllegalArgumentException via onError
• request after cancel -> ignored หรือ error
• cancel multiple times -> safe (idempotent)
• Exceptions in onNext -> propagated via onError