Simple and scalable event subscription with WebSockets, STOMP, SockJS and Spring Framework 4.0

The first intermediate release of the Spring Framework 4.0 M1 provided server-side support for SockJS - the best and most complete alternative implementation of WebSocket. You will need this backup option in browsers that do not support WebSocket and in situations where a proxy interferes with their use. Simply put, SockJS allows you to build WebSocket-applications today, which, among other things, are able to transparently switch to backup capabilities.

But even with backup options, problems remain. A socket is a rather low-level abstraction and the vast majority of web applications today are not adapted for sockets. This is why the WebSocket protocol defines a sub-protocol mechanism that, in essence, allows (and encourages) the use of higher-level protocols over WebSocket, just as we use HTTP over TCP.

The second intermediate release of the Spring Framework 4.0 M2 allows the use of high-level messaging protocols over WebSocket. To demonstrate this, we will analyze an example application.
')

Stock Portfolio app

This test application, available on Github, allows users to download a portfolio of positions, buy and sell stocks; sends price quotes and displays position updates. This is a fairly simple application. However, it covers a number of common tasks that may arise in browser-based messaging applications.


How exactly do we implement an application like this? In HTTP and REST, we are used to relying on URLs and HTTP methods that express what needs to be done. But there is only a socket and a bunch of messages. How to tell who the message is and what it means?


The browser and server must agree on a common message format before this semantics can be expressed. There are several protocols that can help. We chose STOMP in this release due to its simplicity and broad support .

Simple / Streaming Text-Oriented Messaging Protocol (STOMP)

STOMP is a very simple messaging protocol. Frame based on HTTP pattern. A frame consists of a command, optional headers and an optional body.

For example, the Stock Portfolio application can send quotes, the client will send a SUBSCRIBE frame, where the destination header indicates what exactly it subscribes to:

SUBSCRIBE id:sub-1 destination:/topic/price.stock.* 

