Reactive State Machine Example

BeginUserLogin.java

import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import reactor.core.scheduler.Schedulers;
import java.util.HashMap;

/**
 * Service responsible for initiating user login processes.
 * <p>
 * This service interacts with the AuthenticationStateService to handle various
 * stages of user login, from initialization to event emission.
 * </p>
 */
@Service
public class BeginUserLogin {

    private final AuthenticationStateService service;

    /**
     * Logger for capturing run-time behavior and errors.
     */
    private static final org.slf4j.Logger LOGGER = org.slf4j.LoggerFactory.getLogger(BeginUserLogin.class);

    /**
     * Constructs a BeginUserLogin instance.
     *
     * @param service An instance of {@link AuthenticationStateService} used for managing authentication state.
     */
    public BeginUserLogin(AuthenticationStateService service) {
        this.service = service;
    }

    /**
     * Emits an AuthEvent populated with claims from the authenticated OIDC user.
     *
     * @param user An instance of the authenticated OIDC user.
     * @return A constructed {@link AuthEvent} containing details about the authenticated user.
     */
    public AuthEvent emitEventWithClaims(OidcUser user) {
        setPrincipalInContext(user);
        publishAuthEvent().subscribe(service.getEventSubscriber());
        return buildGetAccountSuccessEvent();
    }

    /**
     * Initiates the user login process by creating and emitting an AuthEvent.
     *
     * @return A constructed {@link AuthEvent} that signifies the start of the login process.
     */
    public AuthEvent beginUserLogin() {
        return buildBeginLoginEvent().apply(authEvent -> authEvent);
    }

    // Helper Methods

    /**
     * Updates the event context to include the authenticated OIDC user as the principal.
     *
     * @param user The authenticated OIDC user.
     */
    private void setPrincipalInContext(OidcUser user) {
        this.service.getEventSubscriber().getEventContext().get("_MODEL").put("_principal", user);
    }

    /**
     * Publishes an AuthEvent indicating the start of a user login operation.
     *
     * @return A {@link Mono} wrapping the newly created AuthEvent.
     */
    private Mono<AuthEvent> publishAuthEvent() {
        return handleUserLogin(beginUserLogin(), service.getEventSubscriber().getEventContext().get("_MODEL"))
                .subscribeOn(Schedulers.boundedElastic());
    }

    /**
     * Creates an AuthEvent to indicate a successful account retrieval.
     *
     * @return A constructed {@link AuthEvent} that signifies successful account retrieval.
     */
    private AuthEvent buildGetAccountSuccessEvent() {
        return new AuthEventDefinitionBuilder()
                .withEventType(AuthEventType.getEvent(AuthEventType.Type.GET_ACCOUNT_SUCCESS))
                .withFromState(AuthState.getState(AuthState.Type.GET_ACCOUNT_STARTED))
                .withToState(AuthState.getState(AuthState.Type.GET_ACCOUNT_COMPLETED))
                .build();
    }

    /**
     * Constructs an AuthEvent to initiate the user login process.
     *
     * @return A constructed {@link AuthEvent} that marks the initiation of a user login.
     */
    private AuthEvent buildBeginLoginEvent() {
        return new AuthEventDefinitionBuilder()
                .withEventType(AuthEventType.getEvent(AuthEventType.Type.BEGIN_LOGIN))
                .withFromState(AuthState.getState(AuthState.Type.LOGIN_STARTED))
                .withToState(AuthState.getState(AuthState.Type.GET_ACCOUNT_STARTED))
                .withSubscriber(service.getEventSubscriber())
                .withAction(this::emitNextEvent)
                .build();
    }

    /**
     * Manages the emission of the subsequent AuthEvent in the login process.
     * <p>
     * This method takes in the initiating AuthEvent and its context, processes it,
     * and emits the next AuthEvent in the sequence.
     * </p>
     *
     * @param event   The AuthEvent that initiated the process.
     * @param context The context (typically a HashMap) in which the AuthEvent was created.
     * @return A {@link Mono} wrapping the AuthEvent that will be emitted next.
     */
    private Mono<AuthEvent> emitNextEvent(AuthEvent event, HashMap<Object, Object> context) {
        return handleUserLogin(event, context)
                .doOnNext(service.getEventPublisher()::next)
                .thenReturn(event);
    }
}

ReactiveAuthMachine.java

package ai.phylos.machine.state;

import ai.phylos.machine.domain.authentication.observers.AuthenticationStateService;
import ai.phylos.machine.domain.authentication.processors.*;
import ai.phylos.machine.event.AuthEvent;
import ai.phylos.machine.event.Event;
import ai.phylos.machine.event.EventProcessor;
import ai.phylos.machine.state.types.AuthEventType;
import ai.phylos.machine.state.types.AuthState;
import ai.phylos.machine.state.types.EventType;
import ai.phylos.machine.state.types.StateType;
import reactor.core.publisher.Sinks;

import java.util.ArrayList;
import java.util.Map;

/**
 * Custom reactive state machine specialized for authentication.
 * This extends a generic reactive state machine and adds transitions specific to authentication processes.
 */
public class ReactiveAuthMachine<T extends EventType, S extends StateType, E extends Event<T, S>> extends ReactiveStateMachine<AuthEventType, AuthState, AuthEvent> {
    private final AuthenticationStateService service;

    /**
     * Constructor to set the initial state and authentication service.
     */
    public ReactiveAuthMachine(AuthenticationStateService authenticationStateService) {
        super(AuthState.getState(AuthState.Type.LOGIN_STARTED)); // Sets the initial state for the machine
        this.service = authenticationStateService;
        addTransitions(); // Adds transitions between states and associated processors
    }

    /**
     * Acquire method to handle state transitions based on incoming events.
     */
    @Override
    public void acquire(AuthState currentState, AuthEvent event) {
        super.acquire(currentState, event); // Handles state transition using the super class's acquire method
    }

    /**
     * Adds transitions between authentication states and associates them with event processors.
     */
    private void addTransitions() {
        // Define transitions and link them with processors
        addTransition(AuthState.getState(AuthState.Type.LOGIN_STARTED),
                AuthState.getState(AuthState.Type.GET_ACCOUNT_STARTED),
                new GetUserAccountProcessor(AuthEventType.getEvent(AuthEventType.Type.GET_ACCOUNT)));

        // ... more transitions ...

        addTransition(AuthState.getState(AuthState.Type.REDIRECTED_TO_HOME),
                AuthState.getState(AuthState.Type.LOGIN_COMPLETED),
                new LoginSuccessProcessor(AuthEventType.getEvent(AuthEventType.Type.LOGIN_SUCCESS)));
    }

    /**
     * Helper method to add a transition and associate it with an event processor.
     */
    public void addTransition(AuthState fromState, AuthState targetState, EventProcessor<AuthEventType, AuthState, AuthEvent>  processor) {
        register(fromState, targetState, processor, Sinks.one()); // Registers the transition with its associated processor
    }

    /**
     * Retrieves the event processors associated with the authentication service.
     */
    @Override
    public Map<String, ArrayList< EventProcessor<AuthEventType, AuthState, AuthEvent> >> getProcessors() {
        return (Map<String, ArrayList< EventProcessor<AuthEventType, AuthState, AuthEvent>  >>)
                this.service.getSession().getAttributes().get("_processors");
    }
}