HystrixObservableCollapser implements HystrixObservable<ResponseType> { public Observable<ResponseType> observe() { ReplaySubject<ResponseType> subject = ReplaySubject.create(); final Subscription underlyingSubscription = toObservable().subscribe(subject); return subject.doOnUnsubscribe(new Action0() { @Override public void call() { underlyingSubscription.unsubscribe(); } }); } 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 testTwoRequests() throws Exception { TestCollapserTimer timer = new TestCollapserTimer(); HystrixObservableCollapser<String, String, String, String> collapser1 = new TestRequestCollapser(timer, 1); HystrixObservableCollapser<String, String, String, String> collapser2 = new TestRequestCollapser(timer, 2); Future<String> response1 = collapser1.observe().toBlocking().toFuture(); Future<String> response2 = collapser2.observe().toBlocking().toFuture(); timer.incrementTime(10); assertEquals("1", response1.get()); assertEquals("2", response2.get()); assertEquals(1, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size()); Iterator<HystrixInvokableInfo<?>> cmdIterator = HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().iterator(); assertEquals(2, cmdIterator.next().getNumberCollapsed()); }
@Test public void testTwoRequestsWhichShouldEachEmitTwice() throws Exception { CollapserTimer timer = new RealCollapserTimer(); HystrixObservableCollapser<String, String, String, String> collapser1 = new TestCollapserWithMultipleResponses(timer, 1, 3, false, false, prefixMapper, onMissingComplete); HystrixObservableCollapser<String, String, String, String> collapser2 = new TestCollapserWithMultipleResponses(timer, 2, 3, false, false, prefixMapper, onMissingComplete); TestSubscriber<String> testSubscriber1 = new TestSubscriber<String>(); TestSubscriber<String> testSubscriber2 = new TestSubscriber<String>(); System.out.println(System.currentTimeMillis() + "Starting to observe collapser1"); collapser1.observe().subscribe(testSubscriber1); collapser2.observe().subscribe(testSubscriber2); System.out.println(System.currentTimeMillis() + "Done with collapser observe()s"); testSubscriber1.awaitTerminalEvent(); testSubscriber2.awaitTerminalEvent(); testSubscriber1.assertCompleted(); testSubscriber1.assertNoErrors(); testSubscriber1.assertValues("1:1", "1:2", "1:3"); testSubscriber2.assertCompleted(); testSubscriber2.assertNoErrors(); testSubscriber2.assertValues("2:2", "2:4", "2:6"); }
@Test public void testTwoRequestsWithErrorProducingBatchCommand() { TestCollapserTimer timer = new TestCollapserTimer(); HystrixObservableCollapser<String, String, String, String> collapser1 = new TestCollapserWithMultipleResponses(timer, 1, 3, true); HystrixObservableCollapser<String, String, String, String> collapser2 = new TestCollapserWithMultipleResponses(timer, 2, 3, true); System.out.println("Starting to observe collapser1"); Observable<String> result1 = collapser1.observe(); Observable<String> result2 = collapser2.observe(); timer.incrementTime(10); TestSubscriber<String> testSubscriber1 = new TestSubscriber<String>(); result1.subscribe(testSubscriber1); TestSubscriber<String> testSubscriber2 = new TestSubscriber<String>(); result2.subscribe(testSubscriber2); testSubscriber1.awaitTerminalEvent(); testSubscriber2.awaitTerminalEvent(); testSubscriber1.assertError(RuntimeException.class); testSubscriber1.assertNoValues(); testSubscriber2.assertError(RuntimeException.class); testSubscriber2.assertNoValues(); }
@Test public void testTwoRequestsWithErrorInDemultiplex() { TestCollapserTimer timer = new TestCollapserTimer(); HystrixObservableCollapser<String, String, String, String> collapser1 = new TestCollapserWithMultipleResponses(timer, 1, 3, false, false, mapWithErrorOn1, onMissingError); HystrixObservableCollapser<String, String, String, String> collapser2 = new TestCollapserWithMultipleResponses(timer, 2, 3, false, false, mapWithErrorOn1, onMissingError); System.out.println("Starting to observe collapser1"); Observable<String> result1 = collapser1.observe(); Observable<String> result2 = collapser2.observe(); timer.incrementTime(10); TestSubscriber<String> testSubscriber1 = new TestSubscriber<String>(); result1.subscribe(testSubscriber1); TestSubscriber<String> testSubscriber2 = new TestSubscriber<String>(); result2.subscribe(testSubscriber2); testSubscriber1.awaitTerminalEvent(); testSubscriber2.awaitTerminalEvent(); testSubscriber1.assertError(RuntimeException.class); testSubscriber1.assertNoValues(); testSubscriber2.assertCompleted(); testSubscriber2.assertNoErrors(); testSubscriber2.assertValues("2:2", "2:4", "2:6"); }
@Test public void testTwoRequestsWithEmptyResponseAndOnMissingError() { TestCollapserTimer timer = new TestCollapserTimer(); HystrixObservableCollapser<String, String, String, String> collapser1 = new TestCollapserWithMultipleResponses(timer, 1, 0, onMissingError); HystrixObservableCollapser<String, String, String, String> collapser2 = new TestCollapserWithMultipleResponses(timer, 2, 0, onMissingError); System.out.println("Starting to observe collapser1"); Observable<String> result1 = collapser1.observe(); Observable<String> result2 = collapser2.observe(); timer.incrementTime(10); TestSubscriber<String> testSubscriber1 = new TestSubscriber<String>(); result1.subscribe(testSubscriber1); TestSubscriber<String> testSubscriber2 = new TestSubscriber<String>(); result2.subscribe(testSubscriber2); testSubscriber1.awaitTerminalEvent(); testSubscriber2.awaitTerminalEvent(); testSubscriber1.assertError(IllegalStateException.class); testSubscriber1.assertNoValues(); testSubscriber2.assertError(IllegalStateException.class); testSubscriber2.assertNoValues(); }
@Test public void testTwoRequestsWithEmptyResponseAndOnMissingThrow() { TestCollapserTimer timer = new TestCollapserTimer(); HystrixObservableCollapser<String, String, String, String> collapser1 = new TestCollapserWithMultipleResponses(timer, 1, 0, onMissingThrow); HystrixObservableCollapser<String, String, String, String> collapser2 = new TestCollapserWithMultipleResponses(timer, 2, 0, onMissingThrow); System.out.println("Starting to observe collapser1"); Observable<String> result1 = collapser1.observe(); Observable<String> result2 = collapser2.observe(); timer.incrementTime(10); TestSubscriber<String> testSubscriber1 = new TestSubscriber<String>(); result1.subscribe(testSubscriber1); TestSubscriber<String> testSubscriber2 = new TestSubscriber<String>(); result2.subscribe(testSubscriber2); testSubscriber1.awaitTerminalEvent(); testSubscriber2.awaitTerminalEvent(); testSubscriber1.assertError(RuntimeException.class); testSubscriber1.assertNoValues(); testSubscriber2.assertError(RuntimeException.class); testSubscriber2.assertNoValues(); }
@Test public void testTwoRequestsWithEmptyResponseAndOnMissingComplete() { TestCollapserTimer timer = new TestCollapserTimer(); HystrixObservableCollapser<String, String, String, String> collapser1 = new TestCollapserWithMultipleResponses(timer, 1, 0, onMissingComplete); HystrixObservableCollapser<String, String, String, String> collapser2 = new TestCollapserWithMultipleResponses(timer, 2, 0, onMissingComplete); System.out.println("Starting to observe collapser1"); Observable<String> result1 = collapser1.observe(); Observable<String> result2 = collapser2.observe(); timer.incrementTime(10); TestSubscriber<String> testSubscriber1 = new TestSubscriber<String>(); result1.subscribe(testSubscriber1); TestSubscriber<String> testSubscriber2 = new TestSubscriber<String>(); result2.subscribe(testSubscriber2); testSubscriber1.awaitTerminalEvent(); testSubscriber2.awaitTerminalEvent(); testSubscriber1.assertCompleted(); testSubscriber1.assertNoErrors(); testSubscriber1.assertNoValues(); testSubscriber2.assertCompleted(); testSubscriber2.assertNoErrors(); testSubscriber2.assertNoValues(); }
@Test public void testTwoRequestsWithEmptyResponseAndOnMissingIgnore() { TestCollapserTimer timer = new TestCollapserTimer(); HystrixObservableCollapser<String, String, String, String> collapser1 = new TestCollapserWithMultipleResponses(timer, 1, 0, onMissingIgnore); HystrixObservableCollapser<String, String, String, String> collapser2 = new TestCollapserWithMultipleResponses(timer, 2, 0, onMissingIgnore); System.out.println("Starting to observe collapser1"); Observable<String> result1 = collapser1.observe(); Observable<String> result2 = collapser2.observe(); timer.incrementTime(10); TestSubscriber<String> testSubscriber1 = new TestSubscriber<String>(); result1.subscribe(testSubscriber1); TestSubscriber<String> testSubscriber2 = new TestSubscriber<String>(); result2.subscribe(testSubscriber2); testSubscriber1.awaitTerminalEvent(); testSubscriber2.awaitTerminalEvent(); testSubscriber1.assertCompleted(); testSubscriber1.assertNoErrors(); testSubscriber1.assertNoValues(); testSubscriber2.assertCompleted(); testSubscriber2.assertNoErrors(); testSubscriber2.assertNoValues(); }
@Test public void testTwoRequestsWithEmptyResponseAndOnMissingFillInStaticValue() { TestCollapserTimer timer = new TestCollapserTimer(); HystrixObservableCollapser<String, String, String, String> collapser1 = new TestCollapserWithMultipleResponses(timer, 1, 0, onMissingFillIn); HystrixObservableCollapser<String, String, String, String> collapser2 = new TestCollapserWithMultipleResponses(timer, 2, 0, onMissingFillIn); System.out.println("Starting to observe collapser1"); Observable<String> result1 = collapser1.observe(); Observable<String> result2 = collapser2.observe(); timer.incrementTime(10); TestSubscriber<String> testSubscriber1 = new TestSubscriber<String>(); result1.subscribe(testSubscriber1); TestSubscriber<String> testSubscriber2 = new TestSubscriber<String>(); result2.subscribe(testSubscriber2); testSubscriber1.awaitTerminalEvent(); testSubscriber2.awaitTerminalEvent(); testSubscriber1.assertCompleted(); testSubscriber1.assertNoErrors(); testSubscriber1.assertValues("fillin"); testSubscriber2.assertCompleted(); testSubscriber2.assertNoErrors(); testSubscriber2.assertValues("fillin"); }
@Test public void testTwoRequestsWithValuesForOneArgOnlyAndOnMissingComplete() { TestCollapserTimer timer = new TestCollapserTimer(); HystrixObservableCollapser<String, String, String, String> collapser1 = new TestCollapserWithMultipleResponses(timer, 1, 0, onMissingComplete); HystrixObservableCollapser<String, String, String, String> collapser2 = new TestCollapserWithMultipleResponses(timer, 2, 5, onMissingComplete); System.out.println("Starting to observe collapser1"); Observable<String> result1 = collapser1.observe(); Observable<String> result2 = collapser2.observe(); timer.incrementTime(10); TestSubscriber<String> testSubscriber1 = new TestSubscriber<String>(); result1.subscribe(testSubscriber1); TestSubscriber<String> testSubscriber2 = new TestSubscriber<String>(); result2.subscribe(testSubscriber2); testSubscriber1.awaitTerminalEvent(); testSubscriber2.awaitTerminalEvent(); testSubscriber1.assertCompleted(); testSubscriber1.assertNoErrors(); testSubscriber1.assertNoValues(); testSubscriber2.assertCompleted(); testSubscriber2.assertNoErrors(); testSubscriber2.assertValues("2:2", "2:4", "2:6", "2:8", "2:10"); }
@Test public void testTwoRequestsWithValuesForOneArgOnlyAndOnMissingFillInStaticValue() { TestCollapserTimer timer = new TestCollapserTimer(); HystrixObservableCollapser<String, String, String, String> collapser1 = new TestCollapserWithMultipleResponses(timer, 1, 0, onMissingFillIn); HystrixObservableCollapser<String, String, String, String> collapser2 = new TestCollapserWithMultipleResponses(timer, 2, 5, onMissingFillIn); System.out.println("Starting to observe collapser1"); Observable<String> result1 = collapser1.observe(); Observable<String> result2 = collapser2.observe(); timer.incrementTime(10); TestSubscriber<String> testSubscriber1 = new TestSubscriber<String>(); result1.subscribe(testSubscriber1); TestSubscriber<String> testSubscriber2 = new TestSubscriber<String>(); result2.subscribe(testSubscriber2); testSubscriber1.awaitTerminalEvent(); testSubscriber2.awaitTerminalEvent(); testSubscriber1.assertCompleted(); testSubscriber1.assertNoErrors(); testSubscriber1.assertValues("fillin"); testSubscriber2.assertCompleted(); testSubscriber2.assertNoErrors(); testSubscriber2.assertValues("2:2", "2:4", "2:6", "2:8", "2:10"); }
@Test public void testTwoRequestsWithValuesForOneArgOnlyAndOnMissingIgnore() { TestCollapserTimer timer = new TestCollapserTimer(); HystrixObservableCollapser<String, String, String, String> collapser1 = new TestCollapserWithMultipleResponses(timer, 1, 0, onMissingIgnore); HystrixObservableCollapser<String, String, String, String> collapser2 = new TestCollapserWithMultipleResponses(timer, 2, 5, onMissingIgnore); System.out.println("Starting to observe collapser1"); Observable<String> result1 = collapser1.observe(); Observable<String> result2 = collapser2.observe(); timer.incrementTime(10); TestSubscriber<String> testSubscriber1 = new TestSubscriber<String>(); result1.subscribe(testSubscriber1); TestSubscriber<String> testSubscriber2 = new TestSubscriber<String>(); result2.subscribe(testSubscriber2); testSubscriber1.awaitTerminalEvent(); testSubscriber2.awaitTerminalEvent(); testSubscriber1.assertCompleted(); testSubscriber1.assertNoErrors(); testSubscriber1.assertNoValues(); testSubscriber2.assertCompleted(); testSubscriber2.assertNoErrors(); testSubscriber2.assertValues("2:2", "2:4", "2:6", "2:8", "2:10"); }
@Test public void testTwoRequestsWithValuesForOneArgOnlyAndOnMissingError() { TestCollapserTimer timer = new TestCollapserTimer(); HystrixObservableCollapser<String, String, String, String> collapser1 = new TestCollapserWithMultipleResponses(timer, 1, 0, onMissingError); HystrixObservableCollapser<String, String, String, String> collapser2 = new TestCollapserWithMultipleResponses(timer, 2, 5, onMissingError); System.out.println("Starting to observe collapser1"); Observable<String> result1 = collapser1.observe(); Observable<String> result2 = collapser2.observe(); timer.incrementTime(10); TestSubscriber<String> testSubscriber1 = new TestSubscriber<String>(); result1.subscribe(testSubscriber1); TestSubscriber<String> testSubscriber2 = new TestSubscriber<String>(); result2.subscribe(testSubscriber2); testSubscriber1.awaitTerminalEvent(); testSubscriber2.awaitTerminalEvent(); testSubscriber1.assertError(IllegalStateException.class); testSubscriber1.assertNoValues(); testSubscriber2.assertCompleted(); testSubscriber2.assertNoErrors(); testSubscriber2.assertValues("2:2", "2:4", "2:6", "2:8", "2:10"); }
@Test public void testTwoRequestsWithValuesForOneArgOnlyAndOnMissingThrow() { TestCollapserTimer timer = new TestCollapserTimer(); HystrixObservableCollapser<String, String, String, String> collapser1 = new TestCollapserWithMultipleResponses(timer, 1, 0, onMissingThrow); HystrixObservableCollapser<String, String, String, String> collapser2 = new TestCollapserWithMultipleResponses(timer, 2, 5, onMissingThrow); System.out.println("Starting to observe collapser1"); Observable<String> result1 = collapser1.observe(); Observable<String> result2 = collapser2.observe(); timer.incrementTime(10); TestSubscriber<String> testSubscriber1 = new TestSubscriber<String>(); result1.subscribe(testSubscriber1); TestSubscriber<String> testSubscriber2 = new TestSubscriber<String>(); result2.subscribe(testSubscriber2); testSubscriber1.awaitTerminalEvent(); testSubscriber2.awaitTerminalEvent(); testSubscriber1.assertError(RuntimeException.class); testSubscriber1.assertNoValues(); testSubscriber2.assertCompleted(); testSubscriber2.assertNoErrors(); testSubscriber2.assertValues("2:2", "2:4", "2:6", "2:8", "2:10"); }
@Test public void testTwoRequestsWithValuesForWrongArgs() { TestCollapserTimer timer = new TestCollapserTimer(); HystrixObservableCollapser<String, String, String, String> collapser1 = new TestCollapserWithMultipleResponses(timer, 1, 3, false, false, map1To3And2To2, onMissingError); HystrixObservableCollapser<String, String, String, String> collapser2 = new TestCollapserWithMultipleResponses(timer, 2, 3, false, false, map1To3And2To2, onMissingError); System.out.println("Starting to observe collapser1"); Observable<String> result1 = collapser1.observe(); Observable<String> result2 = collapser2.observe(); timer.incrementTime(10); TestSubscriber<String> testSubscriber1 = new TestSubscriber<String>(); result1.subscribe(testSubscriber1); TestSubscriber<String> testSubscriber2 = new TestSubscriber<String>(); result2.subscribe(testSubscriber2); testSubscriber1.awaitTerminalEvent(); testSubscriber2.awaitTerminalEvent(); testSubscriber1.assertError(RuntimeException.class); testSubscriber1.assertNoValues(); testSubscriber2.assertCompleted(); testSubscriber2.assertNoErrors(); testSubscriber2.assertValues("2:2", "2:4", "2:6"); }
@Test public void testTwoRequestsWhenBatchCommandFails() { TestCollapserTimer timer = new TestCollapserTimer(); HystrixObservableCollapser<String, String, String, String> collapser1 = new TestCollapserWithMultipleResponses(timer, 1, 3, false, true, map1To3And2To2, onMissingError); HystrixObservableCollapser<String, String, String, String> collapser2 = new TestCollapserWithMultipleResponses(timer, 2, 3, false, true, map1To3And2To2, onMissingError); System.out.println("Starting to observe collapser1"); Observable<String> result1 = collapser1.observe(); Observable<String> result2 = collapser2.observe(); timer.incrementTime(10); TestSubscriber<String> testSubscriber1 = new TestSubscriber<String>(); result1.subscribe(testSubscriber1); TestSubscriber<String> testSubscriber2 = new TestSubscriber<String>(); result2.subscribe(testSubscriber2); testSubscriber1.awaitTerminalEvent(); testSubscriber2.awaitTerminalEvent(); testSubscriber1.assertError(RuntimeException.class); testSubscriber1.getOnErrorEvents().get(0).printStackTrace(); testSubscriber1.assertNoValues(); testSubscriber2.assertError(RuntimeException.class); testSubscriber2.assertNoValues(); }
@Test public void testEarlyUnsubscribeFromAllCancelsBatch() throws Exception { TestCollapserTimer timer = new TestCollapserTimer(); HystrixObservableCollapser<String, String, String, String> collapser1 = new TestRequestCollapser(timer, 1); Observable<String> response1 = collapser1.observe(); HystrixObservableCollapser<String, String, String, String> collapser2 = new TestRequestCollapser(timer, 2); Observable<String> response2 = collapser2.observe(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); final AtomicReference<String> value1 = new AtomicReference<String>(null); final AtomicReference<String> value2 = new AtomicReference<String>(null); Subscription s1 = response1 .doOnUnsubscribe(new Action0() { @Override public void call() { System.out.println(System.currentTimeMillis() + " : s1 Unsubscribed!"); latch1.countDown(); } }) .subscribe(new Subscriber<String>() { @Override public void onCompleted() { System.out.println(System.currentTimeMillis() + " : s1 OnCompleted"); latch1.countDown(); } @Override public void onError(Throwable e) { System.out.println(System.currentTimeMillis() + " : s1 OnError : " + e); latch1.countDown(); } @Override public void onNext(String s) { System.out.println(System.currentTimeMillis() + " : s1 OnNext : " + s); value1.set(s); } }); Subscription s2 = response2 .doOnUnsubscribe(new Action0() { @Override public void call() { System.out.println(System.currentTimeMillis() + " : s2 Unsubscribed!"); latch2.countDown(); } }) .subscribe(new Subscriber<String>() { @Override public void onCompleted() { System.out.println(System.currentTimeMillis() + " : s2 OnCompleted"); latch2.countDown(); } @Override public void onError(Throwable e) { System.out.println(System.currentTimeMillis() + " : s2 OnError : " + e); latch2.countDown(); } @Override public void onNext(String s) { System.out.println(System.currentTimeMillis() + " : s2 OnNext : " + s); value2.set(s); } }); s1.unsubscribe(); s2.unsubscribe(); timer.incrementTime(10); assertTrue(latch1.await(1000, TimeUnit.MILLISECONDS)); assertTrue(latch2.await(1000, TimeUnit.MILLISECONDS)); assertNull(value1.get()); assertNull(value2.get()); System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); assertEquals(0, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size()); }
@Test public void testRequestThenCacheHitAndCacheHitUnsubscribed() throws Exception { TestCollapserTimer timer = new TestCollapserTimer(); HystrixObservableCollapser<String, String, String, String> collapser1 = new SuccessfulCacheableCollapsedCommand(timer, "foo", true); Observable<String> response1 = collapser1.observe(); HystrixObservableCollapser<String, String, String, String> collapser2 = new SuccessfulCacheableCollapsedCommand(timer, "foo", true); Observable<String> response2 = collapser2.observe(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); final AtomicReference<String> value1 = new AtomicReference<String>(null); final AtomicReference<String> value2 = new AtomicReference<String>(null); Subscription s1 = response1 .doOnUnsubscribe(new Action0() { @Override public void call() { System.out.println(System.currentTimeMillis() + " : s1 Unsubscribed!"); latch1.countDown(); } }) .subscribe(new Subscriber<String>() { @Override public void onCompleted() { System.out.println(System.currentTimeMillis() + " : s1 OnCompleted"); latch1.countDown(); } @Override public void onError(Throwable e) { System.out.println(System.currentTimeMillis() + " : s1 OnError : " + e); latch1.countDown(); } @Override public void onNext(String s) { System.out.println(System.currentTimeMillis() + " : s1 OnNext : " + s); value1.set(s); } }); Subscription s2 = response2 .doOnUnsubscribe(new Action0() { @Override public void call() { System.out.println(System.currentTimeMillis() + " : s2 Unsubscribed!"); latch2.countDown(); } }) .subscribe(new Subscriber<String>() { @Override public void onCompleted() { System.out.println(System.currentTimeMillis() + " : s2 OnCompleted"); latch2.countDown(); } @Override public void onError(Throwable e) { System.out.println(System.currentTimeMillis() + " : s2 OnError : " + e); latch2.countDown(); } @Override public void onNext(String s) { System.out.println(System.currentTimeMillis() + " : s2 OnNext : " + s); value2.set(s); } }); s2.unsubscribe(); timer.incrementTime(10); assertTrue(latch1.await(1000, TimeUnit.MILLISECONDS)); assertTrue(latch2.await(1000, TimeUnit.MILLISECONDS)); assertEquals("foo", value1.get()); assertNull(value2.get()); System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); assertEquals(1, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size()); HystrixInvokableInfo<?> command = HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().iterator().next(); assertCommandExecutionEvents(command, HystrixEventType.EMIT, HystrixEventType.SUCCESS, HystrixEventType.COLLAPSED); assertEquals(1, command.getNumberCollapsed()); }
@Test public void testRequestThenCacheHitAndOriginalUnsubscribed() throws Exception { TestCollapserTimer timer = new TestCollapserTimer(); HystrixObservableCollapser<String, String, String, String> collapser1 = new SuccessfulCacheableCollapsedCommand(timer, "foo", true); Observable<String> response1 = collapser1.observe(); HystrixObservableCollapser<String, String, String, String> collapser2 = new SuccessfulCacheableCollapsedCommand(timer, "foo", true); Observable<String> response2 = collapser2.observe(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); final AtomicReference<String> value1 = new AtomicReference<String>(null); final AtomicReference<String> value2 = new AtomicReference<String>(null); Subscription s1 = response1 .doOnUnsubscribe(new Action0() { @Override public void call() { System.out.println(System.currentTimeMillis() + " : s1 Unsubscribed!"); latch1.countDown(); } }) .subscribe(new Subscriber<String>() { @Override public void onCompleted() { System.out.println(System.currentTimeMillis() + " : s1 OnCompleted"); latch1.countDown(); } @Override public void onError(Throwable e) { System.out.println(System.currentTimeMillis() + " : s1 OnError : " + e); latch1.countDown(); } @Override public void onNext(String s) { System.out.println(System.currentTimeMillis() + " : s1 OnNext : " + s); value1.set(s); } }); Subscription s2 = response2 .doOnUnsubscribe(new Action0() { @Override public void call() { System.out.println(System.currentTimeMillis() + " : s2 Unsubscribed!"); latch2.countDown(); } }) .subscribe(new Subscriber<String>() { @Override public void onCompleted() { System.out.println(System.currentTimeMillis() + " : s2 OnCompleted"); latch2.countDown(); } @Override public void onError(Throwable e) { System.out.println(System.currentTimeMillis() + " : s2 OnError : " + e); latch2.countDown(); } @Override public void onNext(String s) { System.out.println(System.currentTimeMillis() + " : s2 OnNext : " + s); value2.set(s); } }); s1.unsubscribe(); timer.incrementTime(10); assertTrue(latch1.await(1000, TimeUnit.MILLISECONDS)); assertTrue(latch2.await(1000, TimeUnit.MILLISECONDS)); assertNull(value1.get()); assertEquals("foo", value2.get()); System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); assertEquals(1, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size()); HystrixInvokableInfo<?> command = HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().iterator().next(); assertCommandExecutionEvents(command, HystrixEventType.EMIT, HystrixEventType.SUCCESS, HystrixEventType.COLLAPSED); assertEquals(1, command.getNumberCollapsed()); }
@Test public void testRequestThenTwoCacheHitsOriginalAndOneCacheHitUnsubscribed() throws Exception { TestCollapserTimer timer = new TestCollapserTimer(); HystrixObservableCollapser<String, String, String, String> collapser1 = new SuccessfulCacheableCollapsedCommand(timer, "foo", true); Observable<String> response1 = collapser1.observe(); HystrixObservableCollapser<String, String, String, String> collapser2 = new SuccessfulCacheableCollapsedCommand(timer, "foo", true); Observable<String> response2 = collapser2.observe(); HystrixObservableCollapser<String, String, String, String> collapser3 = new SuccessfulCacheableCollapsedCommand(timer, "foo", true); Observable<String> response3 = collapser3.observe(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); final CountDownLatch latch3 = new CountDownLatch(1); final AtomicReference<String> value1 = new AtomicReference<String>(null); final AtomicReference<String> value2 = new AtomicReference<String>(null); final AtomicReference<String> value3 = new AtomicReference<String>(null); Subscription s1 = response1 .doOnUnsubscribe(new Action0() { @Override public void call() { System.out.println(System.currentTimeMillis() + " : s1 Unsubscribed!"); latch1.countDown(); } }) .subscribe(new Subscriber<String>() { @Override public void onCompleted() { System.out.println(System.currentTimeMillis() + " : s1 OnCompleted"); latch1.countDown(); } @Override public void onError(Throwable e) { System.out.println(System.currentTimeMillis() + " : s1 OnError : " + e); latch1.countDown(); } @Override public void onNext(String s) { System.out.println(System.currentTimeMillis() + " : s1 OnNext : " + s); value1.set(s); } }); Subscription s2 = response2 .doOnUnsubscribe(new Action0() { @Override public void call() { System.out.println(System.currentTimeMillis() + " : s2 Unsubscribed!"); latch2.countDown(); } }) .subscribe(new Subscriber<String>() { @Override public void onCompleted() { System.out.println(System.currentTimeMillis() + " : s2 OnCompleted"); latch2.countDown(); } @Override public void onError(Throwable e) { System.out.println(System.currentTimeMillis() + " : s2 OnError : " + e); latch2.countDown(); } @Override public void onNext(String s) { System.out.println(System.currentTimeMillis() + " : s2 OnNext : " + s); value2.set(s); } }); Subscription s3 = response3 .doOnUnsubscribe(new Action0() { @Override public void call() { System.out.println(System.currentTimeMillis() + " : s3 Unsubscribed!"); latch3.countDown(); } }) .subscribe(new Subscriber<String>() { @Override public void onCompleted() { System.out.println(System.currentTimeMillis() + " : s3 OnCompleted"); latch3.countDown(); } @Override public void onError(Throwable e) { System.out.println(System.currentTimeMillis() + " : s3 OnError : " + e); latch3.countDown(); } @Override public void onNext(String s) { System.out.println(System.currentTimeMillis() + " : s3 OnNext : " + s); value3.set(s); } }); s1.unsubscribe(); s3.unsubscribe(); timer.incrementTime(10); assertTrue(latch1.await(1000, TimeUnit.MILLISECONDS)); assertTrue(latch2.await(1000, TimeUnit.MILLISECONDS)); assertNull(value1.get()); assertEquals("foo", value2.get()); assertNull(value3.get()); System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); assertEquals(1, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size()); HystrixInvokableInfo<?> command = HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().iterator().next(); assertCommandExecutionEvents(command, HystrixEventType.EMIT, HystrixEventType.SUCCESS, HystrixEventType.COLLAPSED); assertEquals(1, command.getNumberCollapsed()); }
@Test public void testRequestThenTwoCacheHitsAllUnsubscribed() throws Exception { TestCollapserTimer timer = new TestCollapserTimer(); HystrixObservableCollapser<String, String, String, String> collapser1 = new SuccessfulCacheableCollapsedCommand(timer, "foo", true); Observable<String> response1 = collapser1.observe(); HystrixObservableCollapser<String, String, String, String> collapser2 = new SuccessfulCacheableCollapsedCommand(timer, "foo", true); Observable<String> response2 = collapser2.observe(); HystrixObservableCollapser<String, String, String, String> collapser3 = new SuccessfulCacheableCollapsedCommand(timer, "foo", true); Observable<String> response3 = collapser3.observe(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); final CountDownLatch latch3 = new CountDownLatch(1); final AtomicReference<String> value1 = new AtomicReference<String>(null); final AtomicReference<String> value2 = new AtomicReference<String>(null); final AtomicReference<String> value3 = new AtomicReference<String>(null); Subscription s1 = response1 .doOnUnsubscribe(new Action0() { @Override public void call() { System.out.println(System.currentTimeMillis() + " : s1 Unsubscribed!"); latch1.countDown(); } }) .subscribe(new Subscriber<String>() { @Override public void onCompleted() { System.out.println(System.currentTimeMillis() + " : s1 OnCompleted"); latch1.countDown(); } @Override public void onError(Throwable e) { System.out.println(System.currentTimeMillis() + " : s1 OnError : " + e); latch1.countDown(); } @Override public void onNext(String s) { System.out.println(System.currentTimeMillis() + " : s1 OnNext : " + s); value1.set(s); } }); Subscription s2 = response2 .doOnUnsubscribe(new Action0() { @Override public void call() { System.out.println(System.currentTimeMillis() + " : s2 Unsubscribed!"); latch2.countDown(); } }) .subscribe(new Subscriber<String>() { @Override public void onCompleted() { System.out.println(System.currentTimeMillis() + " : s2 OnCompleted"); latch2.countDown(); } @Override public void onError(Throwable e) { System.out.println(System.currentTimeMillis() + " : s2 OnError : " + e); latch2.countDown(); } @Override public void onNext(String s) { System.out.println(System.currentTimeMillis() + " : s2 OnNext : " + s); value2.set(s); } }); Subscription s3 = response3 .doOnUnsubscribe(new Action0() { @Override public void call() { System.out.println(System.currentTimeMillis() + " : s3 Unsubscribed!"); latch3.countDown(); } }) .subscribe(new Subscriber<String>() { @Override public void onCompleted() { System.out.println(System.currentTimeMillis() + " : s3 OnCompleted"); latch3.countDown(); } @Override public void onError(Throwable e) { System.out.println(System.currentTimeMillis() + " : s3 OnError : " + e); latch3.countDown(); } @Override public void onNext(String s) { System.out.println(System.currentTimeMillis() + " : s3 OnNext : " + s); value3.set(s); } }); s1.unsubscribe(); s2.unsubscribe(); s3.unsubscribe(); timer.incrementTime(10); assertTrue(latch1.await(1000, TimeUnit.MILLISECONDS)); assertTrue(latch2.await(1000, TimeUnit.MILLISECONDS)); assertNull(value1.get()); assertNull(value2.get()); assertNull(value3.get()); System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); assertEquals(0, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size()); }