关于组合处理器和接收器的问题

Question

我正在使用Spring Can Data Flow 1.7.2。发布并尝试关注这篇博文

创建一个“将处理器接收器组合到单个应用程序中：“一个新的接收器”。”

当我将代码作为博客的示例进行结构化时，我遇到了一些问题，我认为这是因为博客的示例使用了java。util。类似于处理器的函数。

我想我应该使用java。util。消费者，因为我正在尝试将现有接收器更改为处理器接收器混合

我的班级是这样的：

@EnableBinding(Sink.class)
public class SampleCombinedSink extends Something {

    String modifiedPayload;
    Logger log;

    Consumer<String> consumer = i -> { modifiedPayload="STUFF ADDED BY CONSUMER i=["+i+"]"; };

    public void accept(String s){
      log.info("SampleCombinedSink.accept() s="+s);
    }

    @StreamListener(Sink.INPUT)
    public void doSink(String payload) {

      consumer.accept(payload);
      log.info("SampleCombinedSink.doSink() Payload received.");
      log.info("SampleCombinedSink.doSink() payload="+ payload);
      log.info("SampleCombinedSink.doSink() modifiedPayload="+ modifiedPayload);
    }
}

我的输出如下所示：

SampleCombinedSink.doSink() Payload received.
SampleCombinedSink.doSink() payload=Friday 11 January 2019 19:03:53.330+0000
SampleCombinedSink.doSink() modifiedPayload=STUFF ADDED BY CONSUMER i=[Friday 11 January 2019 19:03:53.330+0000]
SampleCombinedSink.doSink() Payload received.
SampleCombinedSink.doSink() payload=Friday 11 January 2019 19:03:54.332+0000
SampleCombinedSink.doSink() modifiedPayload=STUFF ADDED BY CONSUMER i=[Friday 11 January 2019 19:03:54.332+0000]
SampleCombinedSink.doSink() Payload received.
SampleCombinedSink.doSink() payload=Friday 11 January 2019 19:03:55.333+0000
SampleCombinedSink.doSink() modifiedPayload=STUFF ADDED BY CONSUMER i=[Friday 11 January 2019 19:03:55.333+0000]
SampleCombinedSink.doSink() Payload received.
SampleCombinedSink.doSink() payload=Friday 11 January 2019 19:03:56.313+0000
SampleCombinedSink.doSink() modifiedPayload=STUFF ADDED BY CONSUMER i=[Friday 11 January 2019 19:03:56.313+0000]

我的源每秒发出一个时间戳。

我被我的消费者弄糊涂了。

Consumer<String> consumer = i -> { modifiedPayload="STUFF ADDED BY CONSUMER i=["+i+"]"; };

我想我可以做如下事情：

Consumer<String> consumer = i -> { i="STUFF ADDED BY CONSUMER i=["+i+"]"; };

然后输入有效载荷

@StreamListener(Sink.INPUT)
public void doSink(String payload) {

将包含“消费者添加的内容i=[时间戳]”

事实并非如此。

我想更改doSink的输入，并通过添加“消费者添加的内容”进行更改，这样当输入到达doSink（字符串有效负载）时，有效负载将包含“消费者添加的内容I=[时间戳]”

我如何做到这一点？

Answer 1

在这种情况下，您不必更改Sink应用程序的任何内容，只需在Sink应用程序端使用相同的功能管道。

例如，使：

a combination of processor + sink into a single application: “a new sink”."

您所需要的只是将您的函数bean作为接收器应用程序的一部分，或者甚至将函数bean放在单独的工件中，但放在接收器应用程序的类路径中。一旦有了它，就可以定义Spring了。云流动作用接收器应用程序的定义。

您可以在此处看到此示例。由日志组成的应用程序定义了函数bean。

要运行示例，请执行以下操作：

dataflow:>app register --name http-transformer --type source --uri file:///Users/igopinathan/dev/git/ilayaperumalg/sandbox/function-composition/http-transformer/target/http-transformer-2.1.0.BUILD-SNAPSHOT.jar
Successfully registered application 'source:http-transformer'

dataflow:>app register --name log-composed --type sink --uri file:///Users/igopinathan/.m2/repository/org/springframework/cloud/stream/app/log-composed/2.1.0.BUILD-SNAPSHOT/log-composed-2.1.0.BUILD-SNAPSHOT.jar
Successfully registered application 'sink:log-composed'

dataflow:>stream create helloComposed --definition "http-transformer --server.port=9001 | log-composed"
Created new stream 'helloComposed'


dataflow:>stream deploy helloComposed --properties "app.log-composed.spring.cloud.stream.function.definition=upper|concat,deployer.*.local.inheritLogging=true"
Deployment request has been sent for stream 'helloComposed'

dataflow:>http post --data "friend" --target "http://localhost:9001"

在stream deploy命令中，您可以看到用于指定函数组成的应用程序是由日志组成的。