Package rife.engine

Class SseBroadcaster

java.lang.Object

rife.engine.SseBroadcaster

All Implemented Interfaces:

AutoCloseable

public class SseBroadcaster
extends Object
implements AutoCloseable

Manages a collection of detached server-sent events connections and broadcasts events to all of them.

A broadcaster is typically stored in a field of your Site and handed to Context.sse(SseBroadcaster) inside the element of an SSE route. The element returns immediately while the connection stays open, which makes it possible to push events afterwards from anywhere else in the application: another element, a scheduled task, a workflow, or any other thread.

public class MySite extends Site {
     final SseBroadcaster ticker = new SseBroadcaster();
     Route events = get("/events", c -> c.sse(ticker));
 }

 // from anywhere else:
 site.ticker.send(new ServerSentEvent().name("tick").data("42"));

Connections whose clients have disconnected will automatically be removed when the servlet container reports that their request has ended, and otherwise when sending to them fails. Sending a periodic heartbeat comment will both prevent intermediaries from closing idle connections and reap the ones that have disconnected.

When event history is enabled, clients that reconnect will automatically receive the events that they missed.

Since:

1.10

See Also:

• SseConnection
• ServerSentEvent
• Context.sse(SseBroadcaster)

Constructor Summary

Constructors

Constructor	Description
SseBroadcaster()

Method Summary

Modifier and Type	Method	Description
void	close()	Closes all the current connections of this broadcaster, clears the event history and stops the heartbeat.
int	close(SseConnectionFilter filter)	Closes the connections of this broadcaster that match a filter.
int	comment(String comment)	Sends a comment line to all the open connections, typically used as a keep-alive heartbeat that also reaps the stale connections.
int	connectionCount()	Retrieves the number of connections that are currently registered with this broadcaster.
SseBroadcaster	heartbeat(Duration interval)	Automatically sends a keep-alive comment to all the open connections at a fixed interval.
SseBroadcaster	history(int capacity)	Enables event history with the provided capacity, so that clients that reconnect automatically receive the events that they missed.
SseHistoryStats	historyStats()	Retrieves a snapshot of the state of the event history, which is intended to help you tune the history capacity.
String	lastEventId()	Retrieves the ID of the most recently sent event when history is enabled.
int	send(CharSequence data)	Sends an event with the provided text as its data to all the open connections of this broadcaster.
int	send(ServerSentEvent event)	Sends an event to all the open connections of this broadcaster.
int	send(ServerSentEvent event, SseConnectionFilter filter)	Sends an event to the open connections of this broadcaster that match a filter.
int	send(Template template)	Sends an event with the content of the provided template as its data to all the open connections of this broadcaster.
int	send(Template template, String blockId)	Sends an event with the content of a single template block as its data to all the open connections of this broadcaster.
SseBroadcaster	stopHeartbeat()	Stops the heartbeat.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Details

SseBroadcaster

public SseBroadcaster()

Method Details

history

public SseBroadcaster history(int capacity)

Enables event history with the provided capacity, so that clients that reconnect automatically receive the events that they missed.

When history is enabled, the broadcaster assigns the SSE ID of every event that carries a name, data or a template. These IDs are composed of an epoch that identifies the broadcaster instance and of a monotonically increasing sequence number, so that IDs from before an application restart can't be mistaken for positions in the sequence of the new instance. Since application-assigned event IDs and broadcaster history are alternative reconnection strategies that can't be combined, sending an event that carries its own ID will throw an IllegalArgumentException. The last capacity events are buffered, with the template data captured at the moment of broadcasting.

When a browser reconnects, it transmits the ID of the last event that it received, and the buffered events after that ID will be replayed to it before it rejoins the live stream. Events that were targeted with a filter are only replayed to the connections that match the filter: the filters are executed again at replay time, and an exception that is thrown by a filter propagates to the reconnecting request. The lastEventId request parameter is honored in the same way as the reconnection header, which allows pages to receive the events that occurred between their rendering and the establishment of the stream.

