Use Consumer<JsonObject> instead of Consumer<Buffer> to write

Author: lucifer4j

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

@@ -8,6 +8,7 @@
 import io.vertx.core.http.HttpHeaders;
 import io.vertx.core.http.HttpServerRequest;
 import io.vertx.core.http.HttpServerResponse;
+import io.vertx.core.json.Json;
 import io.vertx.core.json.JsonObject;
 import io.vertx.ext.bridge.BridgeEventType;
 import io.vertx.ext.eventbus.bridge.tcp.BridgeEvent;
@@ -62,16 +63,15 @@ public void handle(HttpServerRequest socket) {
               checkCallHook(() -> new BridgeEventImpl<>(BridgeEventType.SOCKET_CLOSED, null, socket));
               registry.clear();
             });
-          Consumer<Buffer> writer;
+          Consumer<JsonObject> writer;
           if (method.equals("register")) {
             response.setChunked(true);
-            writer = body -> {
-              JsonObject object = body.toJsonObject();
-              response.write("event: " + object.getJsonObject("result").getString("address") + "\n");
-              response.write("data: " + object.encode() + "\n\n");
+            writer = payload -> {
+              response.write("event: " + payload.getJsonObject("result").getString("address") + "\n");
+              response.write("data: " + payload.encode() + "\n\n");
             };
           } else {
-            writer = response::end;
+            writer = payload -> response.end(payload.encode());
           }
           dispatch(writer, method, id, msg, registry, replies);
         });

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

@@ -96,7 +96,7 @@ protected boolean isInvalid(JsonObject object) {
     return false;
   }
 
-  protected void dispatch(Consumer<Buffer> socket, String method, Object id, JsonObject msg, Map<String, MessageConsumer<?>> registry, Map<String, Message<JsonObject>> replies) {
+  protected void dispatch(Consumer<JsonObject> socket, String method, Object id, JsonObject msg, Map<String, MessageConsumer<?>> registry, Map<String, Message<JsonObject>> replies) {
     switch (method) {
       case "send":
         checkCallHook(
@@ -135,7 +135,7 @@ protected void dispatch(Consumer<Buffer> socket, String method, Object id, JsonO
     }
   }
 
-  protected void unregister(Consumer<Buffer> socket, Object id, JsonObject msg, Map<String, MessageConsumer<?>> registry, Map<String, Message<JsonObject>> replies) {
+  protected void unregister(Consumer<JsonObject> socket, Object id, JsonObject msg, Map<String, MessageConsumer<?>> registry, Map<String, Message<JsonObject>> replies) {
     final JsonObject params = msg.getJsonObject("params", EMPTY);
     final String address = params.getString("address");
 
@@ -160,7 +160,7 @@ protected void unregister(Consumer<Buffer> socket, Object id, JsonObject msg, Ma
     }
   }
 
-  protected void register(Consumer<Buffer> socket, Object id, JsonObject msg, Map<String, MessageConsumer<?>> registry, Map<String, Message<JsonObject>> replies) {
+  protected void register(Consumer<JsonObject> socket, Object id, JsonObject msg, Map<String, MessageConsumer<?>> registry, Map<String, Message<JsonObject>> replies) {
     final JsonObject params = msg.getJsonObject("params", EMPTY);
     final String address = params.getString("address");
 
@@ -205,7 +205,7 @@ protected void register(Consumer<Buffer> socket, Object id, JsonObject msg, Map<
     }
   }
 
-  protected void publish(Consumer<Buffer> socket, Object id, JsonObject msg, Map<String, MessageConsumer<?>> registry, Map<String, Message<JsonObject>> replies) {
+  protected void publish(Consumer<JsonObject> socket, Object id, JsonObject msg, Map<String, MessageConsumer<?>> registry, Map<String, Message<JsonObject>> replies) {
     final JsonObject params = msg.getJsonObject("params", EMPTY);
     final String address = params.getString("address");
 
@@ -228,7 +228,7 @@ protected void publish(Consumer<Buffer> socket, Object id, JsonObject msg, Map<S
     }
   }
 
-  protected void send(Consumer<Buffer> socket, Object id, JsonObject msg, Map<String, MessageConsumer<?>> registry, Map<String, Message<JsonObject>> replies) {
+  protected void send(Consumer<JsonObject> socket, Object id, JsonObject msg, Map<String, MessageConsumer<?>> registry, Map<String, Message<JsonObject>> replies) {
     final JsonObject params = msg.getJsonObject("params", EMPTY);
     final String address = params.getString("address");
 

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

@@ -13,6 +13,7 @@
 
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Consumer;
 
 public class TCPJsonRPCStreamEventBusBridgeImpl extends JsonRPCStreamEventBusBridgeImpl<NetSocket> {
 
@@ -59,8 +60,10 @@ public void handle(NetSocket socket) {
                 final String method = msg.getString("method");
                 final Object id = msg.getValue("id");
 
+                Consumer<JsonObject> writer = payload -> socket.write(payload.toBuffer().appendString("\r\n"));
+
                 dispatch(
-                  socket::write,
+                  writer,
                   method,
                   id,
                   msg,

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

@@ -32,11 +32,11 @@ public void handle(WebSocketBase socket) {
         final Map<String, MessageConsumer<?>> registry = new ConcurrentHashMap<>();
         final Map<String, Message<JsonObject>> replies = new ConcurrentHashMap<>();
 
-        Consumer<Buffer> consumer;
+        Consumer<JsonObject> consumer;
         if (options.getWebsocketsTextAsFrame()) {
-          consumer = buffer -> socket.writeTextMessage(buffer.toString());
+          consumer = payload -> socket.writeTextMessage(payload.encode());
         } else {
-          consumer = socket::writeBinaryMessage;
+          consumer = payload -> socket.writeBinaryMessage(payload.toBuffer());
         }
 
         socket

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

@@ -18,6 +18,7 @@
 import io.vertx.core.MultiMap;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.eventbus.ReplyException;
+import io.vertx.core.json.Json;
 import io.vertx.core.json.JsonObject;
 
 import java.util.function.Consumer;
@@ -83,16 +84,16 @@ public static void request(String method, JsonObject params, Consumer<Buffer> ha
     request(method, null, params, null, handler);
   }
 
-  public static void response(Object id, Object result, Consumer<Buffer> handler) {
+  public static void response(Object id, Object result, Consumer<JsonObject> handler) {
     final JsonObject payload = new JsonObject()
       .put("jsonrpc", "2.0")
       .put("id", id)
       .put("result", result);
 
-    handler.accept(payload.toBuffer().appendString("\r\n"));
+    handler.accept(payload);
   }
 
-  public static void error(Object id, Number code, String message, Consumer<Buffer> handler) {
+  public static void error(Object id, Number code, String message, Consumer<JsonObject> handler) {
     final JsonObject payload = new JsonObject()
       .put("jsonrpc", "2.0")
       .put("id", id);
@@ -108,14 +109,14 @@ public static void error(Object id, Number code, String message, Consumer<Buffer
       error.put("message", message);
     }
 
-    handler.accept(payload.toBuffer().appendString("\r\n"));
+    handler.accept(payload);
   }
 
-  public static void error(Object id, ReplyException failure, Consumer<Buffer> handler) {
+  public static void error(Object id, ReplyException failure, Consumer<JsonObject> handler) {
     error(id, failure.failureCode(), failure.getMessage(), handler);
   }
 
-  public static void error(Object id, String message, Consumer<Buffer> handler) {
+  public static void error(Object id, String message, Consumer<JsonObject> handler) {
     error(id, -32000, message, handler);
   }
 }