Spring WebFlux and Server-Sent Events: A Match Made in Heaven

Introduction

In the world of web development, real-time communication between a server and its clients is crucial. Whether it’s a live sports update, a stock market feed, or a chat application, users expect to see updates instantly. This is where Server-Sent Events (SSE) come into play.

SSE is a standard that allows a server to push updates to a client whenever there’s new data available. It’s like a one-way ongoing conversation from the server to the client, keeping the client informed about the latest updates without the need for constant polling or manual refresh.

However, implementing SSE can be a challenge. Ensuring that the client can effectively listen to the server’s updates amidst all the other processes can be complex. This is the problem we’re going to tackle in this blog, diving deep into the world of SSE and exploring how to effectively implement it in your applications.

Problem Statement

Imagine you’re engrossed in a thrilling football match. Every goal, every card, every pivotal moment matters. Now, imagine you’re tasked with building a notification system to keep users updated about these crucial game events in real-time. This is the challenge we’re facing.

The traditional approach would be to have the client send a request to the server every time it needs an update. However, this method is far from efficient. It results in numerous network calls, putting unnecessary strain on both the server and the client.

An alternative approach is to use WebSockets, which provide a bidirectional communication channel between the client and the server. While this solves the problem of constant polling, it might be overkill for our use case. We only need updates from the server to the client, not the other way around.

This brings us to Server-Sent Events (SSE), a perfect fit for our scenario. SSE allows the server to push updates to the client whenever there’s new data, without the need for the client to make repeated requests. It’s a more efficient and elegant solution for our football match notification system.

In the following sections of this blog, we’ll delve into the implementation of this solution using Spring WebFlux, a powerful tool that makes working with SSE a breeze.

Sink Object: Unraveling the Core of Server-Sent Events

We’ll explore how the Sink object works, its role in the SSE ecosystem, and how it contributes to efficient, real-time data transmission. By understanding the Sink object, we’ll be one step closer to mastering SSE with Spring WebFlux.

Sink is a powerful Object that acts as both a publisher and a subscriber. This dual role allows it to manage the flow of data efficiently and flexibly.

As a publisher, the Sink emits data to its subscribers. This is typically done through the sink.next() method, which pushes the next piece of data to all subscribers. In our football match scenario, this could be a new goal, a card, or any other game event.

On the other hand, as a subscriber, the Sink can subscribe to another publisher and receive data from it. This is particularly useful when you want to transform or process the data before republishing it. For instance, you might want to filter out certain types of game events or format the data in a specific way before sending it to the client.

By acting as both a publisher and a subscriber, the Sink object allows for a high degree of control over the data flow. This makes it an essential component in implementing Server-Sent Events (SSE) with Spring WebFlux.

In the following section, we’ll explore how to use the Sink object effectively to implement SSE in our football match notification system. Stay tuned!

Configure Sink Object

Start a New Spring Boot Project: Open your favorite IDE (like IntelliJ IDEA or Eclipse) and start a new Spring Boot project. You can also use Spring Initializr to bootstrap your project.

Choose the Required Dependencies: While creating the project, you’ll be asked to select dependencies. For a project using Server-Sent Events (SSE) with Spring WebFlux, you’ll need the following dependencies:

• Spring WebFlux: This is the reactive web framework for Spring, which supports SSE.
• Lombok (optional): This is a Java library that helps to reduce boilerplate code.

Generate the Project: Once you’ve selected the required dependencies, generate the project. This will give you a basic project structure with a pom.xml (if you’re using Maven) or build.gradle (if you’re using Gradle) file containing the dependencies you’ve chosen.

Import the Project: Import the project into your IDE and wait for it to resolve the dependencies.

Create Your Application: Now you’re ready to start building your application.

Get Nour-eddine lachgar’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

now let’s create our Game class :

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

@Data
@AllArgsConstructor
@NoArgsConstructor
public class Game {
    private int scoreTeamA;
    private int scoreTeamB;
    private String message;
}

the Sink configuration class looks like this

import com.noureddine.notification.notificationsystemprojectreactor.entity.Game;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;

@Configuration
public class SinkConfig {
    @Bean
    public Sinks.Many<Game> sink() {
        return Sinks.many().replay().limit(1);
    }

    @Bean
    public Flux<Game> gameBroadcast(Sinks.Many<Game> sink) {
        return sink.asFlux();
    }
}

This class is annotated with @Configuration, indicating that it’s a source of bean definitions for the Spring IoC container.

The sink() method is annotated with @Bean, meaning it’s a bean producer. It returns a Sinks.Many<Game> object. Sinks.Many is a specialized Sink that can have multiple subscribers. The replay().limit(1) part means that it will replay the last emitted element (due to limit(1)) to any late subscribers.

The gameBroadcast() method is another bean producer that returns a Flux<Game> object. It takes the Sinks.Many<Game> object produced by sink() as an argument and converts it into a Flux using the asFlux() method. A Flux is a publisher that can emit 0 to N elements.

So, in essence, this configuration class sets up a Sink that can receive Game objects and a Flux that allows these Game objects to be consumed. This setup is particularly useful for scenarios where you want to broadcast updates (like game events) to multiple subscribers.

now let’s create our controller class

import com.noureddine.notification.notificationsystemprojectreactor.entity.Game;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;

@RestController
@RequestMapping("/api/v1/games")
public class GameController {

    // Controller endpoints
}

Let’s create the endpoint that will be responsible for updating the game information.

we should inject our Sink Bean to our controller

@Autowired
private Sinks.Many<Game> gameSink;

the endpoint will push the game updates to the Sink Object

@PostMapping("/update")
public Mono<Void> updateGame(@RequestBody Game game) {
    gameSink.tryEmitNext(game);
    return Mono.empty();
}

The updateGame(@RequestBody Game game) method is a POST endpoint that updates the game state. It takes a Game object as a request body, emits it to the gameSink, and then returns an empty Mono<Void> object. The @PostMapping("/update") annotation indicates that this method should be invoked for POST requests to the /update endpoint.

now we should create an endpoint to subscribe to that Sink object. to do that we can inject our Flux object

@Autowired
private Flux<Game> gameFlux;

the endpoint looks like this

@GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<Game> getGameUpdates() {
    return this.gameFlux;
}

The getGameUpdates() method is a GET endpoint that returns a Flux<Game> object representing a stream of game updates. The @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE) annotation indicates that this method should be invoked for GET requests to the /stream endpoint and that it produces Server-Sent Events.

Application Demo:

Demo

By now, you should have a solid understanding of how to use SSE with Spring WebFlux to build a real-time notification system. But the learning doesn’t stop here. I encourage you to experiment with the concepts we’ve discussed and see how you can apply them to your own projects.

The complete code for the project we’ve discussed is available in this GitHub repository https://github.com/noureddine409/notification-sink . Feel free to clone it, play around with it, and use it as a starting point for your own applications.

Stackademic

Thank you for reading until the end. Before you go:

• Please consider clapping and following the writer! 👏
• Follow us on Twitter(X), LinkedIn, and YouTube.
• Visit Stackademic.com to find out more about how we are democratizing free programming education around the world.