<commit_before>/*
 * Copyright 2013 <NAME>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.codebullets.sagalib.processing;

import com.codebullets.sagalib.DeadMessage;
import com.codebullets.sagalib.ExecutedRunnable;
import com.codebullets.sagalib.ExecutionContext;
import com.codebullets.sagalib.context.LookupContext;
import com.codebullets.sagalib.Saga;
import com.codebullets.sagalib.SagaLifetimeInterceptor;
import com.codebullets.sagalib.SagaModule;
import com.codebullets.sagalib.context.CurrentExecutionContext;
import com.codebullets.sagalib.context.NeedContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nullable;
import java.lang.reflect.InvocationTargetException;
import java.util.Collection;
import java.util.Map;

import static com.google.common.base.Preconditions.checkNotNull;

/**
 * Perform execution of saga message handling. This class is the execution
 * root unit when handling messages as part of an execution strategy.
 */
class SagaExecutionTask implements ExecutedRunnable {
    private static final Logger LOG = LoggerFactory.getLogger(SagaExecutionTask.class);

    private final HandlerInvoker invoker;
    private final SagaEnvironment env;
    private final LookupContext taskLookupContext;

    @Nullable
    private final ExecutionContext parentContext;

    /**
     * Generates a new instance of SagaExecutionTask.
     */
    public SagaExecutionTask(
            final SagaEnvironment environment,
            final HandlerInvoker invoker,
            final Object message,
            final Map<String, Object> headers,
            @Nullable final ExecutionContext parentContext) {
        this.parentContext = parentContext;
        this.taskLookupContext = new SagaLookupContext(message, headers, parentContext);
        this.env = environment;
        this.invoker = invoker;
    }

    /**
     * Performs synchronous saga handling of the message provided in ctor.
     *
     * @throws InvocationTargetException Thrown when invocation of the handler method fails.
     * @throws IllegalAccessException Thrown when access to the handler method fails.
     */
    public void handle() throws InvocationTargetException, IllegalAccessException {
        checkNotNull(taskLookupContext.message(), "Message to handle must not be null.");
        startExecutionChain(taskLookupContext);
    }

    private boolean startExecutionChain(final LookupContext messageLookupContext) throws InvocationTargetException, IllegalAccessException {
        boolean sagasExecuted = false;

        CurrentExecutionContext executionContext = env.contextProvider().get();
        executionContext.setMessage(messageLookupContext.message());
        executionContext.setParentContext(parentContext);
        setHeaders(executionContext);

        try {
            moduleStarts(executionContext);
            sagasExecuted = executeHandlersForMessage(messageLookupContext, executionContext);
            if (!sagasExecuted) {
                LOG.warn("No saga or saga state found to handle message. (message = {})", taskLookupContext.message());
            }
        } catch (Exception ex) {
            moduleError(executionContext, messageLookupContext.message(), ex);
            throw ex;
        } finally {
            moduleFinished(executionContext);
        }

        return sagasExecuted;
    }

    private boolean executeHandlersForMessage(final LookupContext messageLookupContext, final CurrentExecutionContext executionContext)
            throws InvocationTargetException, IllegalAccessException {
        boolean sagasExecuted = false;
        Object message = executionContext.message();

        Collection<SagaInstanceInfo> sagaDescriptions = env.instanceResolver().resolve(messageLookupContext);
        if (!sagaDescriptions.isEmpty()) {
            invokeSagas(executionContext, sagaDescriptions, message);
            sagasExecuted = true;
        } else if (!(message instanceof DeadMessage)) {
            DeadMessage deadMessage = new DeadMessage(taskLookupContext.message());
            LookupContext deadMessageContext = new SagaLookupContext(deadMessage, taskLookupContext, parentContext);
            sagasExecuted = startExecutionChain(deadMessageContext);
        }

        return sagasExecuted;
    }

