一个工作流建立好后，如何多次重复调用这个工作流

[superclocks]

我想创建一个可以处理流式数据的工作流，例如第一个Operator从某个UDP端口上持续读取视频流，将视频流体传给第二个Operator，在第二个Operator中做目标识别，将识别结果传递给第三个Operator，第三个Operator将结果保存，这三个Operator建立DagEngine后，如果在while中调用runAndWait，工作流会每次都从第一个Operator执行到最后一个Operator吗，类似如下代码，如果在for循环中执行 engine.runAndWait(300_000);，我构建的1-2，3->4 的工作流每次都会执行一次吗

public void test() {
//请求上下文
OpConfig opConfig = OpConfigEntity.getOpConfig();
opConfig.getExtMap().put("tStart", -3.14);
opConfig.getExtMap().put("tEnd", 3.14);
opConfig.getExtMap().put("f", 10.0);
opConfig.getExtMap().put("t", 0);

    System.out.println(GsonUtil.prettyPrint(opConfig));
    System.out.println("==========================");

    DagEngine engine = new DagEngine(opConfig, executor);
    OperatorWrapper wrapper1 = new OperatorWrapper<OpConfig, OpConfig>()
            .id("1")
            .engine(engine)
            .operator(time)
            .next("2", "3");
    OperatorWrapper wrapper2 = new OperatorWrapper<OpConfig, Double>()
            .id("2")
            .engine(engine)
            .operator(sin)
            .next("4")
            .addParamFromWrapperId("1");
    OperatorWrapper wrapper3 = new OperatorWrapper<OpConfig, Double>()
            .id("3")
            .engine(engine)
            .operator(cos)
            .next("4")
            .addParamFromWrapperId("1");;

    OperatorWrapper wrapper4 = new OperatorWrapper<List<Double>, Double>()
            .id("4")
            .engine(engine)
            .operator(multiply).addParamFromWrapperId("2").addParamFromWrapperId("3");
    for(int i = 0; i < 10; i++) {
        engine.runAndWait(300_000);
        System.out.println(GsonUtil.prettyPrint(wrapper4.getOperatorResult().getResult()));
        if (engine.getEx() != null) {
            engine.getEx().printStackTrace();
        }
    }

[ytyht226]

我想创建一个可以处理流式数据的工作流，例如第一个Operator从某个UDP端口上持续读取视频流，将视频流体传给第二个Operator，在第二个Operator中做目标识别，将识别结果传递给第三个Operator，第三个Operator将结果保存，这三个Operator建立DagEngine后，如果在while中调用runAndWait，工作流会每次都从第一个Operator执行到最后一个Operator吗，类似如下代码，如果在for循环中执行 engine.runAndWait(300_000);，我构建的1-2，3->4 的工作流每次都会执行一次吗

public void test() { //请求上下文 OpConfig opConfig = OpConfigEntity.getOpConfig(); opConfig.getExtMap().put("tStart", -3.14); opConfig.getExtMap().put("tEnd", 3.14); opConfig.getExtMap().put("f", 10.0); opConfig.getExtMap().put("t", 0);

    System.out.println(GsonUtil.prettyPrint(opConfig));
    System.out.println("==========================");

    DagEngine engine = new DagEngine(opConfig, executor);
    OperatorWrapper wrapper1 = new OperatorWrapper<OpConfig, OpConfig>()
            .id("1")
            .engine(engine)
            .operator(time)
            .next("2", "3");
    OperatorWrapper wrapper2 = new OperatorWrapper<OpConfig, Double>()
            .id("2")
            .engine(engine)
            .operator(sin)
            .next("4")
            .addParamFromWrapperId("1");
    OperatorWrapper wrapper3 = new OperatorWrapper<OpConfig, Double>()
            .id("3")
            .engine(engine)
            .operator(cos)
            .next("4")
            .addParamFromWrapperId("1");;

    OperatorWrapper wrapper4 = new OperatorWrapper<List<Double>, Double>()
            .id("4")
            .engine(engine)
            .operator(multiply).addParamFromWrapperId("2").addParamFromWrapperId("3");
    for(int i = 0; i < 10; i++) {
        engine.runAndWait(300_000);
        System.out.println(GsonUtil.prettyPrint(wrapper4.getOperatorResult().getResult()));
        if (engine.getEx() != null) {
            engine.getEx().printStackTrace();
        }
    }

一个工作流建立好后只能执行一次，因为DagEngine、Wrapper等都是有状态的，你这个场景下，只需要在读取视频流后，每次都创建完整的工作流即可（DagEngine、Wrapper），工作流处理逻辑都是一样的，虽然是每次都重新创建工作流，但代码只有一份。

[superclocks]

我想创建一个可以处理流式数据的工作流，例如第一个Operator从某个UDP端口上持续读取视频流，将视频流体传给第二个Operator，在第二个Operator中做目标识别，将识别结果传递给第三个Operator，第三个Operator将结果保存，这三个Operator建立DagEngine后，如果在while中调用runAndWait，工作流会每次都从第一个Operator执行到最后一个Operator吗，类似如下代码，如果在for循环中执行 engine.runAndWait(300_000);，我构建的1-2，3->4 的工作流每次都会执行一次吗
public void test() { //请求上下文 OpConfig opConfig = OpConfigEntity.getOpConfig(); opConfig.getExtMap().put("tStart", -3.14); opConfig.getExtMap().put("tEnd", 3.14); opConfig.getExtMap().put("f", 10.0); opConfig.getExtMap().put("t", 0);

    System.out.println(GsonUtil.prettyPrint(opConfig));
    System.out.println("==========================");

    DagEngine engine = new DagEngine(opConfig, executor);
    OperatorWrapper wrapper1 = new OperatorWrapper<OpConfig, OpConfig>()
            .id("1")
            .engine(engine)
            .operator(time)
            .next("2", "3");
    OperatorWrapper wrapper2 = new OperatorWrapper<OpConfig, Double>()
            .id("2")
            .engine(engine)
            .operator(sin)
            .next("4")
            .addParamFromWrapperId("1");
    OperatorWrapper wrapper3 = new OperatorWrapper<OpConfig, Double>()
            .id("3")
            .engine(engine)
            .operator(cos)
            .next("4")
            .addParamFromWrapperId("1");;

    OperatorWrapper wrapper4 = new OperatorWrapper<List<Double>, Double>()
            .id("4")
            .engine(engine)
            .operator(multiply).addParamFromWrapperId("2").addParamFromWrapperId("3");
    for(int i = 0; i < 10; i++) {
        engine.runAndWait(300_000);
        System.out.println(GsonUtil.prettyPrint(wrapper4.getOperatorResult().getResult()));
        if (engine.getEx() != null) {
            engine.getEx().printStackTrace();
        }
    }

一个工作流建立好后只能执行一次，因为DagEngine、Wrapper等都是有状态的，你这个场景下，只需要在读取视频流后，每次都创建完整的工作流即可（DagEngine、Wrapper），工作流处理逻辑都是一样的，虽然是每次都重新创建工作流，但代码只有一份。

还有个问题，我想用图形化工具生成每个Operator 之间的参数映射关系，然后在后台实现一个解析引擎，根据前端的Operator之间的数据传递关系，动态的实现工作流搭建，我现在发现你给的工作流搭建方式都是通过OperatorWrapper实现的，而OperatorWrapper又是个范型，它的范型参数是动态的，从前端配置产生，例如new OperatorWrapper<OpConfig, Double>，我在前端可以配置成输入参数是OpConfig，返回参数是Double，如果通过配置文件信息，动态的实例化具体的OperatorWrapper哪。我看之前你回复过后续支持通过配置定义以来关系，我现在需要这个功能，你实现这个功能是利用现有的OperatorWrapper还是要升级它，能给点你实现的思路吗，我想自己开发一版先用着。

[ytyht226]

通过json配置可以定义DAG工作流中有哪些节点及节点之间的依赖关系，比如下面配置中，tasks是op集合，topology描述的是op之间的依赖关系；解析该配置，通过tasks和topology就可以生成OperatorWrapper；
至于每个op的入参、出参：
1、简单一点每一类DAG流程可以定义一个统一的上下文Context，每个op及wrapper的泛型定义都用这个Context，然后在每个op里面结合业务逻辑具体需要哪些参数再从Context取。
2、如果就想明确定义op及wrapper的出参、入参类型，可以使用json-path，在json配置的topology中通过path定义节点参数的来源；当前版本已实现使用编码方式实现json-path的能力，可参考https://github.com/ytyht226/taskflow/tree/master/taskflow-example/src/main/java/org/taskflow/example/param/demo3

{
"tasks": [
{
"id": "task_1",
"operator": "operator_1"
},
{
"id": "task_2",
"operator": "operator_2"
}
],
"topology": {
"task_1": {
"depend": [

        ]
    },
    "task_2": {
        "depend": [
            "task_1"
        ]
    }
}

}

[superclocks]

现在有这样场景，在一个函数中调用工作流，需要给第一个节点传入自定义参数，只能将参数放入OpConfig传入吗

[ytyht226]

现在有这样场景，在一个函数中调用工作流，需要给第一个节点传入自定义参数，只能将参数放入OpConfig传入吗

你可以在初始化DagEngine时，传入自定义的上下文，看下这个构造函数
org.taskflow.core.DagEngine#DagEngine(java.lang.Object, java.util.concurrent.ExecutorService)

[superclocks]

通过jsonpath将上一个节点的protobuf对象传到下一个节点中，为啥把protobuf对象都转成json了呢，为啥不是Object对象呀