Vertx EventBusBridge server and client in Groovy

groovy_event_bus_bridge

// -- server.groovy --

import io.vertx.ext.web.handler.BodyHandler
import io.vertx.ext.web.handler.sockjs.BridgeEventType
import io.vertx.groovy.core.Vertx
import io.vertx.groovy.ext.web.Router
import io.vertx.groovy.ext.web.handler.sockjs.SockJSHandler


def vertx = Vertx.vertx();

def router = Router.router(vertx)


def options = [
        inboundPermitteds:[
                [
                        addressRegex:".+"
                ]
        ],
        outboundPermitteds:[
                [
                        addressRegex:".+"
                ]
        ]
]


router.route("/eventbus/*").handler(SockJSHandler.create(vertx).bridge(options, { event ->

    if (event.type() == BridgeEventType.SOCKET_CREATED) {
        println("A socket was created")
    }
    if (event.type() == BridgeEventType.SOCKET_CLOSED) {
        println("A socket was closed")
    }
    if (event.type() == BridgeEventType.SEND) {
        println("the client sent a message to the server")
    }
    if (event.type() == BridgeEventType.PUBLISH) {
        println("the client published a message to the server")
    }
    if (event.type() == BridgeEventType.RECEIVE) {
        println("the server sent a message to the client")
    }
    if (event.type() == BridgeEventType.REGISTER) {
        println("a new client registered with the server")
    }
    if (event.type() == BridgeEventType.UNREGISTER) {
        println("an existing client unregistered with the server")
    }

    // This signals that it's ok to process the event
    event.complete(true)

}))

// Serve the static resources
router.route().handler(BodyHandler.create())

vertx.createHttpServer().requestHandler(router.&accept).listen(8080)

// Publish a message to the address "news-feed" every second
vertx.setPeriodic(1000, { t ->
    vertx.eventBus().publish("someaddress", "news from the server!")
})

// todo: make a consumer of some address and send it from client
vertx.eventBus().consumer("someotheraddress", { msg ->
    println(msg.body())
})

// --------------------------------------------------------

// -- client.groovy --

import io.vertx.core.Handler
import io.vertx.core.Vertx
import io.vertx.core.buffer.Buffer
import io.vertx.core.http.HttpClient
import io.vertx.core.http.HttpClientOptions
import io.vertx.core.http.WebSocket
import io.vertx.core.json.JsonObject

def startTestClient() {

    //runnable
    int num_sockets = 500,
        count = 0;

    def options = new HttpClientOptions()
            .setMaxPoolSize(num_sockets + 1);
            //.setKeepAlive(true); // nec for pooling? research

    def client = Vertx.vertx().createHttpClient(options);

    for (int i = 0; i < num_sockets; i++) {
        try {
            addSocket(client);
        } catch (ConnectException e) {
            println(e.getMessage());
        }
        count++;
        println(count);
    }
}
protected void addSocket(HttpClient client) throws ConnectException{
    client.websocket(8080, "localhost", "/eventbus/websocket", new Handler<WebSocket>(){
        @Override
        public void handle(WebSocket websocket) {
            // Register
            JsonObject msg = new JsonObject().put("type", "register").put("address", "someaddress");
            websocket.writeFinalTextFrame(msg.encode());

            // Send
            msg = new JsonObject().put("type", "send").put("address", "someotheraddress").put("body", "hello world");
            websocket.writeFinalTextFrame(msg.encode());

            websocket.handler(new Handler<Buffer>() {
                @Override
                public void handle(Buffer buff) {
                    String message = buff.toString();
                    JsonObject received = new JsonObject(message);

                    System.out.println(received.getValue("body"));
                }
            });
        }
    });
}

startTestClient()