As soon as stock quotes become available, the server sends a MESSAGE frame with the appropriate destination and subscription ID, as well as the content-type header and body:

 MESSAGE subscription:sub-1 message-id:wm2si1tj-4 content-type: application/json destination:/topic/stocks.PRICE.STOCK.NASDAQ.EMC {\"ticker\":\"EMC\",\"price\":24.19} 

To combine this all in the browser, we use stomp.js and the SockJS client :

 var socket = new SockJS('/spring-websocket-portfolio/portfolio'); var client = Stomp.over(socket); var onConnect = function() { client.subscribe("/topic/price.stock.*", function(message) { //   }); }; client.connect('guest', 'guest', onConnect); 

This is a serious advance! We have a standard message format and customer support.

Now we can move further - to the server side.

Message broker

Message-broker is a typical server solution, where messages are sent between traditional brokers such as RabbitMQ, ActiveMQ, etc. Most, if not all, support STOMP over TCP, some - WebSocket, but RabbitMQ has advanced the furthest, and it also works with SockJS, among other things.

Our architecture will look like this:


This is a reliable and scalable solution, but perhaps not the most optimal for the task at hand. Message-broker is commonly used within the enterprise. Expose it directly to the network does not pull on the perfect solution.

If we learned something from the REST approach, it is that we should not disclose the details of the implementation of our system, the database device or the domain model.

In addition, as a Java developer, you probably want to customize permissions, validations, add application logic. In the message-broker approach, the application server is behind the broker, which is significantly different from what most web developers are used to.

That's why libraries like socket.io are popular . It is simple and it focuses on the needs of web applications. On the other hand, we should not ignore the message-broker's ability to process messages, they are really good at it - a difficult dilemma. Take the best of both.

+ Message-broker application

Another approach is to make an application that processes incoming messages and acts as an intermediary between web clients and a broker. Messages from clients to the broker can be sent through the application, the return message will also pass through the application to the client. This allows the application to determine the type of message and the “destination” header, then decide to process it on its own or redirect it to the broker.


We chose this approach. To illustrate it better here are some scenarios.

Loading portfolio items

• The client requests a portfolio of positions
• An application processes a request by loading and returning data for a subscription.
• Message-broker does not participate in this interaction.

Subscribe to stock quotes

• Customer sends subscription request
• The application sends a message to the broker
• Message-broker sends the message to all subscribed customers.

Get stock quotes

• QuoteService sends a stock quote message to a broker
• Message-broker sends the message to all subscribed customers.

Conducting transaction

• Client sends trade request
• The application does not process it, all transactions go through TradeService
• Message-broker does not participate in this interaction.

Getting updated positions

• TradeService sends a position update message to the broker's queue
• Message-broker sends item update to customer
• Sending messages to a specific user is discussed in detail below.

Strictly speaking, the use of a message broker is optional. We offer a “simple” out-of-the-box alternative. However, the use of a message broker is recommended for scalability and for deployment with multiple application servers.

Code snippets

Let's look at some examples of server and client code.

This is a request for a portfolio of positions from portfolio.js :

 stompClient.subscribe("/app/positions", function(message) { self.portfolio().loadPositions(JSON.parse(message.body)); }); 

On the server side, the PortfolioController detects a request and returns a portfolio of positions, demonstrating the request-response interaction, which is very often used in web applications. Because we use Spring Security to protect HTTP requests, including one initial, related to a handshake with WebSocket, as an argument to the method in the example below, the Spring Security user principal obtained from HttpServletRequest is passed:

 @Controller public class PortfolioController { // ... @SubscribeEvent("/app/positions") public List<PortfolioPosition> getPortfolios(Principal principal) { String user = principal.getName(); Portfolio portfolio = this.portfolioService.findPortfolio(user); return portfolio.getPositions(); } } 

Here, portfolio.js sends a trade request:

 stompClient.send("/app/trade", {}, JSON.stringify(trade)); 

On the server side, the PortfolioController submits it for execution:

 @Controller public class PortfolioController { // ... @MessageMapping(value="/app/trade") public void executeTrade(Trade trade, Principal principal) { trade.setUsername(principal.getName()); this.tradeService.executeTrade(trade); } } 

PortfolioController can also handle unexpected exceptions by sending a message to the user:

 @Controller public class PortfolioController { // ... @MessageExceptionHandler @ReplyToUser(value="/queue/errors") public String handleException(Throwable exception) { return exception.getMessage(); } } 

What about sending messages to all subscribed customers? So QuoteService sends quotas:

 @Service public class QuoteService { private final MessageSendingOperations<String> messagingTemplate; @Scheduled(fixedDelay=1000) public void sendQuotes() { for (Quote quote : this.quoteGenerator.generateQuotes()) { String destination = "/topic/price.stock." + quote.getTicker(); this.messagingTemplate.convertAndSend(destination, quote); } } } 

And so TradeService sends the update of positions after it has executed the transaction:

 @Service public class TradeService { // ... @Scheduled(fixedDelay=1500) public void sendTradeNotifications() { for (TradeResult tr : this.tradeResults) { String queue = "/queue/position-updates"; this.messagingTemplate.convertAndSendToUser(tr.user, queue, tr.position); } } } 

And just in case if you are interested ... yes, PortfolioController may contain Spring MVC methods (for example, @RequestMapping ), as suggested by the developer who previously built online games in this ticker :

And just like the QuoteService and TradeService, the Spring MVC methods of the controller can also post messages.

Messaging support in Spring applications

For a long time, Spring Integration provides first-class abstractions for well-known Enterprise Integration patterns, as well as lightweight messaging. While working on this release, we realized that the latter was exactly what we needed to rely on.

As a result, and I am pleased to announce this, we moved the Spring Integration type set to a new Spring Framework module, which is predictably called spring-messaging . In addition to basic abstractions, such as Message, MessageChannel, MessageHandler, etc., the module contains all the annotations and classes to support the new functions described in this post.

Now, with this in mind, we can look at the diagram of the internal structure of the Stock Portfolio application:


StompWebSocketHandler puts incoming client messages on the dispatch channel. This channel has 3 subscribers. The first is responsible for the annotated methods, the second is the message sent to the STOMP broker, while the third processes the messages of individual users by converting the destination address into a queue of unique names to which the client has subscribed.

By default, the application works with a “simple” broker provided for acquaintance. As explained in the README , you can choose between a “simple” and full-featured broker by activating and deactivating profiles.


Another possible configuration change is to move from the Executor to the implementation of a MessageChannel based on Reactor . The Reactor project recently released the first intermediate release and is also used to manage TCP connections between the application and the Message-broker.

Here you can see the complete configuration of the application, which also includes the new configuration of Spring Security. You may also be interested to learn about the improved support for configs in STS .

Send messages to individual users

It's easy to understand how you can send messages to all subscribed clients — just send a message to the channel. Slightly more complicated is the situation with sending a message to a specific client. For example, you caught an exception and want to send an error message. Or you received confirmation of the completion of the transaction and want to please the user.

In traditional messaging applications, this task is usually solved by creating a time queue and setting the reply-to header in all messages that imply a reply. It works, but looks rather cumbersome. The client should not forget to put the necessary headers on all applicable messages and the server application may need to monitor them. Sometimes this information may not be available (for example, if you use HTTP POST instead of messaging).

To support this requirement, we send a unique queue suffix to each connected client. A suffix can then be added to the identifier to create a unique queue name:

 client.connect('guest', 'guest', function(frame) { var suffix = frame.headers['queue-suffix']; client.subscribe("/queue/error" + suffix, function(msg) { //   }); client.subscribe("/queue/position-updates" + suffix, function(msg) { //    }); }); 

Then on the server side, you can add the @ReplyToUser annotation to the method marked @MessageExceptionHandler (or any other message handling method) and send the return value as a message:

 @MessageExceptionHandler @ReplyToUser(value="/queue/errors") public String handleException(Throwable exception) { // ... } 

All other classes, such as TradeService, can use messagingTemplate to achieve the same:

 String user = "fabrice"; String queue = "/queue/position-updates"; this.messagingTemplate.convertAndSendToUser(user, queue, position); 

In both cases, inside we have the suffix of the user queue (through the configured UserQueueSuffixResolver ) in order to restore the correct queue name. At the moment there is only one implementation of a simple resolver. However, it is fairly easy to add a Redis- based implementation that will support the function regardless of whether the user is connected or another application server.

findings

Hope this was a useful introduction to new functionality. Instead of making the post even longer, I advise you to study the example and try to apply it in the applications you write or intend to write. This is an ideal time for feedback, because how we plan release candidate in early September.