A major part of my GSoC proposal was adding stream API to loklak server. In a previous blog post, I discussed the addition of Mosquitto as a message broker for MQTT streaming. After testing this service for a few days and some minor improvements, I was in a position to expose the stream to outside users using a simple API.
In this blog post, I will be discussing the addition of /api/stream.json endpoint to loklak server.
HTTP Server-Sent Events
Server-sent events (SSE) is a technology where a browser receives automatic updates from a server via HTTP connection. The Server-Sent Events EventSource API is standardized as part of HTML5 by the W3C.
This API is supported by all major browsers except Microsoft Edge. For loklak, the plan was to use this event system to send messages, as they arrive, to the connected users. Apart from browser support, EventSource API can also be used with many other technologies too.
Jetty Eventsource Plugin
For Java, we can use Jetty’s EventSource plugin to send events to clients. It is similar to other Jetty servlets when it comes to processing the arguments, handling requests, etc. But it provides a simple interface to send events as they occur to connected users.
To use this plugin, we can add the following line to Gradle dependencies –
The Event Source
An EventSource is the object which is required for EventSourceServlet to send events. All the logics for emitting events needs to be defined in the related class. To link a servlet with an EventSource, we need to override the newEventSource method –
If no channel is provided, the EventSource object will be null and the request will be rejected. Here, the MqttEventSource would be used to handle the stream of Tweets as they arrive from the Mosquitto message broker.
Cross Site Requests
Since the requests to this endpoint can’t be of JSONP type, it is necessary to allow cross site requests on this endpoint. This can be done by overriding the doGet method of the servlet –
Adding MQTT Subscriber
When a request for events arrives, the constructor to MqttEventSource is called. At this stage, we need to connect to the stream from Mosquitto for the channel. To achieve this, we can set the class as MqttCallback using appropriate client configurations –
By setting the callback to this, we can override the messageArrived method to handle the arrival of a new message on the channel. Just to mention, the client library used here is Eclipse Paho.
Connecting MQTT Stream to SSE Stream
Now that we have subscribed to the channel we wish to send events from, we can use the Emitter to send events from our EventSource by implementing it –
Closing Stream on Disconnecting from User
When a client disconnects from the stream, it doesn’t makes sense to stay connected to the server. We can use the onClose method to disconnect the subscriber from the MQTT broker –
In this blog post, I discussed connecting the MQTT stream to SSE stream using Jetty’s EventSource plugin. Once in place, this event system would save us from making too many requests to collect and visualize data. The possibilities of applications of such feature are huge.
This feature can be seen in action at the World Mood Tracker app.
- Using Mosquitto as a Message Broker for MQTT in loklak Server – https://blog.fossasia.org/using-mosquitto-as-a-message-broker-for-mqtt-in-loklak-server/.
- Using server-sent events – Mozilla Developers Network – https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events.
- Server-Sent Events W3C Specifications – https://www.w3.org/TR/2009/WD-eventsource-20090423/.
- More about Eclipse Paho Java Client – http://www.eclipse.org/paho/clients/java/.
- Event Source Example by marschall – https://github.com/marschall/event-source-sample.