@Test
public void testCollapserUnderConcurrency() throws InterruptedException {

    final CollapserTimer timer = new RealCollapserTimer();
    final int NUM_THREADS_SUBMITTING_WORK = 8;
    final int NUM_REQUESTS_PER_THREAD = 8;
    final CountDownLatch latch = new CountDownLatch(NUM_THREADS_SUBMITTING_WORK);
    List<Runnable> runnables = new ArrayList<Runnable>();
    final ConcurrentLinkedQueue<TestSubscriber<String>> subscribers = new ConcurrentLinkedQueue<TestSubscriber<String>>();
    HystrixRequestContext context = HystrixRequestContext.initializeContext();
    final AtomicInteger uniqueInt = new AtomicInteger(0);
    for (int i = 0; i < NUM_THREADS_SUBMITTING_WORK; i++) {
        runnables.add(new Runnable() {
            @Override
            public void run() {
                try {
                    //System.out.println("Runnable starting on thread : " + Thread.currentThread().getName());
                    for (int j = 0; j < NUM_REQUESTS_PER_THREAD; j++) {
                        HystrixObservableCollapser<String, String, String, String> collapser =
                                new TestCollapserWithMultipleResponses(timer, uniqueInt.getAndIncrement(), 3, false);
                        Observable<String> o = collapser.toObservable();
                        TestSubscriber<String> subscriber = new TestSubscriber<String>();
                        o.subscribe(subscriber);
                        subscribers.offer(subscriber);
                    }
                    //System.out.println("Runnable done on thread : " + Thread.currentThread().getName());
                } finally {
                    latch.countDown();
                }
            }
        });
    }
    for (Runnable r: runnables) {
        threadPool.submit(new HystrixContextRunnable(r));
    }
    assertTrue(latch.await(1000, TimeUnit.MILLISECONDS));
    for (TestSubscriber<String> subscriber: subscribers) {
        subscriber.awaitTerminalEvent();
        if (subscriber.getOnErrorEvents().size() > 0) {
            System.out.println("ERROR : " + subscriber.getOnErrorEvents());
            for (Throwable ex: subscriber.getOnErrorEvents()) {
                ex.printStackTrace();
            }
        }
        subscriber.assertCompleted();
        subscriber.assertNoErrors();
        System.out.println("Received : " + subscriber.getOnNextEvents());
        subscriber.assertValueCount(3);
    }
    context.shutdown();
}