reactive-programming - リアクティブ ストリーム SubmissionPublisher からアイテムを受け取る

Question

私は Java 9 の新機能のいくつかを試していました。そこで、特定のレートで数値を発行するパブリッシャーを用意するためのテストをまとめました。また、これらのパブリケーションをリッスンしてコンソールに出力するためのサブスクライバーも実装しました。

onNext()メソッドは何も出力せず、getLastItem()0 しか返さないため、この Api の使用方法を完全には理解していない可能性があります。

機能しているように見える唯一の部分は、変数onSubscribe()を正しく初期化する部分です。lastItem

@Test
public void testReactiveStreams(){
    //Create Publisher
    SubmissionPublisher<Integer> publisher = new SubmissionPublisher<>();

    //Register Subscriber
    TestIntegerSubscriber subscriber = new TestIntegerSubscriber();
    publisher.subscribe(subscriber);

    assertTrue(publisher.hasSubscribers());

    //Publish items
    System.out.println("Publishing Items...");

    List.of(1,2,3,4,5).stream().forEach(i -> {
        publisher.submit(i);
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            // blah
        }
    });
    assertEquals(5, subscriber.getLastItem());

    publisher.close();
}


private class TestIntegerSubscriber implements Flow.Subscriber<Integer> {

    private int lastItem;
    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        System.out.println("Subscribed");
        lastItem = 0;
    }

    @Override
    public void onNext(Integer item) {
        System.out.println("Received : "+item);
        lastItem += 1; // expect increment by 1
        assertTrue(lastItem == item);
    }

    @Override
    public void onError(Throwable throwable) {
        // nothing for the moment
    }

    @Override
    public void onComplete() {
        System.out.println("Completed");
    }

    public int getLastItem(){
        return lastItem;
    }
}

誰かが私のテストで何が間違っているのか教えてもらえますか? テストがこれらの数字を出力し、最後の項目として 5 を返すことを期待します。

Observables と Subjects は Angular2 でのみ使用していると言わざるを得ませんが、理解しやすいようです。