### Question:
DispatchManager extends Thread { public synchronized void subscribe(String name, DispatchChannel dispatchChannel) { Optional<DispatchChannel> previous = Optional.ofNullable(subscriptions.get(name)); subscriptions.put(name, dispatchChannel); try { pubSubConnection.subscribe(name); } catch (IOException e) { logger.warn("Subscription error", e); } if (previous.isPresent()) { dispatchUnsubscription(name, previous.get()); } } DispatchManager(RedisPubSubConnectionFactory redisPubSubConnectionFactory,
                         Optional<DispatchChannel> deadLetterChannel); @Override void start(); void shutdown(); synchronized void subscribe(String name, DispatchChannel dispatchChannel); synchronized void unsubscribe(String name, DispatchChannel channel); boolean hasSubscription(String name); @Override void run();  }### Answer:
@Test public void testSubscribe() throws IOException { DispatchChannel dispatchChannel = mock(DispatchChannel.class); dispatchManager.subscribe("foo", dispatchChannel); pubSubReplyInputStream.write(new PubSubReply(PubSubReply.Type.SUBSCRIBE, "foo", Optional.empty())); verify(dispatchChannel, timeout(1000)).onDispatchSubscribed(eq("foo")); }
@Test public void testMessages() throws IOException { DispatchChannel fooChannel = mock(DispatchChannel.class); DispatchChannel barChannel = mock(DispatchChannel.class); dispatchManager.subscribe("foo", fooChannel); dispatchManager.subscribe("bar", barChannel); pubSubReplyInputStream.write(new PubSubReply(PubSubReply.Type.SUBSCRIBE, "foo", Optional.empty())); pubSubReplyInputStream.write(new PubSubReply(PubSubReply.Type.SUBSCRIBE, "bar", Optional.empty())); verify(fooChannel, timeout(1000)).onDispatchSubscribed(eq("foo")); verify(barChannel, timeout(1000)).onDispatchSubscribed(eq("bar")); pubSubReplyInputStream.write(new PubSubReply(PubSubReply.Type.MESSAGE, "foo", Optional.of("hello".getBytes()))); pubSubReplyInputStream.write(new PubSubReply(PubSubReply.Type.MESSAGE, "bar", Optional.of("there".getBytes()))); ArgumentCaptor<byte[]> captor = ArgumentCaptor.forClass(byte[].class); verify(fooChannel, timeout(1000)).onDispatchMessage(eq("foo"), captor.capture()); assertArrayEquals("hello".getBytes(), captor.getValue()); verify(barChannel, timeout(1000)).onDispatchMessage(eq("bar"), captor.capture()); assertArrayEquals("there".getBytes(), captor.getValue()); }