thingsboard-client-demo/src/main/java/com/tuoheng/machine/command/TransactionExecutor.java

package com.tuoheng.machine.command;

import com.tuoheng.machine.instruction.*;
import com.tuoheng.machine.mqtt.MqttCallbackRegistry;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 事务执行器（支持条件分支）
 */
@Slf4j
@Component
public class TransactionExecutor {

    private final MqttCallbackRegistry callbackRegistry;

    public TransactionExecutor(MqttCallbackRegistry callbackRegistry) {
        this.callbackRegistry = callbackRegistry;
    }

    /**
     * 执行事务
     */
    public CompletableFuture<CommandResult> executeTransaction(Transaction transaction, InstructionContext context) {
        log.info("开始执行事务: transaction={}, sn={}", transaction.getName(), context.getSn());

        CompletableFuture<CommandResult> future = new CompletableFuture<>();
        long startTime = System.currentTimeMillis();

        // 在新线程中执行事务
        CompletableFuture.runAsync(() -> {
            try {
                executeInstructionTree(transaction, context, startTime, future);
            } catch (Exception e) {
                log.error("事务执行异常: transaction={}, sn={}", transaction.getName(), context.getSn(), e);
                future.complete(CommandResult.failure(transaction.getCommandType(), "事务执行异常: " + e.getMessage()));
            }
        });

        return future;
    }

    /**
     * 执行指令树
     */
    private void executeInstructionTree(Transaction transaction, InstructionContext context,
                                         long startTime, CompletableFuture<CommandResult> future) {
        // 从根指令开始执行
        Instruction currentInstruction = transaction.getRootInstruction();
        if (currentInstruction == null) {
            log.error("事务没有根指令: transaction={}", transaction.getName());
            future.complete(CommandResult.failure(transaction.getCommandType(), "事务没有根指令"));
            return;
        }

        // 循环执行指令，直到没有下一个指令
        while (true) {
            // 检查事务是否超时
            if (System.currentTimeMillis() - startTime > transaction.getTimeoutMs()) {
                log.warn("事务执行超时: transaction={}, sn={}", transaction.getName(), context.getSn());
                future.complete(CommandResult.timeout(transaction.getCommandType()));
                return;
            }

            log.debug("执行指令: instruction={}", currentInstruction.getName());
            // 执行指令
            InstructionResult result = executeInstruction(currentInstruction, context);
            // 根据执行结果获取下游指令
            Instruction nextInstruction = currentInstruction.getNextInstruction(result.isSuccess());

            if (nextInstruction != null) {
                // 有下游指令，继续执行
                currentInstruction = nextInstruction;
                log.debug("根据执行结果选择下游指令: success={}, nextInstruction={}",
                        result.isSuccess(), nextInstruction.getName());
            } else {
                // 没有下游指令，当前指令的结果就是事务的结果
                if (!result.isSuccess()) {
                    // 指令失败，事务失败
                    log.error("指令执行失败（无下游指令）: instruction={}, error={}",
                            currentInstruction.getName(), result.getErrorMessage());
                    future.complete(CommandResult.failure(
                            transaction.getCommandType(),
                            result.getErrorMessage(),
                            currentInstruction.getName()
                    ));
                    return;
                } else {
                    // 指令成功，事务成功
                    log.info("指令执行成功（无下游指令），事务完成: instruction={}, sn={}",
                            currentInstruction.getName(), context.getSn());
                    future.complete(CommandResult.success(transaction.getCommandType()));
                    return;
                }
            }
        }
    }

    /**
     * 执行单个指令
     */
    private InstructionResult executeInstruction(Instruction instruction, InstructionContext context) {
        log.debug("开始执行指令: instruction={}, sn={}", instruction.getName(), context.getSn());

        try {
            // a. 判断是否可以执行
            if (!instruction.canExecute(context)) {
                String error = "指令不满足执行条件";
                log.warn("指令不满足执行条件: instruction={}, sn={}", instruction.getName(), context.getSn());
                InstructionResult result = InstructionResult.failure(error);
                instruction.onComplete(context, result);
                return result;
            }

            // b. 执行远程调用
            instruction.executeRemoteCall(context);
            log.debug("远程调用已发送: instruction={}", instruction.getName());

            // c. 等待方法回调
            CallbackConfig methodCallback = instruction.getMethodCallbackConfig(context);
            if (methodCallback != null) {
                InstructionResult methodResult = waitForCallback(methodCallback, context);
                if (!methodResult.isSuccess()) {
                    instruction.onComplete(context, methodResult);
                    return methodResult;
                }
            }

            // d. 等待状态回调
            CallbackConfig stateCallback = instruction.getStateCallbackConfig(context);
            if (stateCallback != null) {
                InstructionResult stateResult = waitForCallback(stateCallback, context);
                // 注意：这里不立即返回，而是将结果传递给上层，由上层决定下一步
                instruction.onComplete(context, stateResult);
                return stateResult;
            }

            // 指令执行成功
            InstructionResult result = InstructionResult.success();
            instruction.onComplete(context, result);
            return result;

        } catch (Exception e) {
            log.error("指令执行异常: instruction={}, sn={}", instruction.getName(), context.getSn(), e);
            InstructionResult result = InstructionResult.failure("指令执行异常: " + e.getMessage());
            instruction.onComplete(context, result);
            return result;
        }
    }

    /**
     * 等待回调
     */
    private InstructionResult waitForCallback(CallbackConfig callbackConfig, InstructionContext context) {
        CompletableFuture<InstructionResult> future = new CompletableFuture<>();
        AtomicBoolean callbackReceived = new AtomicBoolean(false);

        // 注册回调
        String callbackId = callbackRegistry.registerCallback(
                callbackConfig.getTopic(),
                messageBody -> {
                    if (callbackReceived.get()) {
                        return; // 已经收到回调，忽略后续消息
                    }

                    // 判断消息是否匹配
                    if (callbackConfig.matches(messageBody)) {
                        callbackReceived.set(true);
                        future.complete(InstructionResult.success(messageBody));
                        log.debug("收到匹配的回调消息: topic={}", callbackConfig.getTopic());
                    }
                },
                callbackConfig.getTimeoutMs()
        );

        try {
            // 等待回调或超时
            return future.get(callbackConfig.getTimeoutMs(), TimeUnit.MILLISECONDS);
        } catch (Exception e) {
            log.warn("等待回调超时: topic={}, timeout={}ms", callbackConfig.getTopic(), callbackConfig.getTimeoutMs());
            return InstructionResult.timeout();
        } finally {
            // 取消注册回调
            callbackRegistry.unregisterCallback(callbackId);
        }
    }
}