docs: public Observable<R> toObservable() {
          final AbstractCommand<R> _cmd = this;
  
          //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() {
  
              @Override
              public void call() {
                  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
                  }
              }
          };
  
          //mark the command as CANCELLED and store the latency (in addition to standard cleanup)
          final Action0 unsubscribeCommandCleanup = new Action0() {
              @Override
              public void call() {
                  circuitBreaker.markNonSuccess();
                  if (_cmd.commandState.compareAndSet(CommandState.OBSERVABLE_CHAIN_CREATED, CommandState.UNSUBSCRIBED)) {
                      if (!_cmd.executionResult.containsTerminalEvent()) {
                          _cmd.eventNotifier.markEvent(HystrixEventType.CANCELLED, _cmd.commandKey);
                          try {
                              executionHook.onUnsubscribe(_cmd);
                          } catch (Throwable hookEx) {
                              logger.warn("Error calling HystrixCommandExecutionHook.onUnsubscribe", hookEx);
                          }
                          _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)) {
                      if (!_cmd.executionResult.containsTerminalEvent()) {
                          _cmd.eventNotifier.markEvent(HystrixEventType.CANCELLED, _cmd.commandKey);
                          try {
                              executionHook.onUnsubscribe(_cmd);
                          } catch (Throwable hookEx) {
                              logger.warn("Error calling HystrixCommandExecutionHook.onUnsubscribe", hookEx);
                          }
                          _cmd.executionResultAtTimeOfCancellation = _cmd.executionResult
                                  .addEvent((int) (System.currentTimeMillis() - _cmd.commandStartTimestamp), HystrixEventType.CANCELLED);
                      }
                      handleCommandEnd(true); //user code did run
                  }
              }
          };
  
          final Func0<Observable<R>> applyHystrixSemantics = new Func0<Observable<R>>() {
              @Override
              public Observable<R> call() {
                  if (commandState.get().equals(CommandState.UNSUBSCRIBED)) {
                      return Observable.never();
                  }
                  return applyHystrixSemantics(_cmd);
              }
          };
  
          final Func1<R, R> wrapWithAllOnNextHooks = new Func1<R, R>() {
              @Override
              public R call(R r) {
                  R afterFirstApplication = r;
  
                  try {
                      afterFirstApplication = executionHook.onComplete(_cmd, r);
                  } catch (Throwable hookEx) {
                      logger.warn("Error calling HystrixCommandExecutionHook.onComplete", hookEx);
                  }
  
                  try {
                      return executionHook.onEmit(_cmd, afterFirstApplication);
                  } catch (Throwable hookEx) {
                      logger.warn("Error calling HystrixCommandExecutionHook.onEmit", hookEx);
                      return afterFirstApplication;
                  }
              }
          };
  
          final Action0 fireOnCompletedHook = new Action0() {
              @Override
              public void call() {
                  try {
                      executionHook.onSuccess(_cmd);
                  } catch (Throwable hookEx) {
                      logger.warn("Error calling HystrixCommandExecutionHook.onSuccess", hookEx);
                  }
              }
          };
  
          return Observable.defer(new Func0<Observable<R>>() {
              @Override
              public Observable<R> call() {
                   /* this is a stateful object so can only be used once */
                  if (!commandState.compareAndSet(CommandState.NOT_STARTED, CommandState.OBSERVABLE_CHAIN_CREATED)) {
                      IllegalStateException ex = new IllegalStateException("This instance can only be executed once. Please instantiate a new instance.");
                      //TODO make a new error type for this
                      throw new HystrixRuntimeException(FailureType.BAD_REQUEST_EXCEPTION, _cmd.getClass(), getLogMessagePrefix() + " command executed multiple times - this is not permitted.", ex, null);
                  }
  
                  commandStartTimestamp = System.currentTimeMillis();
  
                  if (properties.requestLogEnabled().get()) {
                      // log this command execution regardless of what happened
                      if (currentRequestLog != null) {
                          currentRequestLog.addExecutedCommand(_cmd);
                      }
                  }
  
                  final boolean requestCacheEnabled = isRequestCachingEnabled();
                  final String cacheKey = getCacheKey();
  
                  /* try from cache first */
                  if (requestCacheEnabled) {
                      HystrixCommandResponseFromCache<R> fromCache = (HystrixCommandResponseFromCache<R>) requestCache.get(cacheKey);
                      if (fromCache != null) {
                          isResponseFromCache = true;
                          return handleRequestCacheHitAndEmitValues(fromCache, _cmd);
                      }
                  }
  
                  Observable<R> hystrixObservable =
                          Observable.defer(applyHystrixSemantics)
                                  .map(wrapWithAllOnNextHooks);
  
                  Observable<R> afterCache;
  
                  // put in cache
                  if (requestCacheEnabled && cacheKey != null) {
                      // wrap it for caching
                      HystrixCachedObservable<R> toCache = HystrixCachedObservable.from(hystrixObservable, _cmd);
                      HystrixCommandResponseFromCache<R> fromCache = (HystrixCommandResponseFromCache<R>) requestCache.putIfAbsent(cacheKey, toCache);
                      if (fromCache != null) {
                          // another thread beat us so we'll use the cached value instead
                          toCache.unsubscribe();
                          isResponseFromCache = true;
                          return handleRequestCacheHitAndEmitValues(fromCache, _cmd);
                      } else {
                          // we just created an ObservableCommand so we cast and return it
                          afterCache = toCache.toObservable();
                      }
                  } else {
                      afterCache = hystrixObservable;
                  }
  
                  return afterCache
                          .doOnTerminate(terminateCommandCleanup)     // perform cleanup once (either on normal terminal state (this line), or unsubscribe (next line))
                          .doOnUnsubscribe(unsubscribeCommandCleanup) // perform cleanup once
                          .doOnCompleted(fireOnCompletedHook);
              }
          });
      }