Window { public int getSize() { return this.futures.size(); } Window(int size);  Window(int size, ScheduledExecutorService executor, long monitorInterval, WindowListener<K,R,P> listener);  Window(int size, ScheduledExecutorService executor, long monitorInterval, WindowListener<K,R,P> listener, String monitorThreadName); int getMaxSize(); int getSize(); int getFreeSize(); boolean containsKey(K key); WindowFuture<K,R,P> get(K key); void addListener(WindowListener<K,R,P> listener); void removeListener(WindowListener<K,R,P> listener); synchronized void destroy(); synchronized boolean startMonitor(); synchronized void stopMonitor(); Map<K,WindowFuture<K,R,P>> createSortedSnapshot(); WindowFuture offer(K key, R request, long offerTimeoutMillis); WindowFuture offer(K key, R request, long offerTimeoutMillis, long expireTimeoutMillis); WindowFuture offer(K key, R request, long offerTimeoutMillis, long expireTimeoutMillis, boolean callerWaitingHint); int getPendingOfferCount(); boolean abortPendingOffers(); WindowFuture<K,R,P> complete(K key, P response); WindowFuture<K,R,P> fail(K key, Throwable t); List<WindowFuture<K,R,P>> failAll(Throwable t); WindowFuture<K,R,P> cancel(K key); List<WindowFuture<K,R,P>> cancelAll(); List<WindowFuture<K,R,P>> cancelAllExpired();  }
@Test public void simulatedMultithreadedProcessing() throws Exception { final Window<Integer,String,String> window = new Window<Integer,String,String>(5); final int requestThreadCount = 8; final int requestsPerThread = 10000; final BlockingQueue<Integer> requestQueue = new LinkedBlockingQueue<Integer>(); RequestThread[] requestThreads = new RequestThread[requestThreadCount]; for (int i = 0; i < requestThreadCount; i++) { requestThreads[i] = new RequestThread(window, requestQueue, i, requestsPerThread); } ResponseThread responseThread = new ResponseThread(window, requestQueue, requestThreadCount*requestsPerThread); for (RequestThread requestThread : requestThreads) { requestThread.start(); } responseThread.start(); for (RequestThread requestThread : requestThreads) { requestThread.join(); } responseThread.join(); for (int i = 0; i < requestThreadCount; i++) { if (requestThreads[i].throwable != null) { logger.error("", requestThreads[i].throwable); } Assert.assertNull("RequestThread " + i + " throwable wasn't null: " + requestThreads[i].throwable, requestThreads[i].throwable); } if (responseThread.throwable != null) { logger.error("", responseThread.throwable); } Assert.assertNull("ResponseThread throwable wasn't null", responseThread.throwable); Assert.assertEquals(0, window.getSize()); }