Command Execution and Observation Pipeline

The command execution process in Hystrix involves several key components that work together to ensure proper execution, error handling, and isolation. The core method responsible for orchestrating this process is executeCommandAndObserve.

This method defines various callback handlers for different stages of command execution:

• markEmits: Executed before command execusion
• markOnCompleted: Executed after successful command completion
• handleFallback: Provides fallback logic when command execution fails
• setRequestContext: Sets the request context for each emitted value

The method then obtains the execution Observable by calling executeCommandWithSpecifiedIsolation and applies timeout functionality if enabled.

private Observable<R> executeCommandAndObserve(final AbstractCommand<R> _cmd) {
    final HystrixRequestContext currentRequestContext = HystrixRequestContext.getContextForCurrentThread();

    // Callback for pre-execution operations
    final Action1<R> markEmits = new Action1<R>() {
        @Override
        public void call(R r) {
            // Implementation details
        }
    };
        
    // Callback for post-completion operations
    final Action0 markOnCompleted = new Action0() {
        @Override
        public void call() {
            // Implementation details
        }
    };
        
    // Fallback logic for error scenarios
    final Func1<Throwable, Observable<R>> handleFallback = new Func1<Throwable, Observable<R>>() {
        @Override
        public Observable<R> call(Throwable throwable) {
            // Implementation details
            return Observable.error(throwable);
        }
    };

    // Request context setting for each emitted value
    final Action1<Notification<? super R>> setRequestContext = new Action1<Notification<? super R>>() {
        @Override
        public void call(Notification<? super R> notification) {
            // Implementation details
        }
    };

    Observable<R> execution;
    if (properties.executionTimeoutEnabled().get()) {
        execution = executeCommandWithSpecifiedIsolation(_cmd)
            .lift(new HystrixObservableTimeoutOperator<R>(_cmd));
    } else {
        execution = executeCommandWithSpecifiedIsolation(_cmd);
    }

    return execution.doOnNext(markEmits)
        .doOnCompleted(markOnCompleted)
        .onErrorResumeNext(handleFallback)
        .doOnEach(setRequestContext);
}

Isolation Strategy Implementation

The executeCommandWithSpecifiedIsolation method determines the execution path based on the configured isolation strategy (THREAD or SEMAPHORE).

private Observable<R> executeCommandWithSpecifiedIsolation(final AbstractCommand<R> _cmd) {
    if (properties.executionIsolationStrategy().get() == ExecutionIsolationStrategy.THREAD) {
        // Thread-based isolation implementation
        return Observable.defer(new Func0<Observable<R>>() {
            @Override
            public Observable<R> call() {
                // Thread pool and execution logic
            }
        });
    } else {
        // Semaphore-based isolation implementation
        return Observable.defer(new Func0<Observable<R>>() {
            @Override
            public Observable<R> call() {
                try {
                    executionHook.onRunStart(_cmd);
                    executionHook.onExecutionStart(_cmd);
                    return getUserExecutionObservable(_cmd);
                } catch (Throwable ex) {
                    return Observable.error(ex);
                }
            }
        });
    }
}

User Execution Observable Creation

The getUserExecutionObservable method creates the Observable that represents the actual user command execution. It delegates to the abstract getExecutionObservable method which is implemented by concrete command classes.

private Observable<R> getUserExecutionObservable(final AbstractCommand<R> _cmd) {
    Observable<R> userObservable;

    try {
        userObservable = getExecutionObservable();
    } catch (Throwable ex) {
        userObservable = Observable.error(ex);
    }

    return userObservable
        .lift(new ExecutionHookApplication(_cmd))
        .lift(new DeprecatedOnRunHookApplication(_cmd));
}

HystrixCommand Execution Flow

The HystrixCommand class implements the getExecutionObservable method which ultimately calls the user-defined run() method. This method is wrapped in an Observable using RxJava's just operator.

final protected Observable<R> getExecutionObservable() {
    return Observable.defer(new Func0<Observable<R>>() {
        @Override
        public Observable<R> call() {
            try {
                return Observable.just(run());
            } catch (Throwable ex) {
                return Observable.error(ex);
            }
        }
    }).doOnSubscribe(new Action0() {
        @Override
        public void call() {
            executionThread.set(Thread.currentThread());
        }
    });
}

Generic Command Implementation

The GenericCommand.run() method provides the actual business logic execution. It follows a similar pattern to custom HystrixCommand implementations by:

1. Retrieving the CommandAction
2. Executing the method with specified execution type

protected Object run() throws Exception {
    LOGGER.debug("Executing command: {}", getCommandKey().name());
    return process(new Action() {
        @Override
        Object execute() {
            return getCommandAction().execute(getExecutionType());
        }
    });
}