A Websocket Client using the Java 11 WebSocket API

For a small crypto project i wanted to use the Bitfinex WebSocket API to get market data in real time. The full code is available on github.

Since Java SE 11 the JDK contains a client WebSocket API. The javadoc contains some code examples but you cannot use these examples out of the box. Searching the net for „java websocket client“ reveals mostly examples and guides for the old JSR 352 websocket stuff. Setting up everything to get things done was not so straightforward as expected, so Iwrote this little tutorial.

Whenever it suits, I use vert.x in my projects, and there is a websocket client API, too. But sadly the vert.x websocket client has a little shortcoming, it does not work with redirects ☹ Although we do not use the vert.x websocket client, we use vert.x as base of our small application.

Enough words, we start coding. To build and connect to a websocket server, the builder needs a listener, which will listen to incoming data packets:

class BitfinexListener(val vertx: Vertx) : WebSocket.Listener {

    override fun onOpen(webSocket: WebSocket?) {
        super.onOpen(webSocket)
        LOGGER.info("websocket opened")
        this.vertx.periodicStream(60000).toObservable()
          .subscribe { i ->
            val pingTxt = JsonObject().put("event", "ping")
                           .put("cid", Random(2020).nextInt())
                           .encode()
            webSocket?.sendText(pingTxt, true)?
             .thenRun {-> LOGGER.info("sent ping {}", pingTxt)}
          }
    }

    var parts: MutableList<CharSequence?> = 
            MutableList(0) { index: Int -> "" }
    var accumulatedMessage: CompletableFuture<*> =
            CompletableFuture<Any>()

    override fun onText(webSocket: WebSocket,
                         message: CharSequence?,
                         last: Boolean): CompletionStage<*>? {
        parts.add(message)
        webSocket.request(1)
        if (last) {
            val completeMessage = parts.joinToString(separator = "") 
                   { charSequence -> charSequence ?: "" }
            parts.clear()
            accumulatedMessage.complete(null)
            val cf: CompletionStage<*> = accumulatedMessage
            accumulatedMessage = CompletableFuture<Any>()
            onMessage(completeMessage)
            return cf
        }
        return accumulatedMessage
    }

    fun onMessage(message: String) {
        val bitfinexMessage = Json.decodeValue(message)
        // ... see repo at github for full code
    }
}

We override the onOpen method to establish a periodic ping to bitfinex. One important thing is, that when you override onOpen you must call super.onOpen, otherwise the client does not send any data to the server.

It costs me a lot of time to find this error. As the Listener is an interface i am not used to call super methods of interfaces. But in this case it’s important. The concept of default methods in interfaces introduced in Java 8 has still not become second nature to me.

The onText message collects all transmitted text data until a text is complete (this is usually the case in one call to inText). On text completion the text is send via the vert.x eventbus in the onMessage method.

Now we can setup a verticle that represents a connection to bitfinex using this listener:

class BitfinexConnection : AbstractVerticle() {

    var webSocket: WebSocket? = null

    override fun start() {
        LOGGER.info("deploying BitfinexConnection")
        val subs = vertx.sharedData()
                .getLocalMap<Int,String>("bitfinex.subscriptions")
        val listener = BitfinexListener(this.vertx, subs)
        val client = HttpClient.newHttpClient()
        val uri = URI.create("wss://api-pub.bitfinex.com/ws/2")
        this.webSocket = client.newWebSocketBuilder()
                     .buildAsync(uri), listener).join()
        vertx.eventBus()
             .consumer<JsonObject(BITFINEX_EB_ADDRESS)
             .handler { jsonMsg ->
            
          if ( webSocket==null || webSocket?.isOutputClosed()!!) {
              jsonMsg.reply(JsonObject()
                            .put("message", "websocket closed")
                            .put("statusCode",503))
                return@handler
          }
          val bitfinexMessage = jsonMsg.body().encode()
          this.webSocket?.sendText(bitfinexMessage,true)?
           .thenRun{ LOGGER.debug("delivered {} ", bitfinexMessage)}
        }
    }
}

The class is easy and straightforward, in the start method of the verticle a websocket connection to the public bitfinex api isinitialized and started. Via the vert.x eventbus messages should be sent directly to the websocket, therefore we start a consumer on the address BITFINEX_EB_ADDRESS.

At last step we put everything together and try to subscribe to the tBTCUSD ticker:

fun main() {
  val vertx = Vertx.vertx()
  val symbol = "tBTCUSD"
  vertx.rxDeployVerticle(BitfinexConnection::class.java.name)
    .subscribe(
        { id ->
            LOGGER.info("deployed bitfinex connection {}", id)
            val address = "ticker." + symbol
            vertx.eventBus()
               .consumer<JsonArray>(address)
               .handler { jsonMsg ->
                  LOGGER.info("received {} {}", address,
                        jsonMsg.body().encodePrettily())
               }
            val subscribeMessage = JsonObject()
                  .bfxSubscribeTickerMessage(symbol)           
            vertx.eventBus()
                .send(BITFINEX_EB_ADDRESS, subscribeMessage)

       },
       { t: Throwable? -> 
         LOGGER.error("deployment failed", t) }
   )
}

The main method starts Vertx and deploys the bitfinex websocket verticle. When the deployment is finished, we send a message to the websocket verticle with a subscription payload and start a consumer on a special eventbus address.

You can find all the code on github, clone it, assure that you have Java 11 or higher installed and run it with:

./gradlew run

I hope this tutorial will give you a fast startup with your websocket client code in your project.