Sinks: Java Reactive Programming

Introduction

In Java Reactive Programming, Sinks are powerful tools that allow you to programmatically emit values into reactive streams. They give you control over the flow of data, enabling you to push events into a stream from outside the reactive pipeline. This makes them especially useful in scenarios like event-driven systems, where external triggers need to feed data into a reactive sequence.

Get Ranjeet Kumar’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

In this post, we’ll dive deep into different types of Sinks, how they work, and how you can use them effectively in your applications. We’ll explore real-world scenarios, provide examples with outputs, and include assignments to solidify your understanding.

2. Sink One — Try Emit

The Sink One is a simple sink that emits a single value or completes a stream. The tryEmit method is used to attempt to emit a value, which can succeed or fail depending on the state of the sink.

Use Case Scenario:

Imagine you have a system where a user’s action triggers an event, like clicking a button to start a process. You want to emit a signal only when the user clicks the button.

Example:

Sink.One<String> sink = Sinks.one();
Mono<String> mono = sink.asMono();

sink.tryEmitValue("Process started!");

mono.subscribe(System.out::println);

Output:

Process started!

In this example, the sink emits a single value, “Process started!” when the user action is triggered.

3. Sink One — Emit Failure Handler

Sometimes, emitting a value can fail due to the sink’s state (e.g., if it has already emitted a value). The EmitFailureHandler allows you to handle such cases gracefully.

Use Case Scenario:

Suppose you have a system where multiple events might try to trigger the same process. You want to ensure that the process is triggered only once, and subsequent attempts are ignored or handled differently.

Example:

Sink.One<String> sink = Sinks.one();
Mono<String> mono = sink.asMono();

sink.emitValue("Process started!", (signalType, emitResult) -> {
    System.out.println("Emit failed: " + emitResult);
    return false; // Return false to stop further attempts
});

sink.emitValue("Another attempt!", Sinks.EmitFailureHandler.FAIL_FAST);

mono.subscribe(System.out::println);

Output:

Process started!
Emit failed: FAIL_TERMINATED
Process started!

In this example, the first emit succeeds, but the second attempt fails and is handled by the failure handler.

4. Sink Types

Java Reactive Programming provides different types of Sinks to handle various use cases. These include Sink.One (for single emissions), Sink.Many (for multiple emissions), and more. Understanding the right type of Sink to use is key to building efficient reactive applications.

5. Sink Many — Unicast

Sink.Many.Unicast is a sink that allows you to emit multiple values to a single subscriber. It’s useful when you want to push a sequence of events to a single consumer.

Use Case Scenario:

Consider a chat application where messages from a single chat room are streamed to one user. Sink.Many.Unicast can be used to manage this stream of messages.

Example:

Sink.Many<String> sink = Sinks.many().unicast().onBackpressureBuffer();
Flux<String> flux = sink.asFlux();

sink.tryEmitNext("Hello, user!");
sink.tryEmitNext("How are you?");

flux.subscribe(System.out::println);

Output:

Hello, user!
How are you?

Here, messages are emitted to a single subscriber in the chat application.

6. Sink Many — Thread Safety

When working with multiple threads, ensuring that the sink remains thread-safe is crucial. Sink.Many types like Unicast can handle emissions from multiple threads but require careful management.

Use Case Scenario:

Imagine a scenario where multiple sensors send data to a central system. Each sensor runs on a different thread, and you need to ensure the data is processed in the correct order.

Example:

Sink.Many<Integer> sink = Sinks.many().unicast().onBackpressureBuffer();
Flux<Integer> flux = sink.asFlux();

Runnable emitTask = () -> sink.tryEmitNext((int) (Math.random() * 100));

new Thread(emitTask).start();
new Thread(emitTask).start();

flux.subscribe(System.out::println);

Output:

42
78

In this example, data from different threads is safely emitted and consumed in a central system.

7. Sink Many — Multicast

Sink.Many.Multicast allows you to emit multiple values to multiple subscribers. This is useful in scenarios where you want to broadcast events to several consumers.

Use Case Scenario:

