HystrixCollapser implements HystrixExecutable<ResponseType>, HystrixObservable<ResponseType> { public Observable<ResponseType> toObservable() { return toObservable(Schedulers.computation()); } protected  HystrixCollapser(); protected  HystrixCollapser(HystrixCollapserKey collapserKey); protected  HystrixCollapser(Setter setter);   HystrixCollapser(HystrixCollapserKey collapserKey, Scope scope, CollapserTimer timer, HystrixCollapserProperties.Setter propertiesBuilder);   HystrixCollapser(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); ResponseType execute(); Future<ResponseType> queue();  }
@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.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()); }
@Test public void testUnsubscribeFromSomeDuplicateArgsDoesNotRemoveFromBatch() 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>>(); List<Subscription> subscriptions = new ArrayList<Subscription>(); for (final Observable<Integer> o: observables) { final TestSubscriber<Integer> sub = new TestSubscriber<Integer>(); subscribers.add(sub); Subscription s = o.subscribe(sub); subscriptions.add(s); } for (int i = 0; i < NUM - 1; i++) { Subscription s = subscriptions.get(i); s.unsubscribe(); } Thread.sleep(100); for (TestSubscriber<Integer> sub: subscribers) { if (!sub.isUnsubscribed()) { sub.awaitTerminalEvent(1000, TimeUnit.MILLISECONDS); System.out.println("Subscriber received : " + sub.getOnNextEvents()); sub.assertNoErrors(); sub.assertValues(5); } else { System.out.println("Subscriber is unsubscribed"); } } }