DEV Community

loizenai
loizenai

Posted on

Java 9 Flow API example – Publisher and Subscriber

https://grokonez.com/java/java-9/java-9-flow-api-example-publisher-and-subscriber

Java 9 Flow API example – Publisher and Subscriber

In previous post, we have general knowledge about Reactive Streams and Java 9 Flow API Components and Behaviour. In this tutorial, we're gonna look at an example that implements Publisher and Subscriber for reactive programming.

Related Articles:

I. Technologies

  • Java 9
  • Eclipse with Java 9 Support for Oxygen (4.7)

    II. Project Overview

    We will create a Publisher that is subscribed by two Subscribers.
  • Publisher maintains a list of Subscriptions, each Subscription is correlative to each Subscriber above.
  • Publisher uses one Subscription to push items to correlative Subscriber by Subscriber::onNext() method.
  • Subscriber uses Subscription to request items from Publisher by Subscription::request() method.
  • Publisher defines an Executor for multi-threading. Then request() and onNext() method work asynchronously, producing data to each Subscriber by Subscription is also asynchronous.
  • After receiving all items successfully, Subscriber can request new data or cancel Subscription (random).

    III. Practice

    To understand how Publisher, Subscriber and Subscription behave and way to implementing them, please visit: Java 9 Flow API – Reactive Streams

    1. Create implementation of Publisher

    
    package com.javasampleapproach.java9flow.pubsub;

import static java.lang.Thread.currentThread;
import static java.util.concurrent.Executors.newSingleThreadExecutor;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Flow.Publisher;
import java.util.concurrent.Flow.Subscriber;
import java.util.concurrent.Flow.Subscription;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

public class MyPublisher implements Publisher {

private static final String LOG_MESSAGE_FORMAT = "Publisher >> [%s] %s%n";

final ExecutorService executor = Executors.newFixedThreadPool(4);
private List<MySubscription> subscriptions = Collections.synchronizedList(new ArrayList<MySubscription>());

private final CompletableFuture<Void> terminated = new CompletableFuture<>();

@Override
public void subscribe(Subscriber<? super Integer> subscriber) {
    MySubscription subscription = new MySubscription(subscriber, executor);

    subscriptions.add(subscription);

    subscriber.onSubscribe(subscription);
}

public void waitUntilTerminated() throws InterruptedException {
    try {
        terminated.get();
    } catch (ExecutionException e) {
        System.out.println(e);
    }
}

private class MySubscription implements Subscription {

    private final ExecutorService executor;

    private Subscriber<? super Integer> subscriber;
    private final AtomicInteger value;
    private AtomicBoolean isCanceled;

    public MySubscription(Subscriber<? super Integer> subscriber, ExecutorService executor) {
        this.subscriber = subscriber;
        this.executor = executor;

        value = new AtomicInteger();
        isCanceled = new AtomicBoolean(false);
    }

    @Override
    public void request(long n) {
        if (isCanceled.get())
            return;

        if (n < 0)
            executor.execute(() -> subscriber.onError(new IllegalArgumentException()));
        else
            publishItems(n);
    }

    @Override
    public void cancel() {
        isCanceled.set(true);

        synchronized (subscriptions) {
            subscriptions.remove(this);
            if (subscriptions.size() == 0)
                shutdown();
        }
    }

    private void publishItems(long n) {
        for (int i = 0; i < n; i++) {

            executor.execute(() -> {
                int v = value.incrementAndGet();
                log("publish item: [" + v + "] ...");
                subscriber.onNext(v);
            });
        }
    }

    private void shutdown() {
        log("Shut down executor...");
        executor.shutdown();
        newSingleThreadExecutor().submit(() -> {

            log("Shutdown complete.");
            terminated.complete(null);
        });
    }

}

private void log(String message, Object... args) {
    String fullMessage = String.format(LOG_MESSAGE_FORMAT, currentThread().getName(), message);

    System.out.printf(fullMessage, args);
}

}

2. Create implementation of Subscriber

More at:

https://grokonez.com/java/java-9/java-9-flow-api-example-publisher-and-subscriber

Java 9 Flow API example – Publisher and Subscriber

Image of Timescale

🚀 pgai Vectorizer: SQLAlchemy and LiteLLM Make Vector Search Simple

We built pgai Vectorizer to simplify embedding management for AI applications—without needing a separate database or complex infrastructure. Since launch, developers have created over 3,000 vectorizers on Timescale Cloud, with many more self-hosted.

Read more

Top comments (0)

Sentry image

See why 4M developers consider Sentry, “not bad.”

Fixing code doesn’t have to be the worst part of your day. Learn how Sentry can help.

Learn more

👋 Kindness is contagious

Engage with a sea of insights in this enlightening article, highly esteemed within the encouraging DEV Community. Programmers of every skill level are invited to participate and enrich our shared knowledge.

A simple "thank you" can uplift someone's spirits. Express your appreciation in the comments section!

On DEV, sharing knowledge smooths our journey and strengthens our community bonds. Found this useful? A brief thank you to the author can mean a lot.

Okay