Add more transports, schema validation and refactorings

pom.xml

@@ -24,7 +24,7 @@
   </parent>
 
   <artifactId>vertx-tcp-eventbus-bridge</artifactId>
-  <version>4.3.2-SNAPSHOT</version>
+  <version>999-SNAPSHOT</version>
 
   <name>Vert.x TCP EventBus Bridge</name>
 
@@ -54,7 +54,11 @@
     <dependency>
       <groupId>io.vertx</groupId>
       <artifactId>vertx-bridge-common</artifactId>
-      <version>${project.version}</version>
+    </dependency>
+
+    <dependency>
+     <groupId>io.vertx</groupId>
+     <artifactId>vertx-json-schema</artifactId>
     </dependency>
 
     <dependency>
@@ -73,6 +77,11 @@
       <type>test-jar</type>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>io.vertx</groupId>
+      <artifactId>vertx-web-client</artifactId>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>

src/client/nodejs/package.json

@@ -11,4 +11,4 @@
   "scripts": {
     "test": "mocha ./test/index.js"
   }
-}
+}

src/main/java/examples/HttpBridgeExample.java

@@ -0,0 +1,60 @@
+package examples;
+
+import io.vertx.core.AbstractVerticle;
+import io.vertx.core.Handler;
+import io.vertx.core.Promise;
+import io.vertx.core.Vertx;
+import io.vertx.core.eventbus.Message;
+import io.vertx.core.http.HttpHeaders;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.http.HttpServerResponse;
+import io.vertx.core.http.WebSocketBase;
+import io.vertx.core.json.JsonObject;
+import io.vertx.ext.bridge.PermittedOptions;
+import io.vertx.ext.eventbus.bridge.tcp.JsonRPCBridgeOptions;
+import io.vertx.ext.eventbus.bridge.tcp.JsonRPCStreamEventBusBridge;
+
+public class HttpBridgeExample extends AbstractVerticle {
+
+  public static void main(String[] args) {
+    Vertx.vertx().deployVerticle(new HttpBridgeExample());
+  }
+
+  @Override
+  public void start(Promise<Void> start) {
+    vertx.eventBus().consumer("hello", (Message<JsonObject> msg) -> msg.reply(new JsonObject().put("value", "Hello " + msg.body().getString("value"))));
+    vertx.eventBus().consumer("echo", (Message<JsonObject> msg) -> msg.reply(msg.body()));
+    vertx.setPeriodic(1000L, __ -> vertx.eventBus().send("ping", new JsonObject().put("value", "hi")));
+
+    JsonRPCBridgeOptions options = new JsonRPCBridgeOptions()
+      .addInboundPermitted(new PermittedOptions().setAddress("hello"))
+      .addInboundPermitted(new PermittedOptions().setAddress("echo"))
+      .addInboundPermitted(new PermittedOptions().setAddress("ping"))
+      .addOutboundPermitted(new PermittedOptions().setAddress("echo"))
+      .addOutboundPermitted(new PermittedOptions().setAddress("hello"))
+      .addOutboundPermitted(new PermittedOptions().setAddress("ping"));
+
+    Handler<HttpServerRequest> bridge = JsonRPCStreamEventBusBridge.httpSocketHandler(vertx, options, null);
+
+    vertx
+      .createHttpServer()
+      .requestHandler(req -> {
+        if ("/".equals(req.path())) {
+          req.response()
+            .putHeader(HttpHeaders.CONTENT_TYPE, "text/html")
+            .sendFile("http.html");
+        } else if ("/jsonrpc".equals(req.path())){
+          bridge.handle(req);
+        } else {
+          req.response().setStatusCode(404).end("Not Found");
+        }
+      })
+      .listen(8080)
+      .onFailure(start::fail)
+      .onSuccess(server -> {
+        System.out.println("Server listening at http://localhost:8080");
+        start.complete();
+      });
+
+  }
+}

src/main/java/examples/HttpSSEBridgeExample.java

