HystrixCollapser implements HystrixExecutable<ResponseType>, 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  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 testRequestWithCommandShortCircuited() throws Exception { TestCollapserTimer timer = new TestCollapserTimer(); HystrixCollapser<List<String>, String, String> collapser1 = new TestRequestCollapserWithShortCircuitedCommand(timer, "1"); Observable<String> response1 = collapser1.observe(); Observable<String> response2 = new TestRequestCollapserWithShortCircuitedCommand(timer, "2").observe(); timer.incrementTime(10); try { response1.toBlocking().first(); fail("we should have received an exception"); } catch (Exception e) { e.printStackTrace(); } try { response2.toBlocking().first(); fail("we should have received an exception"); } catch (Exception e) { e.printStackTrace(); } assertEquals(1, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size()); Iterator<HystrixInvokableInfo<?>> cmdIterator = HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().iterator(); assertEquals(2, cmdIterator.next().getNumberCollapsed()); }
@Test public void testEarlyUnsubscribeFromAllCancelsBatch() throws Exception { TestCollapserTimer timer = new TestCollapserTimer(); HystrixCollapser<List<String>, String, String> collapser1 = new TestRequestCollapser(timer, 1); Observable<String> response1 = collapser1.observe(); HystrixCollapser<List<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(); HystrixCollapser<List<String>, String, String> collapser1 = new SuccessfulCacheableCollapsedCommand(timer, "foo", true); Observable<String> response1 = collapser1.observe(); HystrixCollapser<List<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.SUCCESS, HystrixEventType.COLLAPSED); assertEquals(1, command.getNumberCollapsed()); }
@Test public void testRequestThenCacheHitAndOriginalUnsubscribed() throws Exception { TestCollapserTimer timer = new TestCollapserTimer(); HystrixCollapser<List<String>, String, String> collapser1 = new SuccessfulCacheableCollapsedCommand(timer, "foo", true); Observable<String> response1 = collapser1.observe(); HystrixCollapser<List<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.SUCCESS, HystrixEventType.COLLAPSED); assertEquals(1, command.getNumberCollapsed()); }
@Test public void testRequestThenTwoCacheHitsOriginalAndOneCacheHitUnsubscribed() throws Exception { TestCollapserTimer timer = new TestCollapserTimer(); HystrixCollapser<List<String>, String, String> collapser1 = new SuccessfulCacheableCollapsedCommand(timer, "foo", true); Observable<String> response1 = collapser1.observe(); HystrixCollapser<List<String>, String, String> collapser2 = new SuccessfulCacheableCollapsedCommand(timer, "foo", true); Observable<String> response2 = collapser2.observe(); HystrixCollapser<List<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.SUCCESS, HystrixEventType.COLLAPSED); assertEquals(1, command.getNumberCollapsed()); }
@Test public void testRequestThenTwoCacheHitsAllUnsubscribed() throws Exception { TestCollapserTimer timer = new TestCollapserTimer(); HystrixCollapser<List<String>, String, String> collapser1 = new SuccessfulCacheableCollapsedCommand(timer, "foo", true); Observable<String> response1 = collapser1.observe(); HystrixCollapser<List<String>, String, String> collapser2 = new SuccessfulCacheableCollapsedCommand(timer, "foo", true); Observable<String> response2 = collapser2.observe(); HystrixCollapser<List<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()); }