HystrixObservableCollapser implements HystrixObservable<ResponseType> { public Observable<ResponseType> toObservable() { return toObservable(Schedulers.computation()); } protected  HystrixObservableCollapser(); protected  HystrixObservableCollapser(HystrixCollapserKey collapserKey); protected  HystrixObservableCollapser(Setter setter);   HystrixObservableCollapser(HystrixCollapserKey collapserKey, Scope scope, CollapserTimer timer, HystrixCollapserProperties.Setter propertiesBuilder, HystrixCollapserMetrics metrics); HystrixCollapserKey getCollapserKey(); Scope getScope(); HystrixCollapserMetrics getMetrics(); abstract RequestArgumentType getRequestArgument(); Observable<ResponseType> observe(); Observable<ResponseType> toObservable(); Observable<ResponseType> toObservable(Scheduler observeOn);  }
@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 { 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); } } 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(); }
@Test public void testDuplicateArgumentsWithRequestCachingOn() throws Exception { final int NUM = 10; List<Observable<Integer>> observables = new ArrayList<Observable<Integer>>(); for (int i = 0; i < NUM; i++) { MyCollapser c = new MyCollapser("5", true); observables.add(c.toObservable()); } List<TestSubscriber<Integer>> subscribers = new ArrayList<TestSubscriber<Integer>>(); for (final Observable<Integer> o: observables) { final TestSubscriber<Integer> sub = new TestSubscriber<Integer>(); subscribers.add(sub); o.subscribe(sub); } Thread.sleep(100); for (TestSubscriber<Integer> sub: subscribers) { sub.awaitTerminalEvent(1000, TimeUnit.MILLISECONDS); System.out.println("Subscriber received : " + sub.getOnNextEvents()); sub.assertCompleted(); sub.assertNoErrors(); sub.assertValues(5); } }
@Test public void testDuplicateArgumentsWithRequestCachingOff() throws Exception { final int NUM = 10; List<Observable<Integer>> observables = new ArrayList<Observable<Integer>>(); for (int i = 0; i < NUM; i++) { MyCollapser c = new MyCollapser("5", false); observables.add(c.toObservable()); } List<TestSubscriber<Integer>> subscribers = new ArrayList<TestSubscriber<Integer>>(); for (final Observable<Integer> o: observables) { final TestSubscriber<Integer> sub = new TestSubscriber<Integer>(); subscribers.add(sub); o.subscribe(sub); } Thread.sleep(100); AtomicInteger numErrors = new AtomicInteger(0); AtomicInteger numValues = new AtomicInteger(0); for (TestSubscriber<Integer> sub: subscribers) { sub.awaitTerminalEvent(1000, TimeUnit.MILLISECONDS); if (sub.getOnCompletedEvents().isEmpty()) { System.out.println(Thread.currentThread().getName() + " Error : " + sub.getOnErrorEvents()); sub.assertError(IllegalArgumentException.class); sub.assertNoValues(); numErrors.getAndIncrement(); } else { System.out.println(Thread.currentThread().getName() + " OnNext : " + sub.getOnNextEvents()); sub.assertValues(5); sub.assertCompleted(); sub.assertNoErrors(); numValues.getAndIncrement(); } } assertEquals(1, numValues.get()); assertEquals(NUM - 1, numErrors.get()); }