---
title: Fullstack Reactive Server Sent Events
url: https://blog.octoperf.com/fullstack-reactive-server-sent-events/
date: '2019-10-09'
authors:
- gpereira
categories:
- Development
tags:
- Angular
- Spring
- Web Sockets
- Proxy
description: How to handle Server Sent Events in a Fullstack Reactive web application based on Spring WebFlux and Angular.
---

# Fullstack Reactive Server Sent Events

[Reactive programming](https://en.wikipedia.org/wiki/Reactive_programming) is a programming paradigm aimed at maintaining overall coherence by propagating changes from a reactive source (modification of a variable, user input, etc.) to elements dependent on this source.

Reactive applications are more and more important today as they are:

- **Highly Available**: the system must respond quickly no matter what,
- **Resilient**: the system must always remain available, even in case of error or even if it is overloaded,
- **Message Oriented**: the system uses asynchronous messages.

Which meets the requirements of today's users.


## Fullstack Reactive with Spring Webflux and Angular

Technically speaking, meeting such requirements implies using a set of frameworks to ensure a smooth user experience.
**This stack of reactive-compliant web frameworks goes from the backend with [Spring WebFlux](https://docs.spring.io/spring/docs/current/spring-framework-reference/web-reactive.html) to the frontend with [Angular](https://angular.io/).**  

Built on the [Reactive Streams API](https://www.reactive-streams.org/) (a standard for asynchronous stream processing with non-blocking back pressure), Spring WebFlux allows to run non-blocking servers such as Netty. 
It uses the reactive library [Reactor](https://projectreactor.io/) which provides the Mono and Flux API types to work on single data (Mono) and sequences (Flux). 
 
Angular extensively uses the [RxJs library](https://angular.io/guide/rx-library) for user events propagation through UI components and services, but also for connections with the backend:
The [HTTP service returns Observables](https://angular.io/guide/http#observables-and-operators).

![SSE Propagation](https://blog.octoperf.com/img/blog/server-sent-events-reactive/sse-propagation.png "SSE Propagation")

This allows us to use the same mix of asynchronous and functional programming in both our Java server code and in our TypeScript UI code, the APIs for Reactor and [RxJS](https://rxjs.dev/guide/overview) being quite similar.
More importantly, with a fully reactive stack, changes can be propagated seamlessly everywhere in our application.
When a user interacts with the UI, an HTTP request is sent to the server and the change is made on the backend.
**But how to propagate independent changes occurring on the server to the UI? By using Server Sent Events**.

## Server Sent Event Introduction

Server-Sent-Events (aka SSE), is an [HTTP standard](https://www.w3.org/TR/eventsource/) that allows a web application to receive events emitted by the server in a uni-directional stream.

Before using SSE, you should be aware of a few limitations:

- First and foremost, [Microsoft IE and Edge browser are not compatible](https://developer.mozilla.org/en-US/docs/Web/API/EventSource#Browser_compatibility): you will need to [use polyfills to make them work with SSE](#using-polyfills),
- The [EventSource API](https://developer.mozilla.org/en-US/docs/Web/API/EventSource#Properties) has no support for custom headers,
- Web browser have a [maximum number of connections per hostname](https://www.geeksforgeeks.org/computer-networks/what-are-max-parallel-http-connections-in-a-browser/) (usually 6, **for the whole browser**), and since SSE keeps an open connection to the server, you should not open too many SSE streams at the same time, 
- Data transmitted in events is encoded in UTF-8, [unlike WebSocket which support binary data](https://stackoverflow.com/questions/5195452/websockets-vs-server-sent-events-eventsource).

## Code Example

### SSE and Spring Webflux

Please check this [Getting started with WebFlux](https://spring.io/guides/gs/reactive-rest-service/) guide if you're new to it.

Let's start by seeing how to handle Server-Sent-Events in Spring WebFlux to dispatch StorageWatcherEvents.
A `StorageWatcherEvent` is a simple bean that contains a `StorageNode` (the representation of a File in a tree) and an event: created, modified or deleted.
Directory watching is made using the [directory-watcher](https://github.com/gmethvin/directory-watcher) project and exposed by a Spring service called StorageWatcherService. 

#### text/event-stream

This first version of an SSE compliant Spring REST Controller is pretty simple.
We just have to return a `Flux<StorageWatcherEvent>`:

```java
import org.springframework.core.io.InputStreamResource;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.http.codec.multipart.FilePart;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

[...]

@RestController()
@RequestMapping("/files")
class StorageController {

  StorageWatcherService watcher;

  @GetMapping(value = "/watch")
  public Flux<StorageWatcherEvent> watch() {
    return this.watcher.watch();
  }
}
```

Spring will automatically handle all requests with the `Accept` header to `text/event-stream` as an SSE request an answer accordingly:

![text/event-stream](https://blog.octoperf.com/img/blog/server-sent-events-reactive/text-event-stream.png "text/event-stream")

> __Note:__ You can force it with the _produces_  annotation configuration : `@GetMapping(path = "/watch", produces = MediaType.TEXT_EVENT_STREAM_VALUE)`.

There is issue with this solution through: **if no event is generated by the `watcher.watch()` for a while, the connection may be closed if it goes through an HTTP proxy**.

The result is that you must set custom timeouts on your proxy: 
- [_timeout tunnel_ and _timeout server_ for HAProxy](https://stackoverflow.com/questions/21419859/how-can-i-configure-haproxy-to-work-with-server-sent-events),
- [_proxy_read_timeout_ for NGINX](https://stackoverflow.com/questions/21630509/server-sent-events-connection-timeout-on-node-js-via-nginx).

Also, [SSE Polyfill](#using-polyfills) client libraries request keep-alive events to make SSE work on Microsoft IE and Edge.

#### Flux of ServerSentEvents

Let's see **how to improve our server by making it send Keep-alive events**.
Keep-alive are [ServerSentEvent](https://docs.spring.io/spring/docs/current/javadoc-api/org/springframework/http/codec/ServerSentEvent.html) that do not contain `data` but only a `comment`.

Start by creating a utility service called SSEService that transforms a `Flux<T>` into a `Flux<ServerSentEvent<T>>`;

```java
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;

[...]

@Component
final class SSEKeepAliveService implements SSEService {

  long delay;

  @Autowired
  SSEKeepAliveService(@Value("${kraken.sse.keep-alive:#{environment.KRAKEN_SSE_KEEP_ALIVE_DELAY ?: 15}}") final Long delay) {
    this.delay = delay;
  }

  @Override
  public <T> Flux<ServerSentEvent<T>> wrap(Flux<T> flux) {
    return Flux.merge(flux.map(t -> ServerSentEvent.builder(t).build()), Flux.interval(Duration.ofSeconds(this.delay)).map(aLong -> ServerSentEvent.<T>builder().comment("keep alive").build()));
  }

``` 

It allows us to create ServerSentEvent with only a comment `ServerSentEvent.<T>builder().comment("keep alive").build()` and inject them in the Flux periodically.
 
Then we can simply use this service in our REST controller and return directly the `Flux<ServerSentEvent<StorageWatcherEvent>>`;

```java
import org.springframework.core.io.InputStreamResource;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.http.codec.multipart.FilePart;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;

[...]

@RestController()
@RequestMapping("/files")
class StorageController {

  StorageWatcherService watcher;
  SSEService sse;

  @GetMapping(value = "/watch")
  public Flux<ServerSentEvent<StorageWatcherEvent>> watch() {
    return this.sse.wrap(this.watcher.watch());
  }

}
```

The `/files/watch` endpoint now returns a `:keep-alive` comment every 15 second, along with the normal events that contain the JSON of a _StorageWatcherEvent_ object: 

![Keep Alive](https://blog.octoperf.com/img/blog/server-sent-events-reactive/keep-alive.png "Keep Alive")

To call this endpoint from Angular you need an EventSource, let's have a look at it!

### SSE and Angular

On the client side, we use the [EventSource](https://developer.mozilla.org/en-US/docs/Web/API/EventSource) to receive the stream events.
This object has 3 [EventHandler](https://developer.mozilla.org/en-US/docs/Web/Guide/Events/Event_handlers) methods:

- `EventSource.onerror` called when an error occurs,
- `EventSource.onmessage` called when a message event is received.
- `EventSource.onopen` called when an open event is received.

#### EventSource Observable Wrapper

To make it reactive and fit in an Angular application, it's better to wrap it in an RxJS Observable.

Here is the `event-source.service.ts` file:

```javascript
import {Injectable} from '@angular/core';
import {Observable} from 'rxjs';
import * as _ from 'lodash';

@Injectable({
  providedIn: 'root'
})
export class EventSourceService {

  newEventSource(path: string): EventSource {
    return new EventSource(path);
  }

  newObservable<R>(path: string, converter: (data: string) => R = _.identity): Observable<R> {
    return new Observable(observer => {
      const eventSource = this.newEventSource(path);
      eventSource.onmessage = event => {
        observer.next(converter(event.data));
      };
      eventSource.onerror = () => {
        if (eventSource.readyState !== eventSource.CONNECTING) {
          observer.error('An error occurred.');
        }
        eventSource.close();
        observer.complete();
      };
      return () => {
        eventSource.close();
      };
    });
  }
}
```

This service method `newObservable()` can then be subscribed like any other HTTP method: 

```javascript
this.eventSourceService.newObservable('/files/watch').subscribe((watcherEvent: StorageWatcherEvent) => console.log(watcherEvent));
```

#### Using PolyFills

**To make SSE work on both Microsoft IE and Edge browser we need to install a Polyfills client library** (and [send Keep-alive comments from the server](#flux-of-serversentevents)):
A good library for this is [Yaffle/EventSource](https://github.com/Yaffle/EventSource).

The installation is straightforward:

```
npm install event-source-polyfill --save
```

Then update the `polyfills.ts` file at the root of your application to add a reference to this library:

```javascript
import 'event-source-polyfill/src/eventsource.min.js';
```

Finally, use it in the `event-source.service.ts` instead of the native EventSource object:

```javascript
import {NativeEventSource, EventSourcePolyfill} from 'event-source-polyfill';

const EventSource = NativeEventSource || EventSourcePolyfill;

@Injectable({
  providedIn: 'root'
})
export class EventSourceService {

  newEventSource(path: string): EventSource {
    return new EventSource(path);
  }
  
[...]

}
```