Real-Time Streaming with Spring WebFlux:

Live User Feeds via SSE

Generate Users on a Scheduler and Visualize Them in the Browser

Keywords: Spring WebFlux, Server-Sent Events, Reactive Streams, Flux.interval, Live Dashboard

Concept: Streams, Producers and Live Views

Reactive streams shine when data is produced continuously rather than requested once and forgotten. In this follow-up, a background scheduler keeps generating new User documents, saving them into MongoDB and pushing them over Server-Sent Events (SSE) to any connected browser. The browser subscribes to the /users/live-stream endpoint and instantly displays new users as they are created.

This pattern mirrors real-world use cases like live notifications, monitoring dashboards, or activity feeds. Instead of clients polling every few seconds, the server pushes new events as soon as they occur, using minimal resources and maintaining a single HTTP connection per client.

Step 1: Recap of Core Pieces

Before adding streaming, the basic setup remains the same:

• User document annotated with @Document("users")
• ReactiveUserRepository extends ReactiveMongoRepository<User, String>
• Standard CRUD endpoints on /users for listing and creating users

These parts stay unchanged and are used as the basis for the new streaming behavior.

Step 2: Creating the Producer — Scheduled User Generator

First, introduce a service that periodically creates random users and saves them through the reactive MongoDB repository.

UserStreamService.java

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import java.time.Duration;
import java.util.Random;

@Service
public class UserStreamService {

    private final ReactiveUserRepository userRepository;
    private final Random random = new Random();

    public UserStreamService(ReactiveUserRepository userRepository) {
        this.userRepository = userRepository;
    }

    // Infinite stream producing new users every 3-7 seconds
    public Flux<User> generateUsers() {
        return Flux.interval(Duration.ofSeconds(3 + random.nextInt(5)))  // Random 3-8s
                .map(i -> createRandomUser())
                .flatMap(userRepository::save)
                .doOnNext(user -> System.out.println("New user: " + user.getName()));
    }

    private User createRandomUser() {
        String[] names = {"Alice", "Bob", "Charlie", "Diana", "Eve", "Frank", "Mehmood", "Ghaffar"};
        String[] domains = {"gmail.com", "yahoo.com", "outlook.com"};
        return new User(names[random.nextInt(names.length)],
                random.nextInt(10000) + "@" + domains[random.nextInt(domains.length)]);
    }
}

This service uses Flux.interval to emit ticks, maps each tick to a new random User, saves it via userRepository.save, and returns a Flux<User> that never completes. As long as the application runs, new users get inserted and streamed.

Step 3: Exposing the Live Stream Endpoint via SSE

Next, update the controller to expose a dedicated streaming endpoint that uses UserStreamService and returns ServerSentEvent<User> values.

UserController.java (streaming part)

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@RestController
@RequestMapping("/users")
public class UserController {

    @Autowired
    private UserStreamService userStreamService;

    private final ReactiveUserRepository userRepository;

    public UserController(ReactiveUserRepository userRepository) {
        this.userRepository = userRepository;
    }    

    @GetMapping
    public Flux<User> getAllUsers() {
        return userRepository.findAll();
    }

    @GetMapping("/{id}")
    public Mono<User> getUserById(@PathVariable String id) {
        return userRepository.findById(id);
    }

    @PostMapping
    public Mono<User> createUser(@RequestBody User user) {
        return userRepository.save(user);
    }

    // NEW: Live streaming endpoint using the autowired service
    @GetMapping(value = "/live-stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<User>> liveUserStream() {
        return userStreamService.generateUsers()
                .map(user -> ServerSentEvent.<User>builder(user)
                        .event("user-created")
                        .id(user.getId())
                        .build());
    }

    @PostMapping
    public Mono<User> createUser(@RequestBody User user) {
        return userRepository.save(user);
    }

    @DeleteMapping("/clear")
    public Mono<Void> clearAllUsers() {
        return userRepository.deleteAll();
    }
}

Key points:

• produces = MediaType.TEXT_EVENT_STREAM_VALUE enables SSE.
• Every time generateUsers() emits a new User, it is wrapped in a ServerSentEvent with event name user-created.
• The Flux is infinite, so the HTTP connection stays open and keeps sending events.

Step 4: Visualizing the Stream in the Browser

To make the stream visible, create a tiny HTML dashboard that subscribes to the SSE endpoint and renders each incoming user to the page.

index.html (served from src/main/resources/static/index.html)

<!DOCTYPE html>
<html>
<head>
    <title>Live User Stream</title>
    <style>
        #users { list-style: none; padding: 0; }
        #users li {
            background: #f0f8ff; margin: 5px; padding: 10px;
            border-radius: 5px; border-left: 4px solid #007bff;
        }
    </style>
</head>
<body>
<h1>🟢 Live User Registrations</h1>
<ul id="users"></ul>

<script>
    const eventSource = new EventSource('/users/live-stream');
    const userList = document.getElementById('users');

    eventSource.addEventListener('user-created', function(event) {
        const user = JSON.parse(event.data);
        const li = document.createElement('li');
        li.innerHTML = `👤 ${user.name} <small>(${user.email})</small>`;
        userList.insertBefore(li, userList.firstChild);  // Newest first
    });

    eventSource.onerror = function() {
        console.log('SSE connection lost, reconnecting...');
    };
</script>
</body>
</html>

Open http://localhost:8080 in a browser:

• The page connects to /users/live-stream.
• Every few seconds a new User is generated by UserStreamService.
• Each new event triggers a DOM update, showing the latest users at the top of the list.

Step 5: Testing the Streaming Behavior

To see the streaming in action:

1. Ensure MongoDB is running and the app is started.
2. Open http://localhost:8080 in a browser.
3. Watch new user entries appear every 3–7 seconds automatically.
4. Optionally, inspect the stream at the network level:

curl -N -H "Accept: text/event-stream" http://localhost:8080/users/live-stream

You should see lines like:

event: user-created data: {"id":"...","name":"Alice","email":"alice1234@gmail.com"} 
event: user-created data: {"id":"...","name":"Frank","email":"frank9876@yahoo.com"}

This confirms that data is produced continuously on the server side and pushed reactively to all connected clients.

Conclusion

This article showed how to move from a simple reactive CRUD API to a truly streaming system:

• A producer (UserStreamService) emits users on a schedule using Flux.interval.
• A streaming controller endpoint (/users/live-stream) wraps each new user into a ServerSentEvent.
• A lightweight HTML page listens with EventSource and visualizes the data in real time.

The same pattern can power live monitoring dashboards, activity feeds, or notification systems, all built on top of Spring WebFlux and reactive streams.