Optimize protobuf and netty integration.

Signed-off-by: Yury-Fridlyand <[email protected]>

java/client/src/main/java/babushka/connectors/handlers/ChannelBuilder.java

@@ -2,11 +2,15 @@
 
 import babushka.managers.CallbackManager;
 import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
 import io.netty.channel.unix.UnixChannel;
+import io.netty.handler.codec.protobuf.ProtobufDecoder;
+import io.netty.handler.codec.protobuf.ProtobufEncoder;
 import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
 import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;
 import lombok.NonNull;
 import lombok.RequiredArgsConstructor;
+import response.ResponseOuterClass.Response;
 
 /** Builder for the channel used by {@link babushka.connectors.SocketConnection}. */
 @RequiredArgsConstructor
@@ -18,9 +22,11 @@ public class ChannelBuilder extends ChannelInitializer<UnixChannel> {
   public void initChannel(@NonNull UnixChannel ch) {
     ch.pipeline()
         // https://netty.io/4.1/api/io/netty/handler/codec/protobuf/ProtobufEncoder.html
-        .addLast("protobufDecoder", new ProtobufVarint32FrameDecoder())
-        .addLast("protobufEncoder", new ProtobufVarint32LengthFieldPrepender())
+        .addLast("frameDecoder", new ProtobufVarint32FrameDecoder())
+        .addLast("frameEncoder", new ProtobufVarint32LengthFieldPrepender())
+        .addLast("protobufDecoder", new ProtobufDecoder(Response.getDefaultInstance()))
+        .addLast("protobufEncoder", new ProtobufEncoder())
         .addLast(new ReadHandler(callbackManager))
-        .addLast(new WriteHandler());
+        .addLast(new ChannelOutboundHandlerAdapter());
   }
 }

java/client/src/main/java/babushka/connectors/handlers/ChannelHandler.java

@@ -23,16 +23,16 @@ public CompletableFuture<Response> write(RedisRequest.Builder request, boolean f
     request.setCallbackIdx(commandId.getKey());
 
     if (flush) {
-      channel.writeAndFlush(request.build().toByteArray());
+      channel.writeAndFlush(request.build());
     } else {
-      channel.write(request.build().toByteArray());
+      channel.write(request.build());
     }
     return commandId.getValue();
   }
 
   /** Write a protobuf message to the socket. */
   public CompletableFuture<Response> connect(ConnectionRequest request) {
-    channel.writeAndFlush(request.toByteArray());
+    channel.writeAndFlush(request);
     return callbackManager.getConnectionPromise();
   }
 

java/client/src/main/java/babushka/connectors/handlers/ReadHandler.java

@@ -1,38 +1,22 @@
 package babushka.connectors.handlers;
 
 import babushka.managers.CallbackManager;
-import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import lombok.NonNull;
 import lombok.RequiredArgsConstructor;
-import response.ResponseOuterClass;
+import response.ResponseOuterClass.Response;
 
 /** Handler for inbound traffic though UDS. Used by Netty. */
 @RequiredArgsConstructor
 public class ReadHandler extends ChannelInboundHandlerAdapter {
 
   private final CallbackManager callbackManager;
 
-  /**
-   * Handles responses from babushka core:
-   *
-   * <ol>
-   *   <li>Copy to a buffer;
-   *   <li>Parse protobuf packet;
-   *   <li>Find and resolve a corresponding future;
-   * </ol>
-   */
+  /** Submit responses from babushka to an instance {@link CallbackManager} to handle them. */
   @Override
-  public void channelRead(@NonNull ChannelHandlerContext ctx, @NonNull Object msg)
-      throws Exception {
-    var buf = (ByteBuf) msg;
-    var bytes = new byte[buf.readableBytes()];
-    buf.readBytes(bytes);
-    // TODO surround parsing with try-catch, set error to future if parsing failed.
-    var response = ResponseOuterClass.Response.parseFrom(bytes);
-    callbackManager.completeRequest(response);
-    buf.release();
+  public void channelRead(@NonNull ChannelHandlerContext ctx, @NonNull Object msg) {
+    callbackManager.completeRequest((Response) msg);
   }
 
   /** Handles uncaught exceptions from {@link #channelRead(ChannelHandlerContext, Object)}. */