วิธีนำ Subscriber interface ไปใช้ใน Java 9

Java 9 รองรับการสร้าง Reactive Streams โดยแนะนำอินเทอร์เฟซ: Publisher, Subscriber, Subscription, และคลาส SubmissionPublisher ที่ implements Publisher แต่ละอินเทอร์เฟซมีบทบาทต่างกันตามหลัก Reactive Streams

เราสามารถใช้ Subscriber interface เพื่อสมัครรับข้อมูลที่เผยแพร่โดย publisher ต้อง implement Subscriber interface และให้การ implement เมธอด abstract

เมธอดของ Flow.Subscriber interface:

• onComplete(): เรียกเมื่อ Publisher ทำหน้าที่เสร็จสิ้น
• onError(): เรียกเมื่อ Publisher เกิดข้อผิดพลาด และแจ้งให้ Subscriber ทราบ
• onNext(): เรียกทุกครั้งที่ Publisher มีข้อมูลใหม่จะแจ้งให้ Subscriber ทั้งหมดทราบ
• onSubscribe(): เรียกเมื่อ Publisher เพิ่ม Subscriber

ตัวอย่าง

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;
import java.util.stream.IntStream;

public class SubscriberImplTest {
    public static class Subscriber implements Flow.Subscriber<Integer> {
        private Flow.Subscription subscription;
        private boolean isDone;

        @Override
        public void onSubscribe(Flow.Subscription subscription) {
            System.out.println("Subscribed");
            this.subscription = subscription;
            this.subscription.request(1);
        }

        @Override
        public void onNext(Integer item) {
            System.out.println("Processing " + item);
            this.subscription.request(1);
        }

        @Override
        public void onError(Throwable throwable) {
            throwable.printStackTrace();
        }

        @Override
        public void onComplete() {
            System.out.println("Processing done");
            isDone = true;
        }
    }

    public static void main(String args[]) throws InterruptedException {
        SubmissionPublisher<Integer> publisher = new SubmissionPublisher<>();
        Subscriber subscriber = new Subscriber();
        publisher.subscribe(subscriber);
        IntStream intData = IntStream.rangeClosed(1, 10);
        intData.forEach(publisher::submit);
        publisher.close();
        while(!subscriber.isDone) {
            Thread.sleep(10);
        }
        System.out.println("Done");
    }
}

ผลลัพธ์

Subscribed Processing 1 Processing 2 Processing 3 Processing 4 Processing 5 Processing 6 Processing 7 Processing 8 Processing 9 Processing 10 Processing done Done