@@ -0,0 +1,66 @@
+package examples;
+
+import io.vertx.core.AbstractVerticle;
+import io.vertx.core.Promise;
+import io.vertx.core.Vertx;
+import io.vertx.core.eventbus.Message;
+import io.vertx.core.http.HttpHeaders;
+import io.vertx.core.http.HttpServerResponse;
+import io.vertx.core.json.JsonObject;
+import io.vertx.ext.bridge.PermittedOptions;
+import io.vertx.ext.eventbus.bridge.tcp.JsonRPCBridgeOptions;
+import io.vertx.ext.eventbus.bridge.tcp.impl.HttpJsonRPCStreamEventBusBridgeImpl;
+
+import java.util.function.Consumer;
+
+public class HttpSSEBridgeExample extends AbstractVerticle {
+
+  public static void main(String[] args) {
+    Vertx.vertx().deployVerticle(new HttpSSEBridgeExample());
+  }
+
+  @Override
+  public void start(Promise<Void> start) {
+    // just to have some messages flowing around
+    vertx.eventBus().consumer("hello", (Message<JsonObject> msg) -> msg.reply(new JsonObject().put("value", "Hello " + msg.body().getString("value"))));
+    vertx.eventBus().consumer("echo", (Message<JsonObject> msg) -> msg.reply(msg.body()));
+    vertx.setPeriodic(1000L, __ -> vertx.eventBus().send("ping", new JsonObject().put("value", "hi")));
+
+    // use explicit class because SSE method is not on the interface currently
+    HttpJsonRPCStreamEventBusBridgeImpl bridge = new HttpJsonRPCStreamEventBusBridgeImpl(
+        vertx,
+        new JsonRPCBridgeOptions()
+          .addInboundPermitted(new PermittedOptions().setAddress("hello"))
+          .addInboundPermitted(new PermittedOptions().setAddress("echo"))
+          .addInboundPermitted(new PermittedOptions().setAddress("test"))
+          .addOutboundPermitted(new PermittedOptions().setAddress("echo"))
+          .addOutboundPermitted(new PermittedOptions().setAddress("test"))
+          .addOutboundPermitted(new PermittedOptions().setAddress("ping")),
+        event -> event.complete(true)
+    );
+
+    vertx
+      .createHttpServer()
+      .requestHandler(req -> {
+        // this is where any http request will land
+        // serve the base HTML application
+        if ("/".equals(req.path())) {
+          req.response()
+            .putHeader(HttpHeaders.CONTENT_TYPE, "text/html")
+            .sendFile("sse.html");
+        } else if ("/jsonrpc".equals(req.path())) {
+          bridge.handle(req);
+        }  else if ("/jsonrpc-sse".equals(req.path())) {
+          bridge.handleSSE(req, (int) (Math.random() * 100_000), new JsonObject().put("address", "ping"));
+        } else {
+          req.response().setStatusCode(404).end("Not Found");
+        }
+      })
+      .listen(8080)
+      .onFailure(start::fail)
+      .onSuccess(server -> {
+        System.out.println("Server listening at http://localhost:8080");
+        start.complete();
+      });
+  }
+}

src/main/java/examples/WebsocketBridgeExample.java

