วิธีนำ SubmissionPublisher class ไปใช้ใน Java 9

ตั้งแต่ Java 9 สามารถสร้าง Reactive Streams ได้ด้วยการนำเสนอ 4 core interfaces: Publisher Subscriber Subscription Processor และ 1 concrete class: SubmissionPublisher ที่ implement Publisher interface แต่ละ interface มีบทบาทต่างกัน ตามหลักการของ Reactive Streams ใช้เมธอด submit() ของ SubmissionPublisher เพื่อเผยแพร่ item ให้แต่ละ subscriber

ไวยากรณ์

public class SubmissionPublisher<T> extends Object implements Flow.Publisher<T>, AutoCloseable

ตัวอย่างการใช้งาน

import java.util.concurrent.Flow.Subscriber;
import java.util.concurrent.Flow.Subscription;
import java.util.concurrent.SubmissionPublisher;

class MySubscriber<T> implements Subscriber<T> {
    private Subscription subscription;
    private String name;
    
    public MySubscriber(String name) {
        this.name = name;
    }
    
    @Override
    public void onComplete() {
        System.out.println(name + ": onComplete");
    }
    
    @Override
    public void onError(Throwable t) {
        System.out.println(name + ": onError");
        t.printStackTrace();
    }
    
    @Override
    public void onNext(T msg) {
        System.out.println(name + ": " + msg.toString() + " received in onNext");
        subscription.request(1);
    }
    
    @Override
    public void onSubscribe(Subscription subscription) {
        System.out.println(name + ": onSubscribe");
        this.subscription = subscription;
        subscription.request(1);
    }
}

// Main class
public class FlowTest {
    public static void main(String args[]) {
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
        MySubscriber<String> subscriber = new MySubscriber<>("Mine");
        MySubscriber<String> subscriberYours = new MySubscriber<>("Yours");
        MySubscriber<String> subscriberHis = new MySubscriber<>("His");
        MySubscriber<String> subscriberHers = new MySubscriber<>("Her");
        
        publisher.subscribe(subscriber);
        publisher.subscribe(subscriberYours);
        publisher.subscribe(subscriberHis);
        publisher.subscribe(subscriberHers);
        
        publisher.submit("One");
        publisher.submit("Two");
        publisher.submit("Three");
        publisher.submit("Four");
        publisher.submit("Five");
        
        try { Thread.sleep(1000); } 
        catch(InterruptedException e) { e.printStackTrace(); }
        
        publisher.close();
    }
}

ผลลัพธ์

Yours: onSubscribe
His: onSubscribe
Mine: onSubscribe
His: One received in onNext
Yours: One received in onNext
Mine: One received in onNext
Yours: Two received in onNext
His: Two received in onNext
Yours: Three received in onNext
Mine: Two received in onNext
Yours: Four received in onNext
His: Three received in onNext
Yours: Five received in onNext
Mine: Three received in onNext
Her: onSubscribe
His: Four received in onNext
Her: One received in onNext
Mine: Four received in onNext
Her: Two received in onNext
His: Five received in onNext
Her: Three received in onNext
Mine: Five received in onNext
Her: Four received in onNext
Her: Five received in onNext
Yours: onComplete
His: onComplete
Mine: onComplete
Her: onComplete

คำอธิบาย

• SubmissionPublisher เป็น Publisher ที่ thread-safe
• submit() เป็น non-blocking ส่ง item ให้ subscribers ทั้งหมด
• Subscriber ใช้ request(1) เพื่อขอ item ถัดไป (backpressure)
• close() ปิด publisher และส่ง onComplete ให้ subscribers ทั้งหมด
• ลำดับ onNext อาจสลับกันได้ เพราะเป็น concurrent