Merge pull request #190 from fanson/perf/batch-flush

perf: batch flush with write() and channelReadComplete()

Author: tonivade

src/main/java/com/github/tonivade/resp/RespConnectionHandler.java

@@ -41,6 +41,11 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
     }
   }
 
+  @Override
+  public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
+    ctx.flush();
+  }
+
   @Override
   public void channelInactive(ChannelHandlerContext ctx) throws Exception {
     LOGGER.debug("channel inactive");

src/main/java/com/github/tonivade/resp/RespServerContext.java

@@ -148,6 +148,7 @@ protected <T> Observable<T> enqueue(Observable<T> observable) {
 
   private void processResponse(Request request, RedisToken token) {
     request.getSession().publish(token);
+    request.getSession().flush();
     if (request.isExit()) {
       request.getSession().close();
     }

src/main/java/com/github/tonivade/resp/command/DefaultSession.java

@@ -36,7 +36,12 @@ public String getId() {
 
   @Override
   public void publish(RedisToken msg) {
-    ctx.writeAndFlush(msg);
+    ctx.write(msg);
+  }
+
+  @Override
+  public void flush() {
+    ctx.flush();
   }
 
   @Override

src/main/java/com/github/tonivade/resp/command/Session.java

@@ -11,6 +11,7 @@
 public interface Session {
   String getId();
   void publish(RedisToken msg);
+  default void flush() {}
   void close();
   void destroy();
   <T> Optional<T> getValue(String key);