  @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.<byte[]>absent()));
    pubSubReplyInputStream.write(new PubSubReply(PubSubReply.Type.SUBSCRIBE, "bar", Optional.<byte[]>absent()));

    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());
  }