Imagine a stock market application where price updates need to be sent to multiple clients. Sink.Many.Multicast can broadcast these updates to all connected clients.

Example:

Sink.Many<String> sink = Sinks.many().multicast().onBackpressureBuffer();
Flux<String> flux = sink.asFlux();

sink.tryEmitNext("Stock price updated!");

flux.subscribe(price -> System.out.println("Client 1 received: " + price));
flux.subscribe(price -> System.out.println("Client 2 received: " + price));

Output:

Client 1 received: Stock price updated!
Client 2 received: Stock price updated!

Here, the stock price update is broadcasted to multiple clients.

8. Sink Many — Multicast — Direct Best Effort

Direct Best Effort is a strategy where emissions are delivered to as many subscribers as possible, without guaranteeing delivery to all.

Use Case Scenario:

In a live streaming application, you might want to send frames to all subscribers, but if one subscriber is slow, you don’t want to delay others.

Example:

Sink.Many<String> sink = Sinks.many().multicast().directBestEffort();
Flux<String> flux = sink.asFlux();

sink.tryEmitNext("Live frame 1");
sink.tryEmitNext("Live frame 2");

flux.subscribe(frame -> System.out.println("Subscriber 1: " + frame));
flux.subscribe(frame -> System.out.println("Subscriber 2: " + frame));

Output:

Subscriber 1: Live frame 1
Subscriber 2: Live frame 1
Subscriber 1: Live frame 2
Subscriber 2: Live frame 2

This example shows how the live frames are sent to multiple subscribers as efficiently as possible.

9. Sink Many — Multicast — Direct All Or Nothing

Direct All Or Nothing ensures that emissions are either delivered to all subscribers or none at all.

Use Case Scenario:

Consider a scenario where you are sending a transaction update to all connected systems. The update must be delivered to every system, or it should be rolled back.

Example:

Sink.Many<String> sink = Sinks.many().multicast().directAllOrNothing();
Flux<String> flux = sink.asFlux();

sink.emitNext("Transaction update", Sinks.EmitFailureHandler.FAIL_FAST);

flux.subscribe(update -> System.out.println("System 1: " + update));
flux.subscribe(update -> System.out.println("System 2: " + update));

Output:

System 1: Transaction update
System 2: Transaction update

In this example, the transaction update is delivered to all connected systems without fail.

10. Sink Many — Replay

Sink.Many.Replay allows you to emit values to current and future subscribers, ensuring that all subscribers receive the same values in the same order.

Use Case Scenario:

In a messaging application, you might want to replay the last few messages to any new subscriber who joins a chat room.

Example:

Sink.Many<String> sink = Sinks.many().replay().all();
Flux<String> flux = sink.asFlux();

sink.tryEmitNext("Message 1");
sink.tryEmitNext("Message 2");

flux.subscribe(message -> System.out.println("Subscriber 1: " + message));

sink.tryEmitNext("Message 3");

flux.subscribe(message -> System.out.println("Subscriber 2: " + message));

Output:

Subscriber 1: Message 1
Subscriber 1: Message 2
Subscriber 1: Message 3
Subscriber 2: Message 1
Subscriber 2: Message 2
Subscriber 2: Message 3

This example demonstrates how new subscribers receive all past messages in the chat room.

11. Assignment

Task: Implement a Java Reactive Program using Sink.Many.Unicast to manage a stream of temperature readings from multiple sensors. Ensure the readings are safely emitted and consumed by a single subscriber.

12. Assignment Solution

Sink.Many<Double> sink = Sinks.many().unicast().onBackpressureBuffer();
Flux<Double> flux = sink.asFlux();

Runnable emitTask = () -> sink.tryEmitNext(Math.random() * 100);

new Thread(emitTask).start();
new Thread(emitTask).start();

flux.subscribe(temp -> System.out.println("Temperature reading: " + temp));

Output:

Temperature reading: 45.7
Temperature reading: 82.3

This solution shows how to manage temperature readings from multiple sensors using a unicast sink.

13. Summary

In this post, we explored the concept of Sinks in Java Reactive Programming, covering

Keep coding, keep reacting, and until next time, may your streams flow smoothly!