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

Change Session.publish() from writeAndFlush() to write(), and add
channelReadComplete() in RespConnectionHandler to flush once after
all messages in a read batch are processed. This reduces the number
of system calls when handling pipelined commands — instead of one
flush per message, a single flush covers the entire batch.

For the async serial execution path (default mode), responses are
written from the scheduler thread after channelReadComplete has
already fired, so processResponse() explicitly calls flush() after
each write to ensure delivery.

A default no-op flush() is added to the Session interface for
backward compatibility with custom implementations.

Made-with: Cursor

Author: haiyang.zhou

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

@@ -140,6 +140,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);