AbstractSubscribe extends AbstractRedisCommand<Object> implements Subscribe { protected void handleMessage(Object response) { if(!(response instanceof Object[])) { throw new RedisRuntimeException(String.format("Subscribe subscribeChannel response incorrect: %s", response)); } SubscribeMessageHandler handler = getSubscribeMessageHandler(); Pair<String, String> channelAndMessage = handler.handle(payloadToStringArray(response)); if(channelAndMessage != null) { notifyListeners(channelAndMessage); } } protected  AbstractSubscribe(String host, int port, ScheduledExecutorService scheduled,
                                Subscribe.MESSAGE_TYPE messageType, String... subscribeChannel);  AbstractSubscribe(String host, int port, ScheduledExecutorService scheduled, int commandTimeoutMilli,
                             MESSAGE_TYPE messageType, String... subscribeChannel);  AbstractSubscribe(SimpleObjectPool<NettyClient> clientPool, ScheduledExecutorService scheduled,
                             MESSAGE_TYPE messageType, String... subscribeChannel); @Override void addChannelListener(SubscribeListener listener); @Override void removeChannelListener(SubscribeListener listener); @Override ByteBuf getRequest(); @Override void unSubscribe();  }
@Test public void testHandleMessage() throws Exception { String message = "message"; Object[] response = new Object[]{new ByteArrayOutputStreamPayload(Subscribe.MESSAGE_TYPE.MESSAGE.desc()), new ByteArrayOutputStreamPayload(channel), new ByteArrayOutputStreamPayload(message)}; subscribe.addChannelListener((ch, msg) -> {Assert.assertEquals(channel, ch); Assert.assertEquals(message, msg);}); subscribe.handleMessage(response); }