如何将自定义报头从spring云流(aws动态绑定器)发送到“遗留的”spring集成使用者

Question

我有两个应用程序--第一个应用程序是使用spring流/函数生成消息的AWS Kinesis，第二个应用程序是基于spring集成来使用消息的应用程序。两者之间的沟通不是问题-我可以从“流”发送一条信息，并在“集成”中轻松地处理它。

当我想发送一个自定义的标题时，就会出现一个问题。报头作为使用"New“格式的嵌入式报头到达使用者(开头有0 0xff，等等)-参见AbstractMessageChannelBinder#serializeAndEmbedHeadersIfApplicable中的spring云流。

然而，KinesisMessageDrivenChannelAdapter (Spring)似乎不理解“新”嵌入式报头形式。它使用不能“解码”消息的EmbeddedJsonHeadersMessageMapper (参见#toMessage)。它抛出一个com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'ÿ': was expecting (JSON String, Number, Array, Object or token 'null', 'true' or 'false')，因为嵌入的头(0xff等等)中包含了附加信息。

我需要将标题发送到线路上(标题用于在另一边进行路由)，因此不能在生产者上“关闭”标头。我没有看到一种使用“旧”嵌入式头的方法。

我想在制作人方面使用spring云流/功能--太棒了。我希望我能重做消费者，但是.

我可以编写自己的嵌入式头映射器来理解新格式(使用EmbeddedHeaderUtils)，并将其连接到KinesisMessageDrivenChannelAdapter中。

考虑到春季云流和春季集成之间的密切关系，我肯定做错了什么。Spring是否有一个理解新嵌入形式的OutboundMessageMapper？

还是有一种方法可以强迫弹簧云流使用不同的嵌入策略？

我可以在生产者端使用。(悲伤的脸)。

有什么想法吗？提前谢谢。

Answer 1

理解新格式

它不是“新”格式，而是创建的格式，最初是为Kafka创建的，后者只在0.11中增加了头支持。

我可以编写自己的嵌入式头映射器来理解新格式(使用EmbeddedHeaderUtils)，并将其连接到KinesisMessageDrivenChannelAdapter中。

我建议您这样做，并将将其贡献给核心Spring集成项目与EmbeddedJsonHeadersMessageMapper一起考虑，以便它可以与所有本机不支持标头的技术一起使用。