Server extends AbstractServer { public void purge(long windowId) { for (DataList dataList: publisherBuffers.values()) { dataList.purge(windowId); } } Server(EventLoop eventloop, int port);  Server(EventLoop eventloop, int port, int blocksize, int numberOfCacheBlocks); void setSpoolStorage(Storage storage); @Override void registered(SelectionKey key); @Override void unregistered(SelectionKey key); InetSocketAddress run(); InetSocketAddress run(long time); void stop(); void stop(long time); void setAuthToken(byte[] authToken); static void main(String[] args); @Override String toString(); void purge(long windowId); DataList handlePublisherRequest(PublishRequestTuple request, AbstractLengthPrependerClient connection); @Override ClientListener getClientConnection(SocketChannel sc, ServerSocketChannel ssc); @Override void handleException(Exception cce, EventLoop el); static final int DEFAULT_BUFFER_SIZE; static final int DEFAULT_NUMBER_OF_CACHED_BLOCKS; }
@Test(dependsOnMethods = {"testATonOfData"}) @SuppressWarnings("SleepWhileInLoop") public void testPurgeNonExistent() throws InterruptedException { bsc = new Controller("MyController"); eventloopClient.connect(address, bsc); bsc.purge(null, "MyPublisher", 0); for (int i = 0; i < spinCount; i++) { Thread.sleep(10); if (bsc.data != null) { break; } } eventloopClient.disconnect(bsc); assertNotNull(bsc.data); bss = new Subscriber("MyPublisher"); eventloopClient.connect(address, bss); bss.activate(null, "BufferServerOutput/BufferServerSubscriber", "MyPublisher", 0, null, 0L, 0); for (int i = 0; i < spinCount; i++) { Thread.sleep(10); if (bss.tupleCount.get() == 205) { break; } } Thread.sleep(10); eventloopClient.disconnect(bss); assertEquals(bss.tupleCount.get(), 205); }
@Test(dependsOnMethods = {"testPurgeNonExistent"}) @SuppressWarnings("SleepWhileInLoop") public void testPurgeSome() throws InterruptedException { bsc = new Controller("MyController"); eventloopClient.connect(address, bsc); bsc.purge(null, "MyPublisher", 0x7afebabe00000000L); for (int i = 0; i < spinCount; i++) { Thread.sleep(10); if (bsc.data != null) { break; } } eventloopClient.disconnect(bsc); assertNotNull(bsc.data); bss = new Subscriber("MyPublisher"); eventloopClient.connect(address, bss); bss.activate(null, "BufferServerOutput/BufferServerSubscriber", "MyPublisher", 0, null, 0L, 0); for (int i = 0; i < spinCount; i++) { Thread.sleep(10); if (bss.tupleCount.get() == 103) { break; } } eventloopClient.disconnect(bss); assertEquals(bss.tupleCount.get(), 103); }
@Test(dependsOnMethods = {"testPurgeSome"}) @SuppressWarnings("SleepWhileInLoop") public void testPurgeAll() throws InterruptedException { bsc = new Controller("MyController"); eventloopClient.connect(address, bsc); bsc.purge(null, "MyPublisher", 0x7afebabe00000001L); for (int i = 0; i < spinCount; i++) { Thread.sleep(10); if (bsc.data != null) { break; } } eventloopClient.disconnect(bsc); assertNotNull(bsc.data); bss = new Subscriber("MyPublisher"); eventloopClient.connect(address, bss); bss.activate(null, "BufferServerOutput/BufferServerSubscriber", "MyPublisher", 0, null, 0L, 0); for (int i = 0; i < spinCount; i++) { Thread.sleep(10); if (!bss.resetPayloads.isEmpty()) { break; } } Thread.sleep(10); eventloopClient.disconnect(bss); assertEquals(bss.tupleCount.get(), 1); }