docs: @Override
      public ByteBuffer readBytes(int numBytes) throws IOException {
          final int appsz = applicationBufferSize();
          ByteBuf readbuf = m_ce.allocator().ioBuffer(packetBufferSize());
          CompositeByteBuf msgbb = Unpooled.compositeBuffer();
  
          try {
              ByteBuf clear = doUnwrap(readbuf, appsz);
  
              msgbb.addComponent(true, clear);
              int needed = numBytes;
              if (numBytes == NOT_AVAILABLE) {
                  needed = msgbb.readableBytes() >= 4 ? validateLength(msgbb.readInt()) : NOT_AVAILABLE;
              }
              while (msgbb.readableBytes() < (needed == NOT_AVAILABLE ? 4 : needed)) {
                  clear = doUnwrap(readbuf, appsz);
  
                  msgbb.addComponent(true, clear);
  
                  if (needed == NOT_AVAILABLE && msgbb.readableBytes() >= 4) {
                      needed = validateLength(msgbb.readInt());
                  }
              }
  
              ByteBuffer retbb = ByteBuffer.allocate(needed);
              msgbb.readBytes(retbb);
              msgbb.discardReadComponents();
  
              assert !msgbb.isReadable() : "read from unblocked channel that received multiple messages?";
  
              return (ByteBuffer)retbb.flip();
          } finally {
              readbuf.release();
              msgbb.release();
          }
      }