Avoid buffer copies in RedisStateMachine

src/main/java/io/lettuce/core/protocol/CommandHandler.java

@@ -213,7 +213,7 @@ public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
         setState(LifecycleState.REGISTERED);
 
         buffer = ctx.alloc().buffer(8192 * 8);
-        rsm = new RedisStateMachine(ctx.alloc());
+        rsm = new RedisStateMachine();
         ctx.fireChannelRegistered();
     }
 

src/main/java/io/lettuce/core/protocol/RedisStateMachine.java

@@ -231,8 +231,6 @@ public String toString() {
 
     private final boolean debugEnabled = logger.isDebugEnabled();
 
-    private final ByteBuf responseElementBuffer;
-
     private final AtomicBoolean closed = new AtomicBoolean();
 
     private final Resp2LongProcessor longProcessor = new Resp2LongProcessor();
@@ -244,8 +242,8 @@ public String toString() {
     /**
      * Initialize a new instance.
      */
-    public RedisStateMachine(ByteBufAllocator alloc) {
-        this.responseElementBuffer = alloc.buffer(1024);
+    public RedisStateMachine() {
+
     }
 
     public boolean isDiscoverProtocol() {
@@ -591,7 +589,7 @@ public void reset() {
      */
     public void close() {
         if (closed.compareAndSet(false, true)) {
-            responseElementBuffer.release();
+            // free resources if any
         }
     }
 
@@ -673,18 +671,11 @@ private ByteBuffer readBytes(ByteBuf buffer, int count) {
     }
 
     private ByteBuffer readBytes0(ByteBuf buffer, int count) {
+        final ByteBuffer byteBuffer = buffer.internalNioBuffer(buffer.readerIndex(), count);
 
-        ByteBuffer bytes;
-        responseElementBuffer.clear();
-
-        if (responseElementBuffer.capacity() < count) {
-            responseElementBuffer.capacity(count);
-        }
-
-        buffer.readBytes(responseElementBuffer, count);
-        bytes = responseElementBuffer.internalNioBuffer(0, count);
-
-        return bytes;
+        // advance reader index
+        buffer.readerIndex(buffer.readerIndex() + count);
+        return byteBuffer;
     }
 
     /**

src/test/java/io/lettuce/core/protocol/RedisStateMachineResp2UnitTests.java

@@ -74,7 +74,7 @@ static void afterClass() {
     @BeforeEach
     final void createStateMachine() {
         output = new StatusOutput<>(codec);
-        rsm = new RedisStateMachine(ByteBufAllocator.DEFAULT);
+        rsm = new RedisStateMachine();
     }
 
     @AfterEach

src/test/java/io/lettuce/core/protocol/RedisStateMachineResp3UnitTests.java

@@ -74,7 +74,7 @@ static void afterClass() {
     @BeforeEach
     final void createStateMachine() {
         output = new StatusOutput<>(codec);
-        rsm = new RedisStateMachine(ByteBufAllocator.DEFAULT);
+        rsm = new RedisStateMachine();
         rsm.setProtocolVersion(ProtocolVersion.RESP3);
     }
 

src/test/java/io/lettuce/core/protocol/StateMachineUnitTests.java

@@ -67,7 +67,7 @@ static void afterClass() {
     @BeforeEach
     final void createStateMachine() {
         output = new StatusOutput<>(codec);
-        rsm = new RedisStateMachine(ByteBufAllocator.DEFAULT);
+        rsm = new RedisStateMachine();
     }
 
     @AfterEach

src/test/jmh/io/lettuce/core/protocol/RedisStateMachineBenchmark.java

@@ -60,7 +60,7 @@ public void set(long integer) {
 
     private ByteBuf masterBuffer;
 
-    private final RedisStateMachine stateMachine = new RedisStateMachine(ByteBufAllocator.DEFAULT);
+    private final RedisStateMachine stateMachine = new RedisStateMachine();
     private final byte[] payload = ("*3\r\n" + //
             "$4\r\n" + //
             "LLEN\r\n" + //