/** * This class represents a task related to the execution of an action in { @link * ActionExecutionOperator}. * * <p>An action is split into multiple code blocks, and each code block is represented by an { @code * ActionTask}. You can call { @link #invoke()} to execute a code block and obtain invoke result * { @link ActionTaskResult}. If the action contains additional code blocks, you can obtain the next * { @code ActionTask} via { @link ActionTaskResult#getGeneratedActionTask()} and continue executing * it. */ public abstract class ActionTask {     protected final Object key;     protected final Event event;     protected final Action action;     /**     * Since RunnerContextImpl contains references to the Operator and state, it should not be     * serialized and included in the state with ActionTask. Instead, we should check if a valid     * RunnerContext exists before each ActionTask invocation and create a new one if necessary.     */     protected transient RunnerContextImpl runnerContext;     public ActionTask (Object key, Event event, Action action) {         this .key = key;         this .event = event;         this .action = action;    }     public RunnerContextImpl getRunnerContext () {         return runnerContext;    }     public void setRunnerContext (RunnerContextImpl runnerContext) {         this .runnerContext = runnerContext;    }     public Object getKey () {         return key;    }     /** Invokes the action task. */     public abstract ActionTaskResult invoke () throws Exception;     public class ActionTaskResult {         private final boolean finished;         private final List<Event> outputEvents;  msaig.baijinpai.cn       private final Optional<ActionTask> generatedActionTaskOpt;         public ActionTaskResult (                 boolean finished,                List<Event> outputEvents,                 @Nullable ActionTask generatedActionTask) {             this .finished = finished;             this .outputEvents = outputEvents;             this .generatedActionTaskOpt = Optional.ofNullable(generatedActionTask);        }         public boolean isFinished () {             return finished;        }         public List<Event> getOutputEvents () {             return outputEvents;        }

public class PythonActionTask extends ActionTask {     public ActionTaskResult invoke () throws Exception {         PythonActionExecutor pythonActionExecutor = getPythonActionExecutor();         // 这里执行实际的 Python 函数         String pythonGeneratorRef =            pythonActionExecutor.executePythonFunction(                (PythonFunction) action.getExec(), // <-- 这就是对应的函数                (PythonEvent) event,                runnerContext);         // 处理异步情况         if (pythonGeneratorRef != null ) {             // 如果函数返回了生成器，则创建新的任务继续执行             ActionTask tempGeneratedActionTask =                 new PythonGeneratorActionTask (key, event, action, pythonGeneratorRef);            tempGeneratedActionTask.setRunnerContext(runnerContext);         if (pythonGeneratorRef != null ) {             // 如果函数返回了生成器，则创建新的任务继续执行             ActionTask tempGeneratedActionTask =                 new PythonGeneratorActionTask (key, event, action, pythonGeneratorRef);            tempGeneratedActionTask.setRunnerContext(runnerContext);             return tempGeneratedActionTask.invoke();        }         // 否则表示函数已执行完毕  vnul.baijinpai.cn         return new ActionTaskResult (             true ,            runnerContext.drainEvents(event.getSourceTimestamp()),

public class PythonGeneratorActionTask extends PythonActionTask {     private final String pythonGeneratorRef;     public PythonGeneratorActionTask (            Object key, Event event, Action action, String pythonGeneratorRef) {         super (key, event, action);         this .pythonGeneratorRef = pythonGeneratorRef;    }     @Override     public ActionTaskResult invoke () {         boolean finished = getPythonActionExecutor().callPythonGenerator(pythonGeneratorRef);         ActionTask generatedActionTask = finished ? null : this ;         return new ActionTaskResult (                finished,

* A special { @link ActionTask} designed to execute a Java action task. * * <p>Note that Java action currently do not support asynchronous execution. As a result, a Java * action task will be invoked only once. */ public class JavaActionTask extends ActionTask {     private final ClassLoader userCodeClassLoader;     public JavaActionTask (Object key, Event event, Action action, ClassLoader userCodeClassLoader) {         super (key, event, action);        checkState(action.getExec() instanceof JavaFunction);         this .userCodeClassLoader = userCodeClassLoader;    }     @Override   ftwox.baijinpai.cn     public ActionTaskResult invoke () throws Exception {        runnerContext.checkNoPendingEvents();         ClassLoader cl = Thread.currentThread().getContextClassLoader();         try {            Thread.currentThread().setContextClassLoader(userCodeClassLoader);            action.getExec().call(event, runnerContext);        } finally {            Thread.currentThread().setContextClassLoader(cl);        }         return new ActionTaskResult (

    * `tryProcessActionTaskForKey` to continue processing.     */uugac.baijinpai.cn     private void processEvent (Object key, Event event) throws Exception {  buzxc.baijinpai.cn        notifyEventProcessed(event);   rpg.baijinpai.cn      boolean isInputEvent = EventUtil.isInputEvent(event);         if (EventUtil.isOutputEvent(event)) {cw.baijinpai.cn        } else {    nm6sx.baijinpai.cn         // We then obtain the triggered action and add ActionTasks to the waiting processing             // queue.            List<Action> triggerActions = getActionsTriggeredBy(event);             if (triggerActions != null && !triggerActions.isEmpty()) {     btbvn.baijinpai.cn            for (Action triggerAction : triggerActions) {                    actionTasksKState.add(createActionTask(key, triggerAction, event));

  private ActionTask createActionTask (Object key, Action action, Event event) {         if (action.getExec() instanceof JavaFunction) {             return new JavaActionTask (                    key, event, action, getRuntimeContext().getUserCodeClassLoader());        } else if (action.getExec() instanceof PythonFunction) {             return new PythonActionTask (key, event, action);tsepjb.baijinpai.cn        } else {             throw new IllegalStateException (                     "Unsupported action type: " + action.getExec().getClass());

* 专门用于执行 Python 动作任务的特殊 ActionTask。 * * <p>在 Python 中进行异步执行期间，PythonActionTask 可以生成一个 * PythonGeneratorActionTask 来代表后续需要的代码块。 */ public class PythonActionTask extends ActionTask {     public ActionTaskResult invoke () throws Exception {        ...         String pythonGeneratorRef = wap.mfmbsh.com            pythonActionExecutor.executePythonFunction(   yqrcu.baijinpai.cn                (PythonFunction)action.getExec(),m.baijinpai.cn                (PythonEvent)event,                runnerContext);         // 如果用户定义的动作使用接口提交了异步任务，         // 它将在第一次执行后返回一个 Python 生成器对象实例。         // 否则意味着没有提交异步任务且动作已经完成。         if (pythonGeneratorRef != null ) {             // Python 动作生成了一个生成器。我们需要执行一次它，             // 这将提交一个异步任务并返回动作是否已完成。             ActionTask tempGeneratedActionTask = m.mfmbsh.com                 new PythonGeneratorActionTask (key, event, action, pythonGeneratorRef);            tempGeneratedActionTask.setRunnerContext(runnerContext);             return tempGeneratedActionTask.invoke();        }

        return new ActionTaskResult (