Java 9でFlow.Publisherインターフェースを実装する方法を解説

Publisherインターフェースは、無制限の数の順序付き要素を供給するプロバイダーであり、Subscriber（サブスクライバー）から受け取る要求（デマンド）に応じて要素を発行します。Publisher.subscribe(Subscriber)が呼び出されると、Subscriber側のメソッドには以下のような呼び出しシーケンスが発生する可能性があります。

つまり、まずonSubscribe()メソッドが呼び出され、その後、Subscriberから要求された数だけonNext()メソッドが続きます。そして、失敗が発生した場合はonError()メソッドが、すべての要素の供給が完了した場合にはonComplete()メソッドが呼び出されます。ただし、これはSubscriptionがキャンセルされていないことが前提です。

構文

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

実装例

以下のコードでは、1から指定された数までの整数を順番に発行するシンプルなSimplePublisherクラスを、Flow.Publisher<Integer>インターフェースを実装して作成しています。

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) {
        iterator.forEachRemaining(subscriber::onNext);
        subscriber.onComplete();
    }
}

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

コードのポイント

• IntStream.rangeClosed(1, count)を使用して、1からcountまでの整数ストリームを生成し、そのイテレーターを保持しています。
• subscribe()メソッド内で、イテレーターの残りの全要素に対してforEachRemaining()を使ってonNext()を呼び出し、すべての要素を発行します。
• すべての要素の発行が完了したら、onComplete()を呼び出して処理の終了を通知します。

実行結果

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

このように、Java 9で導入されたjava.util.concurrent.FlowクラスのPublisherインターフェースを実装することで、リアクティブストリームの基本的なパブリッシュ・サブスクライブモデルを簡単に構築できます。実際の本格的な実装では、バックプレッシャー（要求ベースのフロー制御）や非同期処理への対応も考慮する必要があります。