Java 9 Flow API example – Publisher and Subscriber
Tags: java, java9, publisher, subscriber
Published: 2021-05-19 10:01:15

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

Java 9 Flow API example – Publisher and Subscriber

In previous <a href="https://grokonez.com/java/java-9-flow-api-reactive-streams">post</a>, we have general knowledge about Reactive Streams and Java 9 <strong>Flow API</strong> Components and Behaviour. In this tutorial, we're gonna look at an example that implements <strong>Publisher</strong> and <strong>Subscriber</strong> for reactive programming.

Related Articles:
- <a href="https://grokonez.com/java/java-9-flow-api-reactive-streams">Java 9 Flow API – Reactive Streams</a>
- <a href="https://grokonez.com/java/java-9-flow-api-example-processor">Java 9 Flow API example – Processor</a>
- <a href="https://grokonez.com/java/java-9-flow-submissionpublisher-concrete-publisher">Java 9 FLow SubmissionPublisher – A Concrete Publisher</a>

<!--more-->
<div id="tuttoc" class="jsa-toc"></div>

<div id="tutcontents">
<h2>I. Technologies</h2>
- Java 9
- Eclipse with Java 9 Support for Oxygen (4.7)
<h2>II. Project Overview</h2>
<img src="https://grokonez.com/wp-content/uploads/2017/04/java-9-flow-pubisher-subscriber-example.png" alt="" width="699" height="365" class="alignnone size-full wp-image-4580" />
We will create a <strong>Publisher</strong> that is subscribed by two <strong>Subscriber</strong>s.
- <strong>Publisher</strong> maintains a list of <strong>Subscription</strong>s, each <strong>Subscription</strong> is correlative to each <strong>Subscriber</strong> above.
- <strong>Publisher</strong> uses one <strong>Subscription</strong> to push items to correlative <strong>Subscriber</strong> by <code>Subscriber::onNext()</code> method.
- <strong>Subscriber</strong> uses <strong>Subscription</strong> to request items from <strong>Publisher</strong> by <code>Subscription::request()</code> method.
- <strong>Publisher</strong> defines an <strong>Executor</strong> for multi-threading. Then <code>request()</code> and <code>onNext()</code> method work asynchronously, producing data to each <strong>Subscriber</strong> by <strong>Subscription</strong> is also asynchronous.
- After receiving all items successfully, <strong>Subscriber</strong> can request new data or cancel <strong>Subscription</strong> (random).
<h2>III. Practice</h2>
To understand how Publisher, Subscriber and Subscription behave and way to implementing them, please visit: <a href="https://grokonez.com/java/java-9-flow-api-reactive-streams">Java 9 Flow API – Reactive Streams</a>
<h3>1. Create implementation of Publisher</h3>
<pre><code class="language-java">
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<Integer> {

	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);
	}
}</code></pre>
<h3>2. Create implementation of Subscriber</h3>

More at:

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

Java 9 Flow API example – Publisher and Subscriber