Subscriptions are for sure one of the more interesting GraphQL features, as they invert the traditional flow of client-server communication: Instead of a client asking for some data, you have a server that can send data to the client anytime, as soon as they're available. It's easy to get fascinated by a world where you don't have to check any minute if the cake in the oven is done, but it's the oven itself that tells you that the cake is ready!

Subscriptions are awesome, but they're also complex to implement, they require an elaborate architecture, and it's often difficult to find exhaustive documentation or working examples.

In this article, you will learn how to implement GraphQL subscriptions in Java, using Spring Boot and the Kickstart library.

Let's Start!

We'll learn how to work with subscriptions by implementing a simple tic-tac-toe game.

❌ │ ⭕ │ 
──────────────
❌ │ ❌ │ 
──────────────
⭕ │ ⭕ │ ❌

In the game, we want to use subscriptions for two purposes:

• to notify a user that another player joined the game and the match can start
• to notify a player that the opponent has made a move

The following picture shows the game flow:

You can find the full GraphQL schema of the application on GitHub; anyway, the game flow is straightforward:

• The first player starts the game by calling the startGame mutation and then waits for an opponent.
• Using the opponentArrived subscription, the first player can be notified when another user comes in.
• When another player calls the startGame mutation, the server sends some data through the opponentArrived subscription: The match can start!
• Each player, in turn, makes a move: While waiting, the other player can use the opponentMove subscription to be notified of the move made by the opponent; this loop goes on until the game ends.

Compared to queries and mutations, subscriptions require us to take care of some additional aspects:

• A different and more complex communication protocol
• A continuous data flow from the server to the client
• A dedicated authentication system
• External persistence and communication layers to maintain the status of the subscription

For sake of simplicity, in our example application, we'll use an in-memory solution, so we'll overlook the latter point. Anyway, please keep in mind that in a real production environment, you'll probably need an external broker or a messaging system (like, for example, Apache Kafka or RabbitMQ).

WebSocket Handling

Let's go deeper into the communication protocol: A subscription opens a persistent channel between the client and the server. In most GraphQL implementations, this channel is established using the WebSocket protocol. For those who are not familiar with this technology, it is a protocol that creates a channel between two parties and can be used to communicate in both directions. To establish the initial connection, the process starts with a traditional HTTP request/response (a handshake), then the communication goes on through the WebSocket protocol.

The GraphQL spec doesn't prescribe the details of the messages that have to be exchanged between the client and the server; anyway, the well-known company Apollo GraphQL wrote its own specification, which has become a de facto standard. It requires the use of a specific format for the messages exchanged through the WebSocket, as shown in the following figure:

Well, now that the boring part is over, let's finally look at the code.

Luckily for us, the Kickstart library takes care of all aspects related to the WebSocket communication protocol and it also implements the Apollo GraphQL specifications. We just have to define a Spring bean that implements the GraphQLSubscriptionResolver interface.

According to the GraphQL spec, a subscription must implement the pub-sub pattern, so Kickstart expects that the resolver method returns an implementation of the Publisher interface from Reactive Streams.

Using the Data Stream

To make our subscriptions work, we have to know how to handle a Reactive Streams' Publisher and how to feed it.

There are several good Java implementations of the Reactive Streams specs: If you're working with Spring, I recommend you try Project Reactor because it's fully integrated with this framework. Let's see how to use it inside our application.

In our basic implementation, the data stream will be completely held in memory inside the application, and so we just need to declare a couple of Spring beans:

I won't go deeper into the details of Project Reactor, but if you're not familiar with reactive programming, you can think of a reactive stream as if it were just a pipe through which the data flow.

The two beans defined above represent the two ends of the pipe:

• the sink is where data goes in
• the flux is where data comes out

The Game object, defined in our application, represents the data type that will flow through the stream.

Several clients (subscribers) can connect to the reactive stream (using the GraphQL subscription) and receive the data that flow into it.

Now that we have the two beans defined in the application, we can autowire them in our resolvers and use them to put or get data into the stream.

Let's see how to feed the stream for the two subscriptions of our application:

• To implement the opponentArrived subscription, we need to put data into the stream when the second player calls the startGame mutation.
• To implement the opponentMove subscription, we have to put data into the sink every time one of the players makes a move, that is, every time the move mutation is invoked.

So we just need to autowire the sink bean in the mutation resolver and call the tryEmitNext() method to put data into the stream:

Now we can autowire the flux bean in our root subscription resolver and return it to the client.

Since both the subscriptions share the same reactive stream, we have to distinguish the data that should go to the opponentArrived publisher and the data that should go to the opponentMove publisher. For this purpose, we can use the isStarted() method of the Game object: If the game isn't started yet, we have to send data to opponentArrived; otherwise, we have to send data to opponentMove:

As you can see, in the opponentArrived() method, the data flowing through the flux is also mapped to return just the name of the second player (according to the GraphQL schema).

Authenticate a Subscription

Now we have to deal with the last issue: We have a single shared reactive stream in our application, but there could be several users playing the game at the same time. And we want to send to the players only information about their own match. How can we achieve it?

We need to put in place an authentication system for our subscriptions so that we can filter out the data belonging to other players before sending them through the WebSocket channel established by a specific user.

We can use Spring security support to manage the authentication in our subscription resolver, but we'll need a dedicated system to authenticate a WebSocket connection.

Again, the Kickstart library comes to our aid, allowing us to intercept the main events occurring during a subscription connection. All we have to do is to implement the ApolloSubscriptionConnectionListener interface and:

• override the onConnect() method to intercept the GQL_CONNECTION_INIT message
• override the onStart() method to intercept the GQL_START message

The most common way to handle a subscription authentication is using the payload transmitted with the GQL_CONNECTION_INIT message.

Unfortunately, this authentication system will work only if the client sends both the GQL_CONNECTION_INIT and the GQL_START messages together so that on the server side they are handled by the same thread.

Otherwise, if the two messages are sent separately, they could potentially be served by two different threads. In this case, the authentication will fail unless it's also handled when the server receives the GQL_START message.

In our code, we decided to handle the authentication in both cases, to be sure that the server will work with most of the clients. In the SubscriptionConnectionListener we will implement both the onConnect() and the onStart() methods: In the first method we will handle the authentication using the payload, and in the second one we will use the Authorization header of the original handshake request:

Now the connection is authenticated, and we can use the current user identity in our subscription resolvers to filter out all the messages that should not be sent to the subscribed user, for example, sending the data only if the current user is on a move in that particular game:

That's it! We finally have our tic-tac-toe GraphQL server implementing the two subscriptions! Our work is done and the game can start! ❌ ⭕️

Recap

Let's quickly recap what we've learned today:

• To build a subscription resolver, it's enough to implement the GraphQLSubscriptionResolver interface.
• A subscription resolver must return an implementation of the Publisher interface from Reactive Streams.
• Using Project Reactor, we can define a reactive stream as a bean that can be autowired and used in the GraphQL resolvers.
• Authentication in subscriptions can be handled by implementing the ApolloSubscriptionConnectionListener interface.

You can find the complete code of the tic-tac-toe server application on GitHub together with a simple CLI client to test it (made using the Apollo Android client and the OkHttp library).

Thank you for reading.

Resources