@@ -0,0 +1,70 @@
+package examples;
+
+import io.vertx.core.AbstractVerticle;
+import io.vertx.core.Handler;
+import io.vertx.core.Promise;
+import io.vertx.core.Vertx;
+import io.vertx.core.eventbus.Message;
+import io.vertx.core.http.HttpHeaders;
+import io.vertx.core.http.WebSocketBase;
+import io.vertx.core.json.JsonObject;
+import io.vertx.ext.bridge.PermittedOptions;
+import io.vertx.ext.eventbus.bridge.tcp.JsonRPCBridgeOptions;
+import io.vertx.ext.eventbus.bridge.tcp.JsonRPCStreamEventBusBridge;
+
+public class WebsocketBridgeExample extends AbstractVerticle {
+
+  public static void main(String[] args) {
+    Vertx.vertx().deployVerticle(new WebsocketBridgeExample());
+  }
+
+  @Override
+  public void start(Promise<Void> start) {
+    vertx.eventBus().consumer("hello", (Message<JsonObject> msg) -> msg.reply(new JsonObject().put("value", "Hello " + msg.body().getString("value"))));
+    vertx.eventBus().consumer("echo", (Message<JsonObject> msg) -> msg.reply(msg.body()));
+    vertx.setPeriodic(1000L, __ -> vertx.eventBus().send("ping", new JsonObject().put("value", "hi")));
+
+    JsonRPCBridgeOptions options = new JsonRPCBridgeOptions()
+      .addInboundPermitted(new PermittedOptions().setAddress("hello"))
+      .addInboundPermitted(new PermittedOptions().setAddress("echo"))
+      .addInboundPermitted(new PermittedOptions().setAddress("ping"))
+      .addOutboundPermitted(new PermittedOptions().setAddress("echo"))
+      .addOutboundPermitted(new PermittedOptions().setAddress("hello"))
+      .addOutboundPermitted(new PermittedOptions().setAddress("ping"))
+      // if set to false, then websockets messages are received on frontend as binary frames
+      .setWebsocketsTextAsFrame(true);
+
+    Handler<WebSocketBase> bridge = JsonRPCStreamEventBusBridge.webSocketHandler(vertx, options, null);
+
+    vertx
+      .createHttpServer()
+      .requestHandler(req -> {
+        // this is where any http request will land
+        if ("/jsonrpc".equals(req.path())) {
+          // we switch from HTTP to WebSocket
+          req.toWebSocket()
+            .onFailure(err -> {
+              err.printStackTrace();
+              req.response().setStatusCode(500).end(err.getMessage());
+            })
+            .onSuccess(bridge::handle);
+        } else {
+          // serve the base HTML application
+          if ("/".equals(req.path())) {
+            req.response()
+              .putHeader(HttpHeaders.CONTENT_TYPE, "text/html")
+              .sendFile("ws.html");
+          } else {
+            // 404 all the rest
+            req.response().setStatusCode(404).end("Not Found");
+          }
+        }
+      })
+      .listen(8080)
+      .onFailure(start::fail)
+      .onSuccess(server -> {
+        System.out.println("Server listening at http://localhost:8080");
+        start.complete();
+      });
+  }
+}

src/main/java/io/vertx/ext/eventbus/bridge/tcp/BridgeEvent.java

@@ -22,6 +22,7 @@
 import io.vertx.core.Future;
 import io.vertx.core.json.JsonObject;
 import io.vertx.core.net.NetSocket;
+import io.vertx.ext.bridge.BaseBridgeEvent;
 
 /**
  * Represents an event that occurs on the event bus bridge.
@@ -31,7 +32,7 @@
  * @author <a href="http://tfox.org">Tim Fox</a>
  */
 @VertxGen
-public interface BridgeEvent extends io.vertx.ext.bridge.BaseBridgeEvent {
+public interface BridgeEvent<T> extends BaseBridgeEvent {
 
   /**
    * Get the raw JSON message for the event. This will be null for SOCKET_CREATED or SOCKET_CLOSED events as there is
@@ -41,13 +42,16 @@ public interface BridgeEvent extends io.vertx.ext.bridge.BaseBridgeEvent {
    * @return this reference, so it can be used fluently
    */
   @Fluent
-  BridgeEvent setRawMessage(JsonObject message);
+  BridgeEvent<T> setRawMessage(JsonObject message);
+
+  // TODO: this will cause problems with WebSockets as they don't share any common base interface
+  //       this will be a breaking change to users, as the return type is now generic, is this OK?
 
   /**
    * Get the SockJSSocket instance corresponding to the event
    *
    * @return  the SockJSSocket instance
    */
   @CacheReturn
-  NetSocket socket();
+  T socket();
 }