AbstractRedisCommand extends AbstractNettyRequestResponseCommand<T> implements LoggableRedisCommand<T> { @Override protected T doReceiveResponse(Channel channel, ByteBuf byteBuf) throws Exception { switch(commandResponseState){ case READING_SIGN: int readable = byteBuf.readableBytes(); for(int i = 0; i < readable ; i++){ sign = byteBuf.readByte(); switch(sign){ case '\r': break; case '\n': break; case RedisClientProtocol.MINUS_BYTE: redisClientProtocol = new RedisErrorParser(); break; case RedisClientProtocol.ASTERISK_BYTE: redisClientProtocol = new ArrayParser().setInOutPayloadFactory(inOutPayloadFactory); break; case RedisClientProtocol.DOLLAR_BYTE: if(inOutPayloadFactory != null) { redisClientProtocol = new BulkStringParser(inOutPayloadFactory.create()); } else { redisClientProtocol = new BulkStringParser(getBulkStringPayload()); } break; case RedisClientProtocol.COLON_BYTE: redisClientProtocol = new LongParser(); break; case RedisClientProtocol.PLUS_BYTE: redisClientProtocol = new SimpleStringParser(); break; default: throw new RedisRuntimeException("unkonwn sign:" + (char)sign); } if(redisClientProtocol != null){ commandResponseState = COMMAND_RESPONSE_STATE.READING_CONTENT; break; } } if(redisClientProtocol == null){ break; } case READING_CONTENT: RedisClientProtocol<?> result = redisClientProtocol.read(byteBuf); if(result != null){ Object payload = result.getPayload(); if(payload instanceof Exception){ handleRedisException((Exception)payload); } return format(payload); } break; default: break; } return null; } AbstractRedisCommand(String host, int port, ScheduledExecutorService scheduled);  AbstractRedisCommand(SimpleObjectPool<NettyClient> clientPool, ScheduledExecutorService scheduled);  AbstractRedisCommand(String host, int port, ScheduledExecutorService scheduled, int commandTimeoutMilli);  AbstractRedisCommand(SimpleObjectPool<NettyClient> clientPool, ScheduledExecutorService scheduled, int commandTimeoutMilli); @Override String getName(); @Override int getCommandTimeoutMilli(); void setCommandTimeoutMilli(int commandTimeoutMilli); @Override void logResponse(boolean logResponse); @Override void logRequest(boolean logRequest); static int DEFAULT_REDIS_COMMAND_TIME_OUT_MILLI; static int PROXYED_REDIS_CONNECTION_COMMAND_TIME_OUT_MILLI; }
@Test public void testDoReceiveResponse() throws Exception { AbstractRedisCommand<Object> redisCommand = new AbstractRedisCommand<Object>( getXpipeNettyClientKeyedObjectPool().getKeyPool(new DefaultEndPoint("127.0.0.1", randomPort())), scheduled) { @Override protected Object format(Object payload) { return payload; } @Override public ByteBuf getRequest() { return new RequestStringParser(getName(), "test").format(); } }; redisCommand.doReceiveResponse(null, Unpooled.copiedBuffer(":1\r\n".getBytes())); }