WebsocketVerticle

WebsocketVerticle.java

@Override
public void start() throws Exception {
    super.start();
    router.route("/ws/websocket/*").handler(request -> {
        ServerWebSocket webSocket = request.request().upgrade();
        logger.debug("New connection {}", webSocket.binaryHandlerID());
        webSocket.handler(buffer -> {
            logger.debug("Received: {} {} ", webSocket.binaryHandlerID(), new String(buffer.getBytes()));
            JsonObject json = new JsonObject(new String(buffer.getBytes()));
            json.put("senderId", webSocket.binaryHandlerID());
            vertx.eventBus().publish("chat.broadcast", json);
        });
        MessageConsumer<JsonObject> consumer = vertx.eventBus().<JsonObject>consumer("chat.broadcast", messageHandler->{
            JsonObject body = messageHandler.body();
            String senderId = body.getString("senderId");
            if(!senderId.equals(webSocket.binaryHandlerID())){
                body.remove("senderId");
                webSocket.writeBinaryMessage(Buffer.buffer(body.encode()));
            }
        });
        webSocket.exceptionHandler(e -> logger.error(e.getMessage(), e));
        webSocket.closeHandler(c -> {
            logger.debug("Websocket {} closed.", webSocket.binaryHandlerID());
            consumer.unregister();
        });
    });
    vertx.createHttpServer().requestHandler(router::accept).listen(8080);
}