RedisSubscribeThread extends Thread { void process(RedisConnection connection) throws Exception { subscribe(connection); listener.onSubscribe(); while (true) { Object value = connection.read(); if (value instanceof String && "OK".equals(value)) { break; } if (value instanceof Object[]) { Object[] message = (Object[]) value; String kind = decode((byte[]) message[0]); if ("message".equals(kind)) { byte[] payload = (byte[]) message[2]; listener.onMessage(payload); continue; } } throw new Error("unexpected response, message=" + value); } } RedisSubscribeThread(String name, RedisImpl redis, RedisChannelListener listener, String channel); @Override void run(); void close();  }
@Test void process() throws Exception { var connection = new RedisConnection(); connection.outputStream = new RedisOutputStream(new ByteArrayOutputStream(), 512); connection.inputStream = new RedisInputStream(new ByteArrayInputStream(Strings.bytes("*3\r\n$9\r\nsubscribe\r\n$7\r\nchannel\r\n:1\r\n" + "*3\r\n$7\r\nmessage\r\n$7\r\nchannel\r\n$5\r\nvalue\r\n" + "+OK\r\n"))); thread.process(connection); verify(listener).onSubscribe(); verify(listener).onMessage(Strings.bytes("value")); }