Subscriber

Description

A receiver of messages. The methods in this interface are invoked in strict sequential order for each Flow.Subscription.

Requirements

JDK 21

Code

package com.jreact;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;
import java.util.concurrent.TimeUnit;

public class MySubscriber implements java.util.concurrent.Flow.Subscriber<String> {

    private Flow.Subscription subscription;

    public void flowSubscriber() {
        System.out.printf("main (tid=%d)%n", Thread.currentThread().threadId());


        try (final var publisher = new SubmissionPublisher<String>()) {

            System.out.println("-- subscribe --");

            publisher.subscribe(new MySubscriber());

            TimeUnit.SECONDS.sleep(1);
            System.out.println("-- submit --");

            publisher.submit("abc");
            publisher.submit("123");
            publisher.submit("XYZ");

            TimeUnit.SECONDS.sleep(1);
            System.out.println("-- close --");
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        System.out.printf("  onSubscribe (tid=%d)%n",
                Thread.currentThread().threadId());

        this.subscription = subscription;
        subscription.request(1);
    }

    @Override
    public void onNext(String item) {
        System.out.printf("  onNext (tid=%d) : %s%n",
                Thread.currentThread().threadId(), item);

        subscription.request(1);
    }

    @Override
    public void onError(Throwable throwable) {
        System.out.printf("  onError (tid=%d) : %s%n",
                Thread.currentThread().threadId(), throwable);
    }

    @Override
    public void onComplete() {
        System.out.printf("  onComplete (tid=%d)%n",
                Thread.currentThread().threadId());
    }
}

Output

main (tid=1)
-- subscribe --
  onSubscribe (tid=29)
-- submit --
  onNext (tid=29) : abc
  onNext (tid=29) : 123
  onNext (tid=29) : XYZ
-- close --
  onComplete (tid=29)

Process finished with exit code 0

Running Example Code