How to stream cryptocurrency prices with Quarkus Websocket in Java
Quarkus is a Java framework that aims to optimize the development of cloud-native applications. It offers a fast boot time, low memory footprint, live reload, and native compilation. Quarkus also provides a rich set of extensions that integrate with popular libraries and technologies, such as RESTEasy, Hibernate, MicroProfile, Kafka, and more.
One of the extensions that Quarkus supports is WebSocket, which is a protocol that enables bidirectional communication between a client and a server over a single TCP connection. WebSocket is useful for scenarios where you need to exchange real-time data, such as chat applications, online games, or stock market updates.
In this article, we will show you how to create a simple Quarkus application that uses WebSocket to stream cryptocurrency prices from the Binance API WebSocket Streams. We will use the Quarkus WebSocket extension, which is based on the Java API for WebSocket (JSR 356) and the Undertow web server.
Set up the Websocket Client
First, we need to create a WebSocket client that connects to the Binance WebSocket API and receives the price updates.
@ClientEndpoint
public class BinanceWebSocketClient {
@OnOpen
public void open(Session session) {
Log.info("Connected open: " + session.getId() + " " + session.getRequestURI());
}
@OnMessage
void onMessage(String tickerData) {
Log.info("Incoming ticker data: " + tickerData);
}
@OnError
public void onError(Session session, Throwable throwable) {
Log.info("Error: " + session.getId() + " " + throwable.getMessage());
}
@OnClose
public void onClose(Session session) {
Log.info("Session " + session.getId() + " has ended");
}
}
We have defined a WebSocket client annotated with the @ClientEndpoint annotation so that it is used to define a POJO (Plain Old Java Object) that will interact with a WebSocket server.
The various @OnOpen, @OnMessage, @OnError, and @OnClose annotations at the method level indicate the WebSocket lifecycle methods that will be invoked at different stages of the WebSocket interaction. In our case we are just logging relevant information about the WebSocket connection, received messages, errors, and session closures.
The Session object injected in some of those methods represents a communication channel between two WebSocket endpoints. After a successful WebSocket handshake, the web socket implementation provides an open session to the endpoint. This session allows the endpoint to register interest in incoming messages by providing a message handler and to send messages to the other endpoint using a RemoteEndpoint object. Once closed, the session becomes invalid for further use, and attempting to call its methods (except the close() methods) will result in an exception.
Binance ticker available symbols
As a trading platform, Binance provides the following public endpoint to get a list of all symbols and their prices from the Binance API:
You can filter out the list of all available symbols using the following curl and jq command:
curl https://api.binance.com/api/v3/ticker/price | jq '.[].symbol'
We are going to use the BTCUSDT symbol for this example, but you can use any other symbol from the list, even multiple symbols at the same time, thanks to the multi-threading and concurrent programming features that java.util.concurrent package provides.
Factory class to create specific WebSocket client (BTCUSDT, ETHUSDT, etc)
To create the WebSocket client based on the symbol, we are going to use a factory class, so that we can inject the symbol as a parameter in the creation method.
Let's first define the Binance websocket URL in the application.properties file:
binance.url=wss://stream.binance.com:9443/ws/{symbol}@ticker
Then we can create the factory class and load the URL from the properties file:
@ApplicationScoped
public class BinanceRunnerFactory {
final String binanceUrl;
public BinanceRunnerFactory(
@ConfigProperty(name = "binance.url") final String binanceUrl) {
this.binanceUrl = binanceUrl;
}
...
To run the WebSocket client and receive messages continuously, we need to run the client on a separate thread and let it be managed by the ExecutorService. So the idea is to create a "Binance Runnable "class based on the symbol from the factory class and then submit it to the ExecutorService:
public Runnable create(final String symbol) {
try {
final URI uri = new URI(binanceUrl.replace("{symbol}", symbol));
return () -> {
try (Session session = ContainerProvider.getWebSocketContainer().connectToServer(
BinanceWebSocketClient.class,
uri)) {
Log.info("Connection with websocket session id: " + session.getId() + " established");
Thread.currentThread().join();
} catch (DeploymentException | IOException e) {
Log.error("Something went wrong on BinanceRunnerFactory", e);
throw new RuntimeException(e);
}catch (InterruptedException e) {
Thread.currentThread().interrupt();
Log.info("Thread interrupted");
}
};
} catch (URISyntaxException e) {
Log.error("Invalid binance url", e);
throw new RuntimeException(e);
}
We are using the ContainerProvider.getWebSocketContainer().connectToServer method to connect to the WebSocket server. This method returns a Session object that represents a conversation between the client and the server. The Session object is used to send messages to the server and for querying the state of the connection.
Also, note that we are using the Thread.currentThread().join() method to keep the thread alive and receive messages continuously, in particular, we are using the Thread.currentThread().interrupt() method to interrupt the thread when the application is stopped, for example when we press Ctrl+C in the terminal.
The entry point to run the WebSocket client
Finally, we can create the entry point to run the WebSocket client:
@ApplicationScoped
public class DemoQuarkusBinanceWebsocketApp {
@Inject
BinanceRunnerFactory binanceRunnerFactory;
private ExecutorService executorService;
void onStart(@Observes StartupEvent ev) {
executorService = Executors.newCachedThreadPool();
executorService.submit(binanceRunnerFactory.create("btcusdt"));
}
...
We are using the @Observes StartupEvent annotation to listen to the startup event and then we are using the ExecutorService to run the WebSocket client on a separate thread.
Build and Run the application
Clone the sample code from the GitHub repository and run the following command to build and run the application in Dev mode:
./gradlew quarkusDev
You should see the following output in the terminal:
__ ____ __ _____ ___ __ ____ ______
--/ __ \/ / / / _ | / _ \/ //_/ / / / __/
-/ /_/ / /_/ / __ |/ , _/ ,< / /_/ /\ \
--\___\_\____/_/ |_/_/|_/_/|_|\____/___/
2023-08-15 21:58:36,924 INFO [io.und.websockets] (Quarkus Main Thread) UT026004: Adding annotated client endpoint class dev.dvalentino.demo.quarkus.binance.websocket.client.BinanceWebSocketClient
2023-08-15 21:58:37,056 INFO [io.quarkus] (Quarkus Main Thread) demo-quarkus-binance-websocket 1.0.0-SNAPSHOT on JVM (powered by Quarkus 3.2.3.Final) started in 0.901s.
2023-08-15 21:58:37,056 INFO [io.quarkus] (Quarkus Main Thread) Profile dev activated. Live Coding activated.
2023-08-15 21:58:37,057 INFO [io.quarkus] (Quarkus Main Thread) Installed features: [cdi, smallrye-context-propagation, vertx, websockets-client]
2023-08-15 21:58:39,081 INFO [dev.dva.dem.qua.bin.web.cli.BinanceWebSocketClient] (pool-8-thread-1) Connected open: 7BWODyNdctvBwrZWMU25kw8QNn3MHFszWXc_JYOp wss://stream.binance.com:9443/ws/btcusdt@ticker
2023-08-15 21:58:39,082 INFO [dev.dva.dem.qua.bin.web.BinanceRunnerFactory] (pool-8-thread-1) Connection with websocket session id: 7BWODyNdctvBwrZWMU25kw8QNn3MHFszWXc_JYOp established
2023-08-15 21:58:39,687 INFO [dev.dva.dem.qua.bin.web.cli.BinanceWebSocketClient] (vert.x-eventloop-thread-0) Incoming ticker data: {"e":"24hrTicker","E":1692133119365,"s":"BTCUSDT","p":"-201.46000000","P":"-0.685","w":"29341.61752250","x":"29400.83000000","c":"29199.37000000","Q":"0.01340000","b":"29199.36000000","B":"28.94628000","a":"29199.37000000","A":"6.59511000","o":"29400.83000000","h":"29499.26000000","l":"29059.60000000","v":"27283.83990000","q":"800551994.89102990","O":1692046719365,"C":1692133119365,"F":3190518676,"L":3191099367,"n":580692}
2023-08-15 21:58:40,603 INFO [dev.dva.dem.qua.bin.web.cli.BinanceWebSocketClient] (vert.x-eventloop-thread-0) Incoming ticker data: {"e":"24hrTicker","E":1692133120127,"s":"BTCUSDT","p":"-201.47000000","P":"-0.685","w":"29341.61664443","x":"29400.83000000","c":"29199.36000000","Q":"0.01196000","b":"29199.36000000","B":"29.00110000","a":"29199.37000000","A":"7.28792000","o":"29400.83000000","h":"29499.26000000","l":"29059.60000000","v":"27284.00831000","q":"800556912.35543080","O":1692046720127,"C":1692133120127,"F":3190518676,"L":3191099377,"n":580702}
...
The application will keep receiving ticker data from the WebSocket server until you stop it by pressing Ctrl+C in the terminal.
As mentioned before, you are free to call BinanceRunnerFactory.create(<symbol>) multiple times to create multiple WebSocket clients and receive prices updates for multiple symbols, for example:
void onStart(@Observes StartupEvent ev) {
executorService = Executors.newCachedThreadPool();
executorService.submit(binanceRunnerFactory.create("btcusdt"));
executorService.submit(binanceRunnerFactory.create("ethusdt"));
executorService.submit(binanceRunnerFactory.create("adausdt"));
...
Conclusion
Quarkus is a great framework to build cloud-native applications and it provides a lot of features out of the box, including the WebSocket client. In this article, we have seen how to use the WebSocket client to connect to the Binance WebSocket API and receive real-time price updates for a given symbol.
It allows you to create a WebSocket connection from your Quarkus application to any WebSocket server and receive real-time data. You can use it to build a real-time dashboard, a real-time notification system, or anything else that requires real-time data.
The full source code of the application is available on GitHub