@Test
public void testSerialization() throws Exception
{

  FastPublisherImpl publisher = new FastPublisherImpl(24 * 1024);
  final String message = "hello!";
  publisher.put(message);
  byte[] buffer = publisher.consume();
  FastSubscriber subscriber = new FastSubscriber("subscriber", 1024);
  subscriber.serde = subscriber.statefulSerde = new DefaultStatefulStreamCodec<>();
  SweepableReservoir sr = subscriber.acquireReservoir("res", 1024);
  sr.setSink(new Sink<Object>()
  {
    @Override
    public void put(Object tuple)
    {
      assert (tuple.equals(message));
    }
    @Override
    public int getCount(boolean reset)
    {
      throw new UnsupportedOperationException("Not supported yet.");
    }
  });
  int size = buffer[0];
  size |= buffer[1] << 8;
  Assert.assertEquals("size", buffer.length - 2, size);
  subscriber.onMessage(buffer, 2, buffer.length - 2);
  sr.sweep();
  sr.sweep();
  for (int i = 0; i < 1024; i++) {
    publisher.put(message);
  }
  buffer = publisher.consume();
  assert (buffer.length == (size + 2) * 1024);
  int index = 0;
  for (int i = 0; i < 1024; i++) {
    size = buffer[index++];
    size |= buffer[index++] << 8;
    subscriber.onMessage(buffer, index, size);
    index += size;
  }
  sr.sweep();
  sr.sweep();
  for (int i = 0; i < 1024; i++) {
    publisher.put(message);
  }
  buffer = publisher.consume();
  assert (buffer.length == (size + 2) * 1024);
  index = 0;
  for (int i = 0; i < 1024; i++) {
    size = buffer[index++];
    size |= buffer[index++] << 8;
    subscriber.onMessage(buffer, index, size);
    index += size;
  }
  sr.sweep();
  sr.sweep();
}