从 spring-cloud-stream app starter 修改消息体 [英] Modify the message body from spring-cloud-stream app starter

问题描述

我是 Spring Cloud Stream 的新手.我的用例是从文件源读取并为文件中的每一行发布消息(到 Kafka).我曾尝试使用文件源应用程序启动器(https://github.com/spring-cloud-stream-app-starters/file/tree/master/spring-cloud-starter-stream-source-file)并有能够发布消息.

I am new to Spring Cloud Stream. My use case is to read from a file source and publish messages (to Kafka) for each line in the file. I have tried using the file source app starter (https://github.com/spring-cloud-stream-app-starters/file/tree/master/spring-cloud-starter-stream-source-file) and have been able to publish messages.

但是，我现在需要在发布之前调整消息的正文.应用启动器生成通用消息，我需要在发布前修改结构.我试过在 SO 上搜索，但没有找到任何合适的例子.任何人都可以就如何实现这一目标提出建议吗?

However the I now need to tweak the body of the message before publishing. The app starter generates generic messages and I need to modify the structure before publishing. I have tried searching on SO but haven't found any suitable example. Can anyone please suggest on how to achieve this?

非常感谢.

推荐答案

实际上，这是一个我们即将在博客中发布的新功能，但我会尝试在此处进行解释.我相信您想扩展现有的应用程序，因此在这种情况下，您只需创建一个扩展 source-file 的新应用程序，然后使用新添加的 Spring Cloud Function 支持将您的转换器简单地组合到现有的应用程序.首先，您需要确保您使用的是最新的 Spring Cloud Stream，它应该是 Fishtown.RC1 (2.1.0.RC1).此外，我们还有一个 示例(将用于博客)，你可能会觉得有用.它实际上完全符合您的要求；只是它扩展了 http-source 而不是扩展 file-source，这意味着您只需从 spring-cloud-starter-stream- 交换 pom 中的依赖项source-http 到 spring-cloud-starter-stream-source-file ，然后简单地定义一个 Function 类型的 Bean，您可以在其中定义转换并在启动时提供一个属性 --spring.cloud.stream.function.definition=uppercase 其中 uppercase 是您要在文件源尾部组成的函数的名称.

Actually this is a new feature that we're going to blog soon, but I'll try to explain it here. I believe you want to extend the existing app, so in this case you simply want to create a new app that extends source-file and then use newly added Spring Cloud Function support to simply compose your transformer into the existing app. First, you need to make sure you're using the newest Spring Cloud Stream which should be Fishtown.RC1 (2.1.0.RC1). Also, we have an example (which is going to be used for the blog) that you may find useful. It actually does exactly what you're looking for; Only instead of extending file-source it extends http-source which means you simply have to swap a dependency in the pom from spring-cloud-starter-stream-source-http to spring-cloud-starter-stream-source-file and then simply define a Bean of type Function where you define your transformation and provide a property during teh startup --spring.cloud.stream.function.definition=uppercase where uppercase is the name of the function you want to compose at the tail of file source.

@SpringBootApplication
public class MyAppExtender {

    public static void main(String[] args) {
        SpringApplication.run(MyAppExtender.class, "--spring.cloud.stream.function.definition=uppercase");
    }

    @Bean
    public Function<String, String> uppercase() {
        return x -> x.toUpperCase();
    } 
}

无论如何，我知道上面的解释可能会遗漏一些部分，但请试一试，看看您是否有后续问题.我会确保我会在博客准备好后发布.

Anyway, I know the above explanation may be missing a few parts, but give it a shot and see if you have a follow up questions. I'll make sure I'll post the blog when it's ready.

这篇关于从 spring-cloud-stream app starter 修改消息体的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！