fix demos

Author: lucifer4j

src/main/java/examples/HttpSSEBridgeExample.java

@@ -1,21 +1,14 @@
 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.json.JsonObject;
 import io.vertx.ext.bridge.PermittedOptions;
 import io.vertx.ext.eventbus.bridge.tcp.JsonRPCBridgeOptions;
-import io.vertx.ext.eventbus.bridge.tcp.JsonRPCStreamEventBusBridge;
-import io.vertx.ext.web.client.WebClient;
-import io.vertx.ext.web.codec.BodyCodec;
-
-import static io.vertx.ext.eventbus.bridge.tcp.impl.protocol.JsonRPCHelper.request;
+import io.vertx.ext.eventbus.bridge.tcp.impl.HttpJsonRPCStreamEventBusBridgeImpl;
 
 public class HttpSSEBridgeExample extends AbstractVerticle {
 
@@ -30,8 +23,8 @@ public void start(Promise<Void> start) {
     vertx.eventBus().consumer("echo", (Message<JsonObject> msg) -> msg.reply(msg.body()));
     vertx.setPeriodic(1000L, __ -> vertx.eventBus().send("ping", new JsonObject().put("value", "hi")));
 
-    // once we fix the interface we can avoid the casts
-    Handler<HttpServerRequest> bridge = JsonRPCStreamEventBusBridge.httpSocketHandler(
+    // use explicit class because SSE method is not on the interface currently
+    HttpJsonRPCStreamEventBusBridgeImpl bridge = new HttpJsonRPCStreamEventBusBridgeImpl(
         vertx,
         new JsonRPCBridgeOptions()
           .addInboundPermitted(new PermittedOptions().setAddress("hello"))
@@ -43,8 +36,6 @@ public void start(Promise<Void> start) {
       null
     );
 
-    WebClient client = WebClient.create(vertx);
-
     vertx
       .createHttpServer()
       .requestHandler(req -> {
@@ -54,19 +45,15 @@ public void start(Promise<Void> start) {
           req.response()
             .putHeader(HttpHeaders.CONTENT_TYPE, "text/html")
             .sendFile("sse.html");
-        } else if ("/jsonrpc".equals(req.path())){
+        } else if ("/jsonrpc".equals(req.path())) {
           bridge.handle(req);
-        } else if ("/jsonrpc-sse".equals(req.path())) {
-          HttpServerResponse resp = req.response().setChunked(true).putHeader("Content-Type", "text/event-stream");
-          request(
-            "register",
-            (int) (Math.random() * 100_000),
-            new JsonObject().put("address", "ping"),
-            buffer -> client
-              .post(8080, "localhost", "/jsonrpc")
-              .as(BodyCodec.pipe(resp))
-              .sendBuffer(buffer)
-          );
+        }  else if ("/jsonrpc-sse".equals(req.path())) {
+          JsonObject message = new JsonObject()
+            .put("jsonrpc", "2.0")
+            .put("method", "register")
+            .put("id", (int) (Math.random() * 100_000))
+            .put("params", new JsonObject().put("address", "ping"));
+          bridge.handleSSE(req.response(), message);
         } else {
           req.response().setStatusCode(404).end("Not Found");
         }

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

@@ -32,16 +32,6 @@
  *
  * @author <a href="mailto:plopes@redhat.com">Paulo Lopes</a>
  */
-
-// TODO: "extends Handler<NetSocket>" was a bad idea because it locks the implementation to TCP sockets. Instead we
-//       should have explicit methods that either handle a NetSocket or a WebSocketBase:
-//       handle(NetSocket socket)  handle(WebSocketBase socket)
-//       or: return a handler, e.g.:
-//         Handler<WebSocketBase> webSocketHandler();
-//         Handler<NetSocket> netSocketHandler();
-
-// How about generifying this interface as in JsonRPCStreamEventBusBridge<T> extends Handler<T> ?
-// similarly create a base class for the impl and concrete impls for each protocol.
 @VertxGen
 public interface JsonRPCStreamEventBusBridge {
 

src/main/java/io/vertx/ext/eventbus/bridge/tcp/impl/HttpJsonRPCStreamEventBusBridgeImpl.java

@@ -46,7 +46,6 @@ public void handle(HttpServerRequest socket) {
 
           // TODO: body may be an array (batching)
           final JsonObject msg = new JsonObject(buffer);
-          System.out.println(msg);
 
           if (this.isInvalid(msg)) {
             return;
@@ -79,4 +78,44 @@ public void handle(HttpServerRequest socket) {
       // on failure
       () ->  socket.response().setStatusCode(500).setStatusMessage("Internal Server Error").end());
   }
+
+  // TODO: discuss implications of accepting response here. bridge events may not be emitted.
+  //  but if accepting request cannot use handler as the request is usually empty and handler is
+  //  not invoked until data has been read. also same thing for other cases
+  public void handleSSE(HttpServerResponse socket, JsonObject msg) {
+    final Map<String, MessageConsumer<?>> registry = new ConcurrentHashMap<>();
+
+    socket.exceptionHandler(t -> {
+      log.error(t.getMessage(), t);
+      registry.values().forEach(MessageConsumer::unregister);
+      registry.clear();
+    });
+    if (this.isInvalid(msg)) {
+      return;
+    }
+
+    HttpServerResponse response = socket
+      .setChunked(true)
+      .putHeader(HttpHeaders.CONTENT_TYPE, "text/event-stream")
+      .endHandler(handler -> {
+        registry.values().forEach(MessageConsumer::unregister);
+        registry.clear();
+      });
+
+    final String method = msg.getString("method");
+    if (!method.equalsIgnoreCase("register")) {
+      log.error("Invalid method for SSE!");
+      return;
+    }
+
+    final Object id = msg.getValue("id");
+    Consumer<JsonObject> writer = payload -> {
+      // TODO: Should we use id or address for event name?
+      response.write("event: " + payload.getJsonObject("result").getString("address") + "\n");
+      response.write("data: " + payload.encode() + "\n\n");
+    };
+    register(writer, id, msg, registry, replies);
+  }
+
+
 }

src/test/resources/http.html src/main/resources/http.htmlsrc/test/resources/http.html renamed to src/main/resources/http.html

@@ -12,7 +12,7 @@
       });
     }
   </script>
-  <title>Websockets Bridge Example</title>
+  <title>HTTP Bridge Example</title>
 </head>
 <body>
  <input type="text" id="payload" value='{"jsonrpc":"2.0","method":"send","id":"2","params":{"address":"echo","body":"Hello World!"}}'> <input type="button" onclick="sendMsg()" value="Send message"/>

src/test/resources/sse.html src/main/resources/sse.htmlsrc/test/resources/sse.html renamed to src/main/resources/sse.html

@@ -8,18 +8,11 @@
 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.core.net.NetSocket;
-import io.vertx.ext.bridge.BridgeOptions;
 import io.vertx.ext.bridge.PermittedOptions;
-import io.vertx.ext.eventbus.bridge.tcp.impl.JsonRPCStreamEventBusBridgeImpl;
-import io.vertx.ext.eventbus.bridge.tcp.impl.TCPJsonRPCStreamEventBusBridgeImpl;
 import io.vertx.ext.web.client.WebClient;
 import io.vertx.ext.web.codec.BodyCodec;
 
-import java.util.Random;
-
 import static io.vertx.ext.eventbus.bridge.tcp.impl.protocol.JsonRPCHelper.request;
 
 public class InteropWebSocketServer extends AbstractVerticle {