RxJava 0.20 and Remove Deprecated Usage

mikedehaan · Sep 17, 2014 · de706a2 · de706a2
1 parent efb889b
commit de706a2
Show file tree

Hide file tree

Showing 14 changed files with 155 additions and 161 deletions.
diff --git a/hystrix-core/build.gradle b/hystrix-core/build.gradle
@@ -4,7 +4,7 @@ apply plugin: 'idea'
 
 dependencies {
     compile 'com.netflix.archaius:archaius-core:0.4.1'
-    compile 'com.netflix.rxjava:rxjava-core:0.18.2'
+    compile 'com.netflix.rxjava:rxjava-core:0.20.4'
     compile 'org.slf4j:slf4j-api:1.7.0'
     testCompile 'junit:junit-dep:4.10'
 }

diff --git a/hystrix-core/src/main/java/com/netflix/hystrix/HystrixCollapser.java b/hystrix-core/src/main/java/com/netflix/hystrix/HystrixCollapser.java
@@ -429,7 +429,7 @@ public ResponseType execute() {
      */
     public Future<ResponseType> queue() {
         final Observable<ResponseType> o = toObservable();
-        return o.toBlockingObservable().toFuture();
+        return o.toBlocking().toFuture();
     }
 
     /**

diff --git a/hystrix-core/src/main/java/com/netflix/hystrix/HystrixExecutableBase.java b/hystrix-core/src/main/java/com/netflix/hystrix/HystrixExecutableBase.java
@@ -339,7 +339,7 @@ public Future<R> queue() {
          * is going to sit waiting on it.
          */
         final ObservableCommand<R> o = toObservable(Schedulers.immediate(), false);
-        final Future<R> f = o.toBlockingObservable().toFuture();
+        final Future<R> f = o.toBlocking().toFuture();
 
         /* special handling of error states that throw immediately */
         if (f.isDone()) {

diff --git a/hystrix-core/src/main/java/com/netflix/hystrix/HystrixObservableCollapser.java b/hystrix-core/src/main/java/com/netflix/hystrix/HystrixObservableCollapser.java
@@ -440,7 +440,7 @@ public ResponseType execute() {
      */
     public Future<ResponseType> queue() {
         final Observable<ResponseType> o = toObservable();
-        return o.toBlockingObservable().toFuture();
+        return o.toBlocking().toFuture();
     }
 
     /**

diff --git a/hystrix-core/src/main/java/com/netflix/hystrix/HystrixObservableCommand.java b/hystrix-core/src/main/java/com/netflix/hystrix/HystrixObservableCommand.java
@@ -839,7 +839,7 @@ public int getIntervalTimeInMilliseconds() {
             /**
              * If this subscriber receives values it means the parent succeeded/completed
              */
-            return new Subscriber<R>(s) {
+            Subscriber<R> parent = new Subscriber<R>() {
 
                 @Override
                 public void onCompleted() {
@@ -873,6 +873,11 @@ private boolean isNotTimedOut() {
                 }
 
             };
+
+            // if s is unsubscribed we want to unsubscribe the parent
+            s.add(parent);
+
+            return parent;
         }
 
     }

diff --git a/hystrix-core/src/test/java/com/netflix/hystrix/HystrixCircuitBreakerTest.java b/hystrix-core/src/test/java/com/netflix/hystrix/HystrixCircuitBreakerTest.java
@@ -574,7 +574,7 @@ protected String getFallback() {
     public class MyHystrixCommandExecutionHook extends HystrixCommandExecutionHook {
 
         @Override
-        public <T> T onComplete(final HystrixCommand<T> command, final T response) {
+        public <T> T onComplete(final HystrixExecutable<T> command, final T response) {
 
             logHC(command, response);
 
@@ -583,16 +583,17 @@ public <T> T onComplete(final HystrixCommand<T> command, final T response) {
 
         private int counter = 0;
 
-        private <T> void logHC(HystrixCommand<T> command, T response) {
+        private <T> void logHC(HystrixExecutable<T> command, T response) {
 
-            //if ((counter++ % 20) == 0) {
-            HystrixCommandMetrics metrics = command.getMetrics();
+            if(command instanceof HystrixExecutableInfo) {
+                HystrixExecutableInfo<T> commandInfo = (HystrixExecutableInfo<T>)command;
+            HystrixCommandMetrics metrics = commandInfo.getMetrics();
             System.out.println("cb/error-count/%/total: "
-                    + command.isCircuitBreakerOpen() + " "
+                    + commandInfo.isCircuitBreakerOpen() + " "
                     + metrics.getHealthCounts().getErrorCount() + " "
                     + metrics.getHealthCounts().getErrorPercentage() + " "
-                    + metrics.getHealthCounts().getTotalRequests() + "  => " + response + "  " + command.getExecutionEvents());
-            //}
+                    + metrics.getHealthCounts().getTotalRequests() + "  => " + response + "  " + commandInfo.getExecutionEvents());
+            }
         }
     }
 }
diff --git a/hystrix-core/src/test/java/com/netflix/hystrix/HystrixCollapserTest.java b/hystrix-core/src/test/java/com/netflix/hystrix/HystrixCollapserTest.java
@@ -62,7 +62,7 @@ public void testTwoRequests() throws Exception {
 
         assertEquals(1, counter.get());
 
-        assertEquals(1, HystrixRequestLog.getCurrentRequest().getExecutedCommands().size());
+        assertEquals(1, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size());
     }
 
     @Test
@@ -85,7 +85,7 @@ public void testMultipleBatches() throws Exception {
 
         // we should have had it execute twice now
         assertEquals(2, counter.get());
-        assertEquals(2, HystrixRequestLog.getCurrentRequest().getExecutedCommands().size());
+        assertEquals(2, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size());
     }
 
     @Test
@@ -102,7 +102,7 @@ public void testMaxRequestsInBatch() throws Exception {
 
         // we should have had it execute twice because the batch size was 2
         assertEquals(2, counter.get());
-        assertEquals(2, HystrixRequestLog.getCurrentRequest().getExecutedCommands().size());
+        assertEquals(2, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size());
     }
 
     @Test
@@ -131,7 +131,7 @@ public void testRequestsOverTime() throws Exception {
 
         System.out.println("number of executions: " + counter.get());
         assertEquals(3, counter.get());
-        assertEquals(3, HystrixRequestLog.getCurrentRequest().getExecutedCommands().size());
+        assertEquals(3, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size());
     }
 
     @Test
@@ -157,7 +157,7 @@ public void testUnsubscribeOnOneDoesntKillBatch() throws Exception {
 
         assertEquals(1, counter.get());
 
-        assertEquals(1, HystrixRequestLog.getCurrentRequest().getExecutedCommands().size());
+        assertEquals(1, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size());
     }
 
     @Test
@@ -176,7 +176,7 @@ public void testShardedRequests() throws Exception {
 
         /* we should get 2 batches since it gets sharded */
         assertEquals(2, counter.get());
-        assertEquals(2, HystrixRequestLog.getCurrentRequest().getExecutedCommands().size());
+        assertEquals(2, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size());
     }
 
     @Test
@@ -200,7 +200,7 @@ public void testRequestScope() throws Exception {
 
         // 2 different batches should execute, 1 per request
         assertEquals(2, counter.get());
-        assertEquals(2, HystrixRequestLog.getCurrentRequest().getExecutedCommands().size());
+        assertEquals(2, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size());
     }
 
     @Test
@@ -224,7 +224,7 @@ public void testGlobalScope() throws Exception {
 
         // despite having cleared the cache in between we should have a single execution because this is on the global not request cache
         assertEquals(1, counter.get());
-        assertEquals(1, HystrixRequestLog.getCurrentRequest().getExecutedCommands().size());
+        assertEquals(1, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size());
     }
 
     @Test
@@ -248,7 +248,7 @@ public void testErrorHandlingViaFutureException() throws Exception {
         }
 
         assertEquals(0, counter.get());
-        assertEquals(0, HystrixRequestLog.getCurrentRequest().getExecutedCommands().size());
+        assertEquals(0, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size());
     }
 
     @Test
@@ -274,7 +274,7 @@ public void testErrorHandlingWhenMapToResponseFails() throws Exception {
         // the batch failed so no executions
         assertEquals(0, counter.get());
         // but it still executed the command once
-        assertEquals(1, HystrixRequestLog.getCurrentRequest().getExecutedCommands().size());
+        assertEquals(1, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size());
     }
 
     @Test
@@ -441,9 +441,9 @@ public void testRequestCache1() {
 
         // we should still have executed only one command
         assertEquals(1, counter.get());
-        assertEquals(1, HystrixRequestLog.getCurrentRequest().getExecutedCommands().size());
+        assertEquals(1, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size());
 
-        HystrixCommand<?> command = HystrixRequestLog.getCurrentRequest().getExecutedCommands().toArray(new HystrixCommand<?>[1])[0];
+        HystrixExecutableInfo<?> command = HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().toArray(new HystrixExecutableInfo<?>[1])[0];
         System.out.println("command.getExecutionEvents(): " + command.getExecutionEvents());
         assertEquals(2, command.getExecutionEvents().size());
         assertTrue(command.getExecutionEvents().contains(HystrixEventType.SUCCESS));
@@ -493,9 +493,9 @@ public void testRequestCache2() {
 
         // we should still have executed only one command
         assertEquals(1, counter.get());
-        assertEquals(1, HystrixRequestLog.getCurrentRequest().getExecutedCommands().size());
+        assertEquals(1, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size());
 
-        HystrixCommand<?> command = HystrixRequestLog.getCurrentRequest().getExecutedCommands().toArray(new HystrixCommand<?>[1])[0];
+        HystrixExecutableInfo<?> command = HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().toArray(new HystrixExecutableInfo<?>[1])[0];
         assertEquals(2, command.getExecutionEvents().size());
         assertTrue(command.getExecutionEvents().contains(HystrixEventType.SUCCESS));
         assertTrue(command.getExecutionEvents().contains(HystrixEventType.COLLAPSED));
@@ -549,9 +549,9 @@ public void testRequestCache3() {
 
         // we should still have executed only one command
         assertEquals(1, counter.get());
-        assertEquals(1, HystrixRequestLog.getCurrentRequest().getExecutedCommands().size());
+        assertEquals(1, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size());
 
-        HystrixCommand<?> command = HystrixRequestLog.getCurrentRequest().getExecutedCommands().toArray(new HystrixCommand<?>[1])[0];
+        HystrixExecutableInfo<?> command = HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().toArray(new HystrixExecutableInfo<?>[1])[0];
         assertEquals(2, command.getExecutionEvents().size());
         assertTrue(command.getExecutionEvents().contains(HystrixEventType.SUCCESS));
         assertTrue(command.getExecutionEvents().contains(HystrixEventType.COLLAPSED));
@@ -605,16 +605,16 @@ public void testNoRequestCache3() {
 
         // request caching is turned off on this so we expect 2 command executions
         assertEquals(2, counter.get());
-        assertEquals(2, HystrixRequestLog.getCurrentRequest().getExecutedCommands().size());
+        assertEquals(2, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size());
 
         // we expect to see it with SUCCESS and COLLAPSED and both
-        HystrixCommand<?> commandA = HystrixRequestLog.getCurrentRequest().getExecutedCommands().toArray(new HystrixCommand<?>[2])[0];
+        HystrixExecutableInfo<?> commandA = HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().toArray(new HystrixExecutableInfo<?>[2])[0];
         assertEquals(2, commandA.getExecutionEvents().size());
         assertTrue(commandA.getExecutionEvents().contains(HystrixEventType.SUCCESS));
         assertTrue(commandA.getExecutionEvents().contains(HystrixEventType.COLLAPSED));
 
         // we expect to see it with SUCCESS and COLLAPSED and both
-        HystrixCommand<?> commandB = HystrixRequestLog.getCurrentRequest().getExecutedCommands().toArray(new HystrixCommand<?>[2])[1];
+        HystrixExecutableInfo<?> commandB = HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().toArray(new HystrixExecutableInfo<?>[2])[1];
         assertEquals(2, commandB.getExecutionEvents().size());
         assertTrue(commandB.getExecutionEvents().contains(HystrixEventType.SUCCESS));
         assertTrue(commandB.getExecutionEvents().contains(HystrixEventType.COLLAPSED));
@@ -675,9 +675,9 @@ public void testRequestCacheWithException() {
 
         // it should still be 1 ... no new executions
         assertEquals(1, commands.size());
-        assertEquals(1, HystrixRequestLog.getCurrentRequest().getExecutedCommands().size());
+        assertEquals(1, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size());
 
-        HystrixCommand<?> command = HystrixRequestLog.getCurrentRequest().getExecutedCommands().toArray(new HystrixCommand<?>[1])[0];
+        HystrixExecutableInfo<?> command = HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().toArray(new HystrixExecutableInfo<?>[1])[0];
         assertEquals(2, command.getExecutionEvents().size());
         assertTrue(command.getExecutionEvents().contains(HystrixEventType.FAILURE));
         assertTrue(command.getExecutionEvents().contains(HystrixEventType.COLLAPSED));
@@ -737,7 +737,7 @@ public void testRequestCacheWithTimeout() {
 
         // it should still be 1 ... no new executions
         assertEquals(1, commands.size());
-        assertEquals(1, HystrixRequestLog.getCurrentRequest().getExecutedCommands().size());
+        assertEquals(1, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size());
     }
 
     /**
@@ -767,7 +767,7 @@ public void testRequestWithCommandShortCircuited() throws Exception {
 
         assertEquals(0, counter.get());
         // it will execute once (short-circuited)
-        assertEquals(1, HystrixRequestLog.getCurrentRequest().getExecutedCommands().size());
+        assertEquals(1, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size());
     }
 
     /**
@@ -789,7 +789,7 @@ public void testVoidResponseTypeFireAndForgetCollapsing1() throws Exception {
 
         assertEquals(1, counter.get());
 
-        assertEquals(1, HystrixRequestLog.getCurrentRequest().getExecutedCommands().size());
+        assertEquals(1, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size());
     }
 
     /**
@@ -815,7 +815,7 @@ public void testVoidResponseTypeFireAndForgetCollapsing2() throws Exception {
 
         assertEquals(1, counter.get());
 
-        assertEquals(1, HystrixRequestLog.getCurrentRequest().getExecutedCommands().size());
+        assertEquals(1, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size());
     }
 
     /**
@@ -830,7 +830,7 @@ public void testVoidResponseTypeFireAndForgetCollapsing3() throws Exception {
 
         assertEquals(1, counter.get());
 
-        assertEquals(1, HystrixRequestLog.getCurrentRequest().getExecutedCommands().size());
+        assertEquals(1, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size());
     }
 
     private static class TestRequestCollapser extends HystrixCollapser<List<String>, String, String> {