Fixed the application of the CANCELLED state to do it only when no te…

…rminals have been emitted so far
rwilliams-r7 · Aug 3, 2016 · 5b9bdae · 5b9bdae
1 parent 51531d2
commit 5b9bdae
Show file tree

Hide file tree

Showing 3 changed files with 30 additions and 20 deletions.
diff --git a/hystrix-core/src/main/java/com/netflix/hystrix/AbstractCommand.java b/hystrix-core/src/main/java/com/netflix/hystrix/AbstractCommand.java
@@ -363,19 +363,6 @@ public void call() {
     public Observable<R> toObservable() {
         final AbstractCommand<R> _cmd = this;
 
-        final Action1<R> scalarEmitCleanup = new Action1<R>() {
-            @Override
-            public void call(R r) {
-                if (commandIsScalar()) {
-                    if (_cmd.commandState.compareAndSet(CommandState.OBSERVABLE_CHAIN_CREATED, CommandState.TERMINAL)) {
-                        handleCommandEnd(false); //user code never ran
-                    } else if (_cmd.commandState.compareAndSet(CommandState.USER_CODE_EXECUTED, CommandState.TERMINAL)) {
-                        handleCommandEnd(true); //user code did run
-                    }
-                }
-            }
-        };
-
         //doOnCompleted handler already did all of the SUCCESS work
         //doOnError handler already did all of the FAILURE/TIMEOUT/REJECTION/BAD_REQUEST work
         final Action0 terminateCommandCleanup = new Action0() {
@@ -395,14 +382,18 @@ public void call() {
             @Override
             public void call() {
                 if (_cmd.commandState.compareAndSet(CommandState.OBSERVABLE_CHAIN_CREATED, CommandState.UNSUBSCRIBED)) {
-                    _cmd.eventNotifier.markEvent(HystrixEventType.CANCELLED, _cmd.commandKey);
-                    _cmd.executionResultAtTimeOfCancellation = _cmd.executionResult
-                            .addEvent((int) (System.currentTimeMillis() - _cmd.commandStartTimestamp), HystrixEventType.CANCELLED);
+                    if (!_cmd.executionResult.containsTerminalEvent()) {
+                        _cmd.eventNotifier.markEvent(HystrixEventType.CANCELLED, _cmd.commandKey);
+                        _cmd.executionResultAtTimeOfCancellation = _cmd.executionResult
+                                .addEvent((int) (System.currentTimeMillis() - _cmd.commandStartTimestamp), HystrixEventType.CANCELLED);
+                    }
                     handleCommandEnd(false); //user code never ran
                 } else if (_cmd.commandState.compareAndSet(CommandState.USER_CODE_EXECUTED, CommandState.UNSUBSCRIBED)) {
-                    _cmd.eventNotifier.markEvent(HystrixEventType.CANCELLED, _cmd.commandKey);
-                    _cmd.executionResultAtTimeOfCancellation = _cmd.executionResult
-                            .addEvent((int) (System.currentTimeMillis() - _cmd.commandStartTimestamp), HystrixEventType.CANCELLED);
+                    if (!_cmd.executionResult.containsTerminalEvent()) {
+                        _cmd.eventNotifier.markEvent(HystrixEventType.CANCELLED, _cmd.commandKey);
+                        _cmd.executionResultAtTimeOfCancellation = _cmd.executionResult
+                                .addEvent((int) (System.currentTimeMillis() - _cmd.commandStartTimestamp), HystrixEventType.CANCELLED);
+                    }
                     handleCommandEnd(true); //user code did run
                 }
             }
@@ -502,7 +493,6 @@ public Observable<R> call() {
                 }
 
                 return afterCache
-                        .doOnNext(scalarEmitCleanup)
                         .doOnTerminate(terminateCommandCleanup)     // perform cleanup once (either on normal terminal state (this line), or unsubscribe (next line))
                         .doOnUnsubscribe(unsubscribeCommandCleanup) // perform cleanup once
                         .doOnCompleted(fireOnCompletedHook);

diff --git a/hystrix-core/src/main/java/com/netflix/hystrix/ExecutionResult.java b/hystrix-core/src/main/java/com/netflix/hystrix/ExecutionResult.java
@@ -43,11 +43,16 @@ public class ExecutionResult {
     private static final HystrixEventType[] ALL_EVENT_TYPES = HystrixEventType.values();
     private static final int NUM_EVENT_TYPES = ALL_EVENT_TYPES.length;
     private static final BitSet EXCEPTION_PRODUCING_EVENTS = new BitSet(NUM_EVENT_TYPES);
+    private static final BitSet TERMINAL_EVENTS = new BitSet(NUM_EVENT_TYPES);
 
     static {
         for (HystrixEventType eventType: HystrixEventType.EXCEPTION_PRODUCING_EVENT_TYPES) {
             EXCEPTION_PRODUCING_EVENTS.set(eventType.ordinal());
         }
+
+        for (HystrixEventType eventType: HystrixEventType.TERMINAL_EVENT_TYPES) {
+            TERMINAL_EVENTS.set(eventType.ordinal());
+        }
     }
 
     public static class EventCounts {
@@ -353,6 +358,10 @@ public boolean executionOccurred() {
         return executionOccurred;
     }
 
+    public boolean containsTerminalEvent() {
+        return eventCounts.containsAnyOf(TERMINAL_EVENTS);
+    }
+
     @Override
     public String toString() {
         return "ExecutionResult{" +

diff --git a/hystrix-core/src/main/java/com/netflix/hystrix/HystrixEventType.java b/hystrix-core/src/main/java/com/netflix/hystrix/HystrixEventType.java
@@ -82,11 +82,22 @@ public static HystrixEventType from(HystrixRollingNumberEvent event) {
      */
     public final static List<HystrixEventType> EXCEPTION_PRODUCING_EVENT_TYPES = new ArrayList<HystrixEventType>();
 
+    /**
+     * List of events that are terminal
+     */
+    public final static List<HystrixEventType> TERMINAL_EVENT_TYPES = new ArrayList<HystrixEventType>();
+
     static {
         EXCEPTION_PRODUCING_EVENT_TYPES.add(BAD_REQUEST);
         EXCEPTION_PRODUCING_EVENT_TYPES.add(FALLBACK_FAILURE);
         EXCEPTION_PRODUCING_EVENT_TYPES.add(FALLBACK_MISSING);
         EXCEPTION_PRODUCING_EVENT_TYPES.add(FALLBACK_REJECTION);
+
+        for (HystrixEventType eventType: HystrixEventType.values()) {
+            if (eventType.isTerminal()) {
+                TERMINAL_EVENT_TYPES.add(eventType);
+            }
+        }
     }
 
     public enum ThreadPool {