Subscriber extends AuthClient { public void activate(final String version, final String type, final String sourceId, final int mask, final Collection<Integer> partitions, final long windowId, final int bufferSize) { sendAuthenticate(); write(getSerializedRequest(version, id, type, sourceId, mask, partitions, windowId, bufferSize)); } Subscriber(String id); void activate(final String version, final String type, final String sourceId, final int mask,
      final Collection<Integer> partitions, final long windowId, final int bufferSize); @Override String toString();  }
@Test(timeOut = 1000) @SuppressWarnings("SleepWhileInLoop") public void test() throws InterruptedException { final Publisher bsp1 = new Publisher("MyPublisher"); eventloopClient.connect(address, bsp1); final Subscriber bss1 = new Subscriber("MySubscriber") { @Override public void beginWindow(int windowId) { super.beginWindow(windowId); if (windowId > 9) { synchronized (SubscriberTest.this) { SubscriberTest.this.notifyAll(); } } } @Override public String toString() { return "BufferServerSubscriber"; } }; eventloopClient.connect(address, bss1); final int baseWindow = 0x7afebabe; bsp1.activate(null, baseWindow, 0); bss1.activate(null, "BufferServerOutput/BufferServerSubscriber", "MyPublisher", 0, null, 0L, 0); final AtomicBoolean publisherRun = new AtomicBoolean(true); new Thread("publisher") { @Override @SuppressWarnings("SleepWhileInLoop") public void run() { bsp1.publishMessage(ResetWindowTuple.getSerializedTuple(baseWindow, 500)); long windowId = 0x7afebabe00000000L; try { while (publisherRun.get()) { bsp1.publishMessage(BeginWindowTuple.getSerializedTuple((int)windowId)); bsp1.publishMessage(PayloadTuple.getSerializedTuple(0, 0)); bsp1.publishMessage(EndWindowTuple.getSerializedTuple((int)windowId)); windowId++; Thread.sleep(5); } } catch (InterruptedException | CancelledKeyException e) { logger.debug("{}", e); } finally { logger.debug("publisher the middle of window = {}", Codec.getStringWindowId(windowId)); } } }.start(); synchronized (this) { wait(); } publisherRun.set(false); eventloopClient.disconnect(bsp1); eventloopClient.disconnect(bss1); final Publisher bsp2 = new Publisher("MyPublisher"); eventloopClient.connect(address, bsp2); bsp2.activate(null, 0x7afebabe, 5); final Subscriber bss2 = new Subscriber("MyPublisher") { @Override public void beginWindow(int windowId) { super.beginWindow(windowId); if (windowId > 14) { synchronized (SubscriberTest.this) { SubscriberTest.this.notifyAll(); } } } }; eventloopClient.connect(address, bss2); bss2.activate(null, "BufferServerOutput/BufferServerSubscriber", "MyPublisher", 0, null, 0x7afebabe00000008L, 0); publisherRun.set(true); new Thread("publisher") { @Override @SuppressWarnings("SleepWhileInLoop") public void run() { long windowId = 0x7afebabe00000005L; try { while (publisherRun.get()) { bsp2.publishMessage(BeginWindowTuple.getSerializedTuple((int)windowId)); byte[] buff = PayloadTuple.getSerializedTuple(0, 1); buff[buff.length - 1] = 'a'; bsp2.publishMessage(buff); bsp2.publishMessage(EndWindowTuple.getSerializedTuple((int)windowId)); windowId++; Thread.sleep(5); } } catch (InterruptedException | CancelledKeyException e) { logger.debug("", e); } finally { logger.debug("publisher in the middle of window = {}", Codec.getStringWindowId(windowId)); } } }.start(); synchronized (this) { wait(); } publisherRun.set(false); eventloopClient.disconnect(bsp2); eventloopClient.disconnect(bss2); assertTrue((bss2.lastPayload.getWindowId() - 8) * 3 <= bss2.tupleCount.get()); }