    private void invokeSagas(final CurrentExecutionContext context, final Iterable<SagaInstanceInfo> sagaDescriptions, final Object invokeParam)
            throws InvocationTargetException, IllegalAccessException {
        for (SagaInstanceInfo sagaDescription : sagaDescriptions) {
            Saga saga = sagaDescription.getSaga();
            context.setSaga(saga);
            setSagaExecutionContext(saga, context);

            // call interceptor pre handling hooks
            interceptorStart(sagaDescription, context, invokeParam);
            interceptorHandling(saga, context, invokeParam);

            // perform actual saga invoke
            invoker.invoke(saga, invokeParam);

            // call interceptor handler finished hooks
            interceptorHandlingExecuted(saga, context, invokeParam);
            interceptorFinished(saga, context);
            updateStateStorage(sagaDescription);

            if (context.dispatchingStopped()) {
                LOG.debug("Handler dispatching stopped after invoking saga {}.", sagaDescription.getSaga().getClass().getSimpleName());
                break;
            }
        }
    }

    private void interceptorHandling(final Saga saga, final ExecutionContext context, final Object invokeParam) {
        for (SagaLifetimeInterceptor interceptor : env.interceptors()) {
            interceptor.onHandlerExecuting(saga, context, invokeParam);
        }
    }

    private void interceptorHandlingExecuted(final Saga saga, final ExecutionContext context, final Object invokeParam) {
        for (SagaLifetimeInterceptor interceptor : env.interceptors()) {
            interceptor.onHandlerExecuted(saga, context, invokeParam);
        }
    }

    private void interceptorFinished(final Saga saga, final ExecutionContext context) {
        if (saga.isFinished()) {
            for (SagaLifetimeInterceptor interceptor : env.interceptors()) {
                interceptor.onFinished(saga, context);
            }
        }
    }

    private void interceptorStart(final SagaInstanceInfo sagaDescription, final ExecutionContext context, final Object invokeParam) {
        if (sagaDescription.isStarting()) {
            for (SagaLifetimeInterceptor interceptor : env.interceptors()) {
                interceptor.onStarting(sagaDescription.getSaga(), context, invokeParam);
            }
        }
    }

    private void moduleError(final ExecutionContext context, final Object invokeParam, final Exception ex) {
        for (SagaModule module : env.modules()) {
            module.onError(context, invokeParam, ex);
        }
    }

    private void moduleStarts(final ExecutionContext context) {
        for (SagaModule module : env.modules()) {
            module.onStart(context);
        }
    }

    private void moduleFinished(final ExecutionContext context) {
        for (SagaModule module : env.modules()) {
            module.onFinished(context);
        }
    }

    private void setHeaders(final CurrentExecutionContext context) {
        for (String key : taskLookupContext.getHeaders()) {
            context.setHeaderValue(key, taskLookupContext.getHeaderValue(key));
        }
    }

    private void setSagaExecutionContext(final Saga saga, final ExecutionContext context) {
        if (saga instanceof NeedContext) {
            ((NeedContext) saga).setExecutionContext(context);
        }
    }

    /**
     * Updates the state storage depending on whether the saga is completed or keeps on running.
     */
    private void updateStateStorage(final SagaInstanceInfo description) {
        Saga saga = description.getSaga();

        // if saga has finished delete existing state and possible timeouts
        // if saga has just been created state has never been save and there
        // is no need to delete it.
        if (saga.isFinished() && !description.isStarting()) {
            env.storage().delete(saga.state().getSagaId());
            env.timeoutManager().cancelTimeouts(saga.state().getSagaId());
        } else if (!saga.isFinished()) {
            env.storage().save(saga.state());
        }
    }

