วิธี Implement Flow.Publisher Interface ใน Java 9

Publisher interface เป็นผู้ให้ unbounded number of sequenced elements เผยแพร่ตาม demand จาก Subscriber(s) ตอบสนองต่อการเรียก Publisher.subscribe(Subscriber) ลำดับการ invoke methods บน Subscriber คือ: onSubscribe() ตามด้วย onNext() จำนวนไม่จำกัด (ตามที่ Subscriber request) ตามด้วย onError() ถ้าล้มเหลว หรือ onComplete() เมื่อไม่มี elements เหลือแล้ว ตราบใดที่ Subscription ไม่ถูก cancel

Syntax

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

ตัวอย่าง Implementation

import java.util.concurrent.*;
import java.util.*;
import java.util.stream.*;

class SimplePublisher implements Flow.Publisher<Integer> {
    private final Iterator<Integer> iterator;
    
    SimplePublisher(int count) {
        this.iterator = IntStream.rangeClosed(1, count).iterator();
    }
    
    @Override
    public void subscribe(Flow.Subscriber<? super Integer> subscriber) {
        subscriber.onSubscribe(new Flow.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
            }
        });
    }
}

public class SimplePublisherImplTest {
    public static void main(String args[]) {
        new SimplePublisher(10).subscribe(new Flow.Subscriber<>() {
            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                // Could store subscription for later cancel/request more
            }
            
            @Override
            public void onNext(Integer item) {
                System.out.println("item = [" + item + "]");
            }
            
            @Override
            public void onError(Throwable throwable) {
                throwable.printStackTrace();
            }
            
            @Override
            public void onComplete() {
                System.out.println("complete");
            }
        });
    }
}

ผลลัพธ์

item = [1]
item = [2]
item = [3]
item = [4]
item = [5]
item = [6]
item = [7]
item = [8]
item = [9]
item = [10]
complete

Complete Publisher Implementation Pattern

public class RobustPublisher<T> implements Publisher<T> {
    private final List<T> items;
    private final Executor executor;
    
    public RobustPublisher(List<T> items, Executor executor) {
        this.items = items;
        this.executor = executor;
    }
    
    @Override
    public void subscribe(Subscriber<? super T> subscriber) {
        // Validate subscriber
        Objects.requireNonNull(subscriber);
        
        Subscription subscription = new Subscription() {
            private long requested = 0;
            private boolean cancelled = false;
            private int index = 0;
            
            @Override
            public synchronized void request(long n) {
                if (cancelled || n <= 0) {
                    if (n <= 0) subscriber.onError(new IllegalArgumentException());
                    return;
                }
                requested += n;
                drain();
            }
            
            @Override
            public synchronized void cancel() {
                cancelled = true;
            }
            
            private void drain() {
                executor.execute(() -> {
                    synchronized (this) {
                        while (requested > 0 && index < items.size() && !cancelled) {
                            subscriber.onNext(items.get(index++));
                            requested--;
                        }
                        if (index >= items.size() && !cancelled) {
                            subscriber.onComplete();
                        }
                    }
                });
            }
        };
        
        subscriber.onSubscribe(subscription);
    }
}

Reactive Streams Rules สำหรับ Publisher

1. Total onNext signals ≤ total requested
2. May signal fewer than requested, then terminate
3. Signals must be serial: onSubscribe, onNext*, (onError|onComplete)
4. onError on failure
5. onComplete on success
6. Terminal signal cancels subscription
7. No signals after terminal state
8. Cancelled subscription stops signals
9. subscribe() must call onSubscribe() first, throw NPE if subscriber null
10. subscribe() may be called multiple times
11. May support multiple subscribers (unicast/multicast)

Testing with JShell

jshell> var pub = new SimplePublisher(5)
jshell> pub.subscribe(new Flow.Subscriber<>() {
   ...>   public void onSubscribe(Flow.Subscription s) { s.request(2); }
   ...>   public void onNext(Integer i) { System.out.println(i); s.request(1); }
   ...>   public void onError(Throwable t) {}
   ...>   public void onComplete() { System.out.println("Done"); }
   ...> })
1
2
3
4
5
Done