When a client reconnects after having missed more events than the history contains, nothing will be replayed and the reconnection is reported by historyStats() and logged to the rife.engine logger. Reconnections that present an ID from before an application restart are reported in the same way, since the history doesn't bridge restarts. You should size the capacity so that it covers the events that can occur during the longest disconnection that has to be bridged seamlessly, since applications typically handle longer gaps by rendering the full state on the page. Note that the reconnection ID is provided by the client, which means that each reconnection can request up to capacity events to be replayed, so you should keep the capacity proportionate for streams that are accessible without authentication.

Heartbeat comments aren't buffered and don't receive IDs. Only the events that are sent through the broadcaster become part of the history, since events that are sent directly to an individual connection bypass it. The history is kept in memory and doesn't survive an application restart.

Events are delivered to the connections outside of the internal history lock. When multiple threads broadcast concurrently, the order in which their events reach an individual connection isn't guaranteed, even though the history preserves the order of the IDs, and such events can be delivered again after a reconnection. Events that are broadcast sequentially from a single thread will always arrive in order. A connection whose client stops reading without disconnecting can block a sending thread while it's being written to, and since that write doesn't fail, the heartbeat can't reap such a connection: it's only reaped when the connection timeout of the servlet container fails the write. You should bound that timeout in the container configuration, which is done for the embedded servers through Server.connectionIdleTimeout(long) or TomcatServer.connectionTimeout(int), and keep it longer than the heartbeat interval.

Parameters:

capacity - the number of events to buffer

Returns:

this broadcaster instance

Since:

1.10

See Also:

• historyStats()
• lastEventId()

send

public int send(ServerSentEvent event)

Sends an event to all the open connections of this broadcaster.

Connections that fail to receive the event will be closed and removed.

Parameters:

event - the event to broadcast

Returns:

the number of connections that received the event; an event without any fields set isn't sent and returns 0

Throws:

IllegalArgumentException - when the event is null

Since:

1.10

See Also:

• send(ServerSentEvent, SseConnectionFilter)

send

public int send(ServerSentEvent event,
 SseConnectionFilter filter)

Sends an event to the open connections of this broadcaster that match a filter.

The filter receives each connection and can inspect its context to decide whether the event should be sent to it, for instance based on the authenticated identity or on the request parameters:

broadcaster.send(event, connection -> {
     var identity = Identified.getIdentity(connection.context());
     return identity != null && identity.getAttributes().isInRole("admin");
 });

The context of a connection is the request that established it, which means that the authentication state and the parameters that a filter sees are the ones from the moment the client connected. Connections whose context is no longer appropriate, for instance after a logout, can be terminated with close(SseConnectionFilter).

Connections that fail to receive the event will be closed and removed, while the connections that are filtered out are left untouched.

Exceptions that are thrown by the filter propagate to the caller, and the connections that weren't evaluated yet won't receive the event.

Parameters:

event - the event to broadcast

filter - the filter that determines which connections receive the event

Returns:

the number of connections that received the event; an event without any fields set isn't sent and returns 0

Throws:

IllegalArgumentException - when the event or the filter is null

Since:

1.10

See Also:

• send(ServerSentEvent)
• close(SseConnectionFilter)

send

public int send(CharSequence data)

Sends an event with the provided text as its data to all the open connections of this broadcaster.

Parameters:

data - the data of the event

Returns:

the number of connections that received the event

Since:

1.10

See Also:

• send(ServerSentEvent)
• send(Template)

send

public int send(Template template)

Sends an event with the content of the provided template as its data to all the open connections of this broadcaster.

The filtered tags of the template will be resolved against the context of each individual connection.

Parameters:

template - the template whose content will be used as event data

Returns:

the number of connections that received the event

Since:

1.10

See Also:

• send(ServerSentEvent)
• send(Template, String)

send

public int send(Template template,
 String blockId)

Sends an event with the content of a single template block as its data to all the open connections of this broadcaster.

The filtered tags of the template will be resolved against the context of each individual connection.

Parameters:

template - the template that contains the block