    /**
     * Similar to {@link #handle()} but intended for execution on any thread.<p/>
     * May throw a runtime exception in case something went wrong invoking the target saga message handler.
     */
    @Override
    public void run() {
        try {
            handle();
        } catch (InvocationTargetException | IllegalAccessException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public Object message() {
        return taskLookupContext.message();
    }
}<commit_msg>A SagaModule should be able to stop dispatching of a message.<commit_after>/*
 * Copyright 2013 <NAME>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.codebullets.sagalib.processing;

import com.codebullets.sagalib.DeadMessage;
import com.codebullets.sagalib.ExecutedRunnable;
import com.codebullets.sagalib.ExecutionContext;
import com.codebullets.sagalib.context.LookupContext;
import com.codebullets.sagalib.Saga;
import com.codebullets.sagalib.SagaLifetimeInterceptor;
import com.codebullets.sagalib.SagaModule;
import com.codebullets.sagalib.context.CurrentExecutionContext;
import com.codebullets.sagalib.context.NeedContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nullable;
import java.lang.reflect.InvocationTargetException;
import java.util.Collection;
import java.util.Map;

import static com.google.common.base.Preconditions.checkNotNull;

/**
 * Perform execution of saga message handling. This class is the execution
 * root unit when handling messages as part of an execution strategy.
 */
class SagaExecutionTask implements ExecutedRunnable {
    private static final Logger LOG = LoggerFactory.getLogger(SagaExecutionTask.class);

    private final HandlerInvoker invoker;
    private final SagaEnvironment env;
    private final LookupContext taskLookupContext;

    @Nullable
    private final ExecutionContext parentContext;

    /**
     * Generates a new instance of SagaExecutionTask.
     */
    public SagaExecutionTask(
            final SagaEnvironment environment,
            final HandlerInvoker invoker,
            final Object message,
            final Map<String, Object> headers,
            @Nullable final ExecutionContext parentContext) {
        this.parentContext = parentContext;
        this.taskLookupContext = new SagaLookupContext(message, headers, parentContext);
        this.env = environment;
        this.invoker = invoker;
    }

    /**
     * Performs synchronous saga handling of the message provided in ctor.
     *
     * @throws InvocationTargetException Thrown when invocation of the handler method fails.
     * @throws IllegalAccessException Thrown when access to the handler method fails.
     */
    public void handle() throws InvocationTargetException, IllegalAccessException {
        checkNotNull(taskLookupContext.message(), "Message to handle must not be null.");
        startExecutionChain(taskLookupContext);
    }

    private boolean startExecutionChain(final LookupContext messageLookupContext) throws InvocationTargetException, IllegalAccessException {
        boolean sagasExecuted = false;

        CurrentExecutionContext executionContext = env.contextProvider().get();
        executionContext.setMessage(messageLookupContext.message());
        executionContext.setParentContext(parentContext);
        setHeaders(executionContext);

        try {
            moduleStarts(executionContext);
            sagasExecuted = executeHandlersForMessage(messageLookupContext, executionContext);
            if (!sagasExecuted) {
                LOG.warn("No saga or saga state found to handle message. (message = {})", taskLookupContext.message());
            }
        } catch (Exception ex) {
            moduleError(executionContext, messageLookupContext.message(), ex);
            throw ex;
        } finally {
            moduleFinished(executionContext);
        }

        return sagasExecuted;
    }

    private boolean executeHandlersForMessage(final LookupContext messageLookupContext, final CurrentExecutionContext executionContext)
            throws InvocationTargetException, IllegalAccessException {
        boolean sagasExecuted = false;
        Object message = executionContext.message();

        Collection<SagaInstanceInfo> sagaDescriptions = env.instanceResolver().resolve(messageLookupContext);
        if (!sagaDescriptions.isEmpty()) {
            invokeSagas(executionContext, sagaDescriptions, message);
            sagasExecuted = true;
        } else if (!(message instanceof DeadMessage)) {
            DeadMessage deadMessage = new DeadMessage(taskLookupContext.message());
            LookupContext deadMessageContext = new SagaLookupContext(deadMessage, taskLookupContext, parentContext);
            sagasExecuted = startExecutionChain(deadMessageContext);
        }

        return sagasExecuted;
    }

    private void invokeSagas(final CurrentExecutionContext context, final Iterable<SagaInstanceInfo> sagaDescriptions, final Object invokeParam)
            throws InvocationTargetException, IllegalAccessException {
        if (context.dispatchingStopped()) {
            LOG.debug("Handler dispatching stopped before invoking any saga.");
        } else {
            for (SagaInstanceInfo sagaDescription : sagaDescriptions) {
                Saga saga = sagaDescription.getSaga();
                context.setSaga(saga);
                setSagaExecutionContext(saga, context);

                // call interceptor pre handling hooks
                interceptorStart(sagaDescription, context, invokeParam);
                interceptorHandling(saga, context, invokeParam);

                // perform actual saga invoke
                invoker.invoke(saga, invokeParam);

                // call interceptor handler finished hooks
                interceptorHandlingExecuted(saga, context, invokeParam);
                interceptorFinished(saga, context);
                updateStateStorage(sagaDescription);

                if (context.dispatchingStopped()) {
                    LOG.debug("Handler dispatching stopped after invoking saga {}.", sagaDescription.getSaga().getClass().getSimpleName());
                    break;
                }
            }
        }
    }

    private void interceptorHandling(final Saga saga, final ExecutionContext context, final Object invokeParam) {
        for (SagaLifetimeInterceptor interceptor : env.interceptors()) {
            interceptor.onHandlerExecuting(saga, context, invokeParam);
        }
    }

    private void interceptorHandlingExecuted(final Saga saga, final ExecutionContext context, final Object invokeParam) {
        for (SagaLifetimeInterceptor interceptor : env.interceptors()) {
            interceptor.onHandlerExecuted(saga, context, invokeParam);
        }
    }

    private void interceptorFinished(final Saga saga, final ExecutionContext context) {
        if (saga.isFinished()) {
            for (SagaLifetimeInterceptor interceptor : env.interceptors()) {
                interceptor.onFinished(saga, context);
            }
        }
    }

    private void interceptorStart(final SagaInstanceInfo sagaDescription, final ExecutionContext context, final Object invokeParam) {
        if (sagaDescription.isStarting()) {
            for (SagaLifetimeInterceptor interceptor : env.interceptors()) {
                interceptor.onStarting(sagaDescription.getSaga(), context, invokeParam);
            }
        }
    }

    private void moduleError(final ExecutionContext context, final Object invokeParam, final Exception ex) {
        for (SagaModule module : env.modules()) {
            module.onError(context, invokeParam, ex);
        }
    }

    private void moduleStarts(final ExecutionContext context) {
        for (SagaModule module : env.modules()) {
            module.onStart(context);
        }
    }

    private void moduleFinished(final ExecutionContext context) {
        for (SagaModule module : env.modules()) {
            module.onFinished(context);
        }
    }

    private void setHeaders(final CurrentExecutionContext context) {
        for (String key : taskLookupContext.getHeaders()) {
            context.setHeaderValue(key, taskLookupContext.getHeaderValue(key));
        }
    }

    private void setSagaExecutionContext(final Saga saga, final ExecutionContext context) {
        if (saga instanceof NeedContext) {
            ((NeedContext) saga).setExecutionContext(context);
        }
    }

    /**
     * Updates the state storage depending on whether the saga is completed or keeps on running.
     */
    private void updateStateStorage(final SagaInstanceInfo description) {
        Saga saga = description.getSaga();

        // if saga has finished delete existing state and possible timeouts
        // if saga has just been created state has never been save and there
        // is no need to delete it.
        if (saga.isFinished() && !description.isStarting()) {
            env.storage().delete(saga.state().getSagaId());
            env.timeoutManager().cancelTimeouts(saga.state().getSagaId());
        } else if (!saga.isFinished()) {
            env.storage().save(saga.state());
        }
    }

    /**
     * Similar to {@link #handle()} but intended for execution on any thread.<p/>
     * May throw a runtime exception in case something went wrong invoking the target saga message handler.
     */
    @Override
    public void run() {
        try {
            handle();
        } catch (InvocationTargetException | IllegalAccessException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public Object message() {
        return taskLookupContext.message();
    }
}
{'fraction_non_alphanumeric': 0.07820733552459957, 'fraction_numerical': 0.0010434601137371525, 'mean_word_length': 2.948094747682801, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '985885', 'n_tokens_mistral': 4797, 'n_tokens_neox': 4637, 'n_words': 1308}