GreetServiceVerticle.java

package com.example;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.eventbus.Message;
import io.vertx.core.json.JsonObject;
import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;

import java.util.HashSet;

/**
 * Created by developer on 30.04.16.
 */
public class GreetServiceVerticle extends AbstractVerticle {
    Logger logger = LoggerFactory.getLogger(GreetServiceVerticle.class);
    //HashMap<String, MessageProducer<String>> connectedUsers = new HashMap<>();
    HashSet<String> activeUsers = new HashSet<>();

    public void start() throws Exception {
        this.vertx.eventBus().consumer(Channels.CONNECTION_CHANNEL, (Message<JsonObject> event) -> {
            JsonObject msg = event.body();
            String socketId = msg.getString("socketId");
            boolean connected = msg.getBoolean("connected");
            if (connected) {
                onUserConnected(socketId);
            } else onUserLogout(socketId);
        });
        this.vertx.eventBus().consumer(Channels.USER_SAID_CHANNEL, (Message<String> msg)->{
            for(String sockId : activeUsers){
                String msgToBroadcast = msg.body();//String.format("[%s]: %s", msg.address(), msg.body());
                logger.info("msgToBroadcast="+msgToBroadcast+"/sockId="+sockId);
                vertx.eventBus().send(sockId, msgToBroadcast);
            }
        });
    }

    private void onUserConnected(String socketId){
        String msg = "user "+socketId+" connected!";
        logger.info(msg);
        for(String userSocketId : activeUsers){
            logger.info("sending to " + userSocketId);
            vertx.eventBus().send(userSocketId, msg);
        }
        activeUsers.add(socketId);
        logger.info("total:" + activeUsers.size());
    }
    private void onUserLogout(String socketId){
        activeUsers.remove(socketId);
        String msg = "user "+socketId+" disconnected!";
        logger.info(msg);
        logger.info("total:"+activeUsers.size());
        for(String userSocketId : activeUsers){
            logger.info("sending to " + userSocketId);
            vertx.eventBus().send(userSocketId, msg);
        }
    }
}

MyWSServerVerticle.java

package com.example;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Future;
import io.vertx.core.Handler;
import io.vertx.core.http.HttpServerOptions;
import io.vertx.core.http.ServerWebSocket;
import io.vertx.core.json.JsonObject;
import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;

/**
 * Created by developer on 29.04.16.
 */
public class MyWSServerVerticle extends AbstractVerticle {
    //private JsonObject

    @Override
    public void start(Future<Void> startFuture) throws Exception {
        Logger log = LoggerFactory.getLogger(Main.class);
        long main_pid = Thread.currentThread().getId();
        Handler<ServerWebSocket> wsHandler = serverWebSocket -> {
            if(!serverWebSocket.path().equalsIgnoreCase("/ws")){
                log.info("rejected");
                serverWebSocket.reject();
            } else {
                log.info("accepted...");
                long socket_pid = Thread.currentThread().getId();
                String socketId = serverWebSocket.textHandlerID();
                JsonObject msg = new JsonObject().put("socketId", socketId).put("connected",true);
                vertx.eventBus().send(Channels.CONNECTION_CHANNEL, msg);
                serverWebSocket.handler(buffer -> {
                    String str = buffer.getString(0, buffer.length());
                    long handler_pid = Thread.currentThread().getId();
                    log.info("Got ws msg: " + str);
                    String res = String.format("[%s](req:%s)main:%d sock:%d handlr:%d", serverWebSocket.textHandlerID(), str, main_pid, socket_pid, handler_pid);
                    serverWebSocket.writeFinalTextFrame(res);
                    vertx.eventBus().send(Channels.USER_SAID_CHANNEL, String.format("[%s]: %s", serverWebSocket.textHandlerID(), str));
                });
                serverWebSocket.closeHandler(event -> {
                    JsonObject msg2 = new JsonObject().put("socketId", serverWebSocket.textHandlerID()).put("connected", false);
                    vertx.eventBus().send(Channels.CONNECTION_CHANNEL, msg2);
                });
            }
        };
        vertx
                .createHttpServer(new HttpServerOptions().setUsePooledBuffers(true))
                .websocketHandler(wsHandler)
                .listen(8080);
    }

}