blockId - the ID of the block whose content will be used as event data

Returns:

the number of connections that received the event

Since:

1.10

See Also:

• send(ServerSentEvent)
• send(Template)

comment

public int comment(String comment)

Sends a comment line to all the open connections, typically used as a keep-alive heartbeat that also reaps the stale connections.

Parameters:

comment - the comment text

Returns:

the number of connections that received the comment

Since:

1.10

See Also:

• send(ServerSentEvent)
• heartbeat(java.time.Duration)

heartbeat

public SseBroadcaster heartbeat(Duration interval)

Automatically sends a keep-alive comment to all the open connections at a fixed interval.

Heartbeats prevent intermediaries from closing idle connections and reap the connections whose clients have disconnected, which would otherwise hold on to their resources until the next application-initiated broadcast. A client that stops reading without disconnecting stalls its write instead of failing it, so such a connection is only reaped when the connection timeout of the servlet container fails the write. Intervals in the range of 15 to 30 seconds are typical, and they should stay shorter than the connection timeout of the container so that healthy idle streams are kept alive: the embedded Jetty Server closes connections that have been idle for 30 seconds by default, which makes a 15 second heartbeat appropriate with that default, while a 30 second heartbeat races the timeout. Embedded Tomcat defaults to 60 seconds.

The heartbeat runs on a daemon thread. Calling this method again will replace the previous interval, while stopHeartbeat() stops the heartbeat, as does close().

Parameters:

interval - the interval between heartbeats

Returns:

this broadcaster instance

Since:

1.10

See Also:

• comment(java.lang.String)
• stopHeartbeat()

stopHeartbeat

public SseBroadcaster stopHeartbeat()

Stops the heartbeat.

The connections and the event history are unaffected, and a new heartbeat can be established afterwards. Calling this method when no heartbeat is active has no effect.

Returns:

this broadcaster instance

Since:

1.10

See Also:

• heartbeat(java.time.Duration)

connectionCount

public int connectionCount()

Retrieves the number of connections that are currently registered with this broadcaster.

Stale connections are detected when the servlet container reports that their request has ended, and otherwise when sending to them fails, which means that this count can still include clients that have already disconnected.

Returns:

the number of registered connections

Since:

1.10

See Also:

• close(SseConnectionFilter)

lastEventId

public String lastEventId()

Retrieves the ID of the most recently sent event when history is enabled.

Pages can embed this ID in the URL of their SSE connection as the lastEventId parameter, so that the events that occur between the rendering of the page and the establishment of the stream will be replayed.

Returns:

the ID of the most recently sent event; or

the ID with sequence number 0 when no event has been sent yet

Since:

1.10

See Also:

• history(int)
• SseConnection.lastEventId()

historyStats

public SseHistoryStats historyStats()

Retrieves a snapshot of the state of the event history, which is intended to help you tune the history capacity.

Returns:

the current history statistics

Since:

1.10

See Also:

• SseHistoryStats
• history(int)

close

public int close(SseConnectionFilter filter)

Closes the connections of this broadcaster that match a filter.

This makes it possible to terminate connections when their context is no longer appropriate, for instance after the user that they belong to logged out or lost a permission:

broadcaster.close(connection -> {
     var identity = Identified.getIdentity(connection.context());
     return identity != null && identity.getLogin().equals(login);
 });

The filter receives each connection like send(ServerSentEvent, SseConnectionFilter) does, with the context of the request that established the connection. The event history and the heartbeat are unaffected.

Parameters:

filter - the filter that determines which connections are closed

Returns:

the number of connections that were closed

Throws:

IllegalArgumentException - when the filter is null

Since:

1.10

See Also:

• close()
• send(ServerSentEvent, SseConnectionFilter)

close

public void close()

Closes all the current connections of this broadcaster, clears the event history and stops the heartbeat.

The broadcaster itself remains usable: new connections can still register, subsequent events will be sent to them, and a heartbeat can be established again.

Specified by:

close in interface AutoCloseable

Since:

1.10

See Also:

• close(SseConnectionFilter)