Server extends AbstractServer { public void setAuthToken(byte[] authToken) { this.authToken = authToken; } 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 = {"testEarlySubscriberForLaterWindow"}) public void testAuth() throws InterruptedException { instance.setAuthToken(authToken); bsp = new Publisher("MyPublisher"); bsp.setToken(authToken); eventloopClient.connect(address, bsp); bss = new Subscriber("MySubscriber"); bss.setToken(authToken); eventloopClient.connect(address, bss); bsp.activate(null, 0L); bss.activate(null, "BufferServerOutput/BufferServerSubscriber", "MyPublisher", 0, null, 0L, 0); long resetInfo = 0x7afebabe000000faL; bsp.publishMessage(ResetWindowTuple.getSerializedTuple((int)(resetInfo >> 32), 500)); for (int i = 0; i < spinCount; i++) { Thread.sleep(10); if (!bss.resetPayloads.isEmpty()) { break; } } Thread.sleep(10); eventloopClient.disconnect(bss); eventloopClient.disconnect(bsp); assertEquals(bss.tupleCount.get(), 1); assertFalse(bss.resetPayloads.isEmpty()); }