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

Subscriber interface สมัครรับจาก publishers เพื่อรับ items ผ่าน onNext(), error messages ผ่าน onError(), หรือสัญญาณจบผ่าน onComplete() ก่อนทุกอย่าง publisher จะเรียก onSubscribe()

public interface Subscriber<T> {
    public void onSubscribe(Subscription s);
    public void onNext(T t);
    public void onError(Throwable t);
    public void onComplete();
}

กฎสำหรับ Subscriber Interface

1. Subscriber ต้องเรียกผ่าน Subscription.request(long n) เพื่อรับสัญญาณ onNext()
2. Subscriber.onComplete() และ Subscriber.onError(Throwable t) ห้ามเรียกเมธอดบน Subscription หรือ Publisher
3. Subscriber.onComplete() และ Subscriber.onError(Throwable t) ต้องถือว่า Subscription ถูก cancel หลังได้รับสัญญาณ
4. Subscriber ต้องเรียก Subscription.cancel() บน Subscription ที่ได้รับหลังสัญญาณ onSubscribe() หากมี active Subscription อยู่แล้ว
5. Subscriber ต้องเรียก Subscription.cancel() หาก Subscription ไม่จำเป็นแล้ว
6. Subscriber ต้อง ensure ว่าการเรียก Subscription's request และ cancel methods ทำแบบ serial
7. Subscriber อาจยังได้รับ onNext() signals หนึ่งหรือมากหลังจากเรียก Subscription.cancel() หากยังมี requested elements ค้างอยู่ เมธอดนี้ไม่รับประกัน underlying cleanup ทันที
8. Subscriber ต้องรับสัญญาณ onComplete() พร้อมหรือไม่มี Subscription.request(long n) ก่อนหน้า
9. Subscriber ต้องรับสัญญาณ onError() พร้อมหรือไม่มี Subscription.request(long n) ก่อนหน้า
10. Subscriber ต้อง ensure ว่าการเรียก signal methods ทั้งหมดเกิดขึ้นก่อน processing ของ respective signals หมายถึง Subscriber ต้องดูแลการ publish signal ให้ processing logic อย่างเหมาะสม
11. Subscriber.onSubscribe() ต้องถูกเรียกสูงสุด 1 ครั้งต่อ Subscriber
12. การเรียก onSubscribe(), onNext(), onError() หรือ onComplete() ต้องคืนค่าปกติ ยกเว้น parameter ที่ให้มาเป็น null ต้องโยน NullPointerException ให้ caller

Subscriber Lifecycle

1. Publisher.subscribe(subscriber)
2. Publisher -> subscriber.onSubscribe(subscription)
3. Subscriber -> subscription.request(n)     // Demand
4. Publisher -> subscriber.onNext(item)      // Up to n items
5. Repeat 3-4 as needed
6. Terminal: Publisher -> subscriber.onComplete() OR onError(throwable)
7. Subscriber considers subscription cancelled
8. Optional: subscriber -> subscription.cancel()

Implementation Example

import java.util.concurrent.Flow.*;

public class SimpleSubscriber implements Subscriber<Integer> {
    private Subscription subscription;
    private final List<Integer> buffer = new ArrayList<>();
    private final int bufferSize;
    
    public SimpleSubscriber(int bufferSize) {
        this.bufferSize = bufferSize;
    }
    
    @Override
    public void onSubscribe(Subscription s) {
        this.subscription = s;
        s.request(bufferSize);  // Initial demand
    }
    
    @Override
    public void onNext(Integer item) {
        buffer.add(item);
        System.out.println("Received: " + item);
        
        // Request more when buffer half empty
        if (buffer.size() >= bufferSize / 2) {
            subscription.request(bufferSize / 2);
        }
    }
    
    @Override
    public void onError(Throwable t) {
        System.err.println("Error: " + t.getMessage());
        // subscription considered cancelled
    }
    
    @Override
    public void onComplete() {
        System.out.println("Completed. Total: " + buffer.size());
        // subscription considered cancelled
    }
}

Key Responsibilities

Backpressure Strategies

• Fixed buffer: Request fixed amount, refill when half empty
• Dynamic: Adjust demand based on processing speed
• Unbounded: request(Long.MAX_VALUE) (ระวัง OOM)
• Drop latest/oldest: Buffer full policy (ต้อง implement เอง)