HystrixThreadEventStream { public void executionDone(ExecutionResult executionResult, HystrixCommandKey commandKey, HystrixThreadPoolKey threadPoolKey) { HystrixCommandCompletion event = HystrixCommandCompletion.from(executionResult, commandKey, threadPoolKey); writeOnlyCommandCompletionSubject.onNext(event); } HystrixThreadEventStream(Thread thread); static HystrixThreadEventStream getInstance(); void shutdown(); void commandExecutionStarted(HystrixCommandKey commandKey, HystrixThreadPoolKey threadPoolKey,
                                        HystrixCommandProperties.ExecutionIsolationStrategy isolationStrategy, int currentConcurrency); void executionDone(ExecutionResult executionResult, HystrixCommandKey commandKey, HystrixThreadPoolKey threadPoolKey); void collapserResponseFromCache(HystrixCollapserKey collapserKey); void collapserBatchExecuted(HystrixCollapserKey collapserKey, int batchSize); @Override String toString();  }
@Test public void testThreadIsolatedSuccess() throws InterruptedException { CountDownLatch commandLatch = new CountDownLatch(1); CountDownLatch threadPoolLatch = new CountDownLatch(1); Subscriber<HystrixCommandCompletion> commandSubscriber = getLatchedSubscriber(commandLatch); readCommandStream.observe().take(1).subscribe(commandSubscriber); Subscriber<HystrixCommandCompletion> threadPoolSubscriber = getLatchedSubscriber(threadPoolLatch); readThreadPoolStream.observe().take(1).subscribe(threadPoolSubscriber); ExecutionResult result = ExecutionResult.from(HystrixEventType.SUCCESS).setExecutedInThread(); writeToStream.executionDone(result, commandKey, threadPoolKey); assertTrue(commandLatch.await(1000, TimeUnit.MILLISECONDS)); assertTrue(threadPoolLatch.await(1000, TimeUnit.MILLISECONDS)); }
@Test public void testSemaphoreIsolatedSuccess() throws Exception { CountDownLatch commandLatch = new CountDownLatch(1); CountDownLatch threadPoolLatch = new CountDownLatch(1); Subscriber<HystrixCommandCompletion> commandSubscriber = getLatchedSubscriber(commandLatch); readCommandStream.observe().take(1).subscribe(commandSubscriber); Subscriber<HystrixCommandCompletion> threadPoolSubscriber = getLatchedSubscriber(threadPoolLatch); readThreadPoolStream.observe().take(1).subscribe(threadPoolSubscriber); ExecutionResult result = ExecutionResult.from(HystrixEventType.SUCCESS); writeToStream.executionDone(result, commandKey, threadPoolKey); assertTrue(commandLatch.await(1000, TimeUnit.MILLISECONDS)); assertFalse(threadPoolLatch.await(1000, TimeUnit.MILLISECONDS)); }
@Test public void testThreadIsolatedFailure() throws Exception { CountDownLatch commandLatch = new CountDownLatch(1); CountDownLatch threadPoolLatch = new CountDownLatch(1); Subscriber<HystrixCommandCompletion> commandSubscriber = getLatchedSubscriber(commandLatch); readCommandStream.observe().take(1).subscribe(commandSubscriber); Subscriber<HystrixCommandCompletion> threadPoolSubscriber = getLatchedSubscriber(threadPoolLatch); readThreadPoolStream.observe().take(1).subscribe(threadPoolSubscriber); ExecutionResult result = ExecutionResult.from(HystrixEventType.FAILURE).setExecutedInThread(); writeToStream.executionDone(result, commandKey, threadPoolKey); assertTrue(commandLatch.await(1000, TimeUnit.MILLISECONDS)); assertTrue(threadPoolLatch.await(1000, TimeUnit.MILLISECONDS)); }
@Test public void testSemaphoreIsolatedFailure() throws Exception { CountDownLatch commandLatch = new CountDownLatch(1); CountDownLatch threadPoolLatch = new CountDownLatch(1); Subscriber<HystrixCommandCompletion> commandSubscriber = getLatchedSubscriber(commandLatch); readCommandStream.observe().take(1).subscribe(commandSubscriber); Subscriber<HystrixCommandCompletion> threadPoolSubscriber = getLatchedSubscriber(threadPoolLatch); readThreadPoolStream.observe().take(1).subscribe(threadPoolSubscriber); ExecutionResult result = ExecutionResult.from(HystrixEventType.FAILURE); writeToStream.executionDone(result, commandKey, threadPoolKey); assertTrue(commandLatch.await(1000, TimeUnit.MILLISECONDS)); assertFalse(threadPoolLatch.await(1000, TimeUnit.MILLISECONDS)); }
@Test public void testThreadIsolatedTimeout() throws Exception { CountDownLatch commandLatch = new CountDownLatch(1); CountDownLatch threadPoolLatch = new CountDownLatch(1); Subscriber<HystrixCommandCompletion> commandSubscriber = getLatchedSubscriber(commandLatch); readCommandStream.observe().take(1).subscribe(commandSubscriber); Subscriber<HystrixCommandCompletion> threadPoolSubscriber = getLatchedSubscriber(threadPoolLatch); readThreadPoolStream.observe().take(1).subscribe(threadPoolSubscriber); ExecutionResult result = ExecutionResult.from(HystrixEventType.TIMEOUT).setExecutedInThread(); writeToStream.executionDone(result, commandKey, threadPoolKey); assertTrue(commandLatch.await(1000, TimeUnit.MILLISECONDS)); assertTrue(threadPoolLatch.await(1000, TimeUnit.MILLISECONDS)); }
@Test public void testSemaphoreIsolatedTimeout() throws Exception { CountDownLatch commandLatch = new CountDownLatch(1); CountDownLatch threadPoolLatch = new CountDownLatch(1); Subscriber<HystrixCommandCompletion> commandSubscriber = getLatchedSubscriber(commandLatch); readCommandStream.observe().take(1).subscribe(commandSubscriber); Subscriber<HystrixCommandCompletion> threadPoolSubscriber = getLatchedSubscriber(threadPoolLatch); readThreadPoolStream.observe().take(1).subscribe(threadPoolSubscriber); ExecutionResult result = ExecutionResult.from(HystrixEventType.TIMEOUT); writeToStream.executionDone(result, commandKey, threadPoolKey); assertTrue(commandLatch.await(1000, TimeUnit.MILLISECONDS)); assertFalse(threadPoolLatch.await(1000, TimeUnit.MILLISECONDS)); }
@Test public void testThreadIsolatedBadRequest() throws Exception { CountDownLatch commandLatch = new CountDownLatch(1); CountDownLatch threadPoolLatch = new CountDownLatch(1); Subscriber<HystrixCommandCompletion> commandSubscriber = getLatchedSubscriber(commandLatch); readCommandStream.observe().take(1).subscribe(commandSubscriber); Subscriber<HystrixCommandCompletion> threadPoolSubscriber = getLatchedSubscriber(threadPoolLatch); readThreadPoolStream.observe().take(1).subscribe(threadPoolSubscriber); ExecutionResult result = ExecutionResult.from(HystrixEventType.BAD_REQUEST).setExecutedInThread(); writeToStream.executionDone(result, commandKey, threadPoolKey); assertTrue(commandLatch.await(1000, TimeUnit.MILLISECONDS)); assertTrue(threadPoolLatch.await(1000, TimeUnit.MILLISECONDS)); }
@Test public void testSemaphoreIsolatedBadRequest() throws Exception { CountDownLatch commandLatch = new CountDownLatch(1); CountDownLatch threadPoolLatch = new CountDownLatch(1); Subscriber<HystrixCommandCompletion> commandSubscriber = getLatchedSubscriber(commandLatch); readCommandStream.observe().take(1).subscribe(commandSubscriber); Subscriber<HystrixCommandCompletion> threadPoolSubscriber = getLatchedSubscriber(threadPoolLatch); readThreadPoolStream.observe().take(1).subscribe(threadPoolSubscriber); ExecutionResult result = ExecutionResult.from(HystrixEventType.BAD_REQUEST); writeToStream.executionDone(result, commandKey, threadPoolKey); assertTrue(commandLatch.await(1000, TimeUnit.MILLISECONDS)); assertFalse(threadPoolLatch.await(1000, TimeUnit.MILLISECONDS)); }
@Test public void testThreadRejectedCommand() throws Exception { CountDownLatch commandLatch = new CountDownLatch(1); CountDownLatch threadPoolLatch = new CountDownLatch(1); Subscriber<HystrixCommandCompletion> commandSubscriber = getLatchedSubscriber(commandLatch); readCommandStream.observe().take(1).subscribe(commandSubscriber); Subscriber<HystrixCommandCompletion> threadPoolSubscriber = getLatchedSubscriber(threadPoolLatch); readThreadPoolStream.observe().take(1).subscribe(threadPoolSubscriber); ExecutionResult result = ExecutionResult.from(HystrixEventType.THREAD_POOL_REJECTED); writeToStream.executionDone(result, commandKey, threadPoolKey); assertTrue(commandLatch.await(1000, TimeUnit.MILLISECONDS)); assertTrue(threadPoolLatch.await(1000, TimeUnit.MILLISECONDS)); }
@Test public void testSemaphoreRejectedCommand() throws Exception { CountDownLatch commandLatch = new CountDownLatch(1); CountDownLatch threadPoolLatch = new CountDownLatch(1); Subscriber<HystrixCommandCompletion> commandSubscriber = getLatchedSubscriber(commandLatch); readCommandStream.observe().take(1).subscribe(commandSubscriber); Subscriber<HystrixCommandCompletion> threadPoolSubscriber = getLatchedSubscriber(threadPoolLatch); readThreadPoolStream.observe().take(1).subscribe(threadPoolSubscriber); ExecutionResult result = ExecutionResult.from(HystrixEventType.SEMAPHORE_REJECTED); writeToStream.executionDone(result, commandKey, threadPoolKey); assertTrue(commandLatch.await(1000, TimeUnit.MILLISECONDS)); assertFalse(threadPoolLatch.await(1000, TimeUnit.MILLISECONDS)); }
@Test public void testThreadIsolatedResponseFromCache() throws Exception { CountDownLatch commandLatch = new CountDownLatch(1); CountDownLatch threadPoolLatch = new CountDownLatch(1); Subscriber<List<HystrixCommandCompletion>> commandListSubscriber = getLatchedSubscriber(commandLatch); readCommandStream.observe().buffer(500, TimeUnit.MILLISECONDS).take(1) .doOnNext(new Action1<List<HystrixCommandCompletion>>() { @Override public void call(List<HystrixCommandCompletion> hystrixCommandCompletions) { System.out.println("LIST : " + hystrixCommandCompletions); assertEquals(3, hystrixCommandCompletions.size()); } }) .subscribe(commandListSubscriber); Subscriber<HystrixCommandCompletion> threadPoolSubscriber = getLatchedSubscriber(threadPoolLatch); readThreadPoolStream.observe().take(1).subscribe(threadPoolSubscriber); ExecutionResult result = ExecutionResult.from(HystrixEventType.SUCCESS).setExecutedInThread(); ExecutionResult cache1 = ExecutionResult.from(HystrixEventType.RESPONSE_FROM_CACHE); ExecutionResult cache2 = ExecutionResult.from(HystrixEventType.RESPONSE_FROM_CACHE); writeToStream.executionDone(result, commandKey, threadPoolKey); writeToStream.executionDone(cache1, commandKey, threadPoolKey); writeToStream.executionDone(cache2, commandKey, threadPoolKey); assertTrue(commandLatch.await(1000, TimeUnit.MILLISECONDS)); assertTrue(threadPoolLatch.await(1000, TimeUnit.MILLISECONDS)); }
@Test public void testSemaphoreIsolatedResponseFromCache() throws Exception { CountDownLatch commandLatch = new CountDownLatch(1); CountDownLatch threadPoolLatch = new CountDownLatch(1); Subscriber<List<HystrixCommandCompletion>> commandListSubscriber = getLatchedSubscriber(commandLatch); readCommandStream.observe().buffer(500, TimeUnit.MILLISECONDS).take(1) .doOnNext(new Action1<List<HystrixCommandCompletion>>() { @Override public void call(List<HystrixCommandCompletion> hystrixCommandCompletions) { System.out.println("LIST : " + hystrixCommandCompletions); assertEquals(3, hystrixCommandCompletions.size()); } }) .subscribe(commandListSubscriber); Subscriber<HystrixCommandCompletion> threadPoolSubscriber = getLatchedSubscriber(threadPoolLatch); readThreadPoolStream.observe().take(1).subscribe(threadPoolSubscriber); ExecutionResult result = ExecutionResult.from(HystrixEventType.SUCCESS); ExecutionResult cache1 = ExecutionResult.from(HystrixEventType.RESPONSE_FROM_CACHE); ExecutionResult cache2 = ExecutionResult.from(HystrixEventType.RESPONSE_FROM_CACHE); writeToStream.executionDone(result, commandKey, threadPoolKey); writeToStream.executionDone(cache1, commandKey, threadPoolKey); writeToStream.executionDone(cache2, commandKey, threadPoolKey); assertTrue(commandLatch.await(1000, TimeUnit.MILLISECONDS)); assertFalse(threadPoolLatch.await(1000, TimeUnit.MILLISECONDS)); }
@Test public void testShortCircuit() throws Exception { CountDownLatch commandLatch = new CountDownLatch(1); CountDownLatch threadPoolLatch = new CountDownLatch(1); Subscriber<HystrixCommandCompletion> commandSubscriber = getLatchedSubscriber(commandLatch); readCommandStream.observe().take(1).subscribe(commandSubscriber); Subscriber<HystrixCommandCompletion> threadPoolSubscriber = getLatchedSubscriber(threadPoolLatch); readThreadPoolStream.observe().take(1).subscribe(threadPoolSubscriber); ExecutionResult result = ExecutionResult.from(HystrixEventType.SHORT_CIRCUITED); writeToStream.executionDone(result, commandKey, threadPoolKey); assertTrue(commandLatch.await(1000, TimeUnit.MILLISECONDS)); assertFalse(threadPoolLatch.await(1000, TimeUnit.MILLISECONDS)); }