如何将自定义标头从spring-cloud-stream（aws kinesis binder）发送到“传统”Spring集成消费者

Question

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

当我想发送自定义标题时，就会出现问题。标头作为使用“New”格式的嵌入式标头到达使用者（在开头有一个0xff等）-请参阅AbstractMessageChannelBinder#serializeAndEmbedHeadersIfApplicable in spring cloud stream。

然而，KinesisMessageDrivenChannelAdapter（spring integration aws）似乎并不理解“新的”嵌入头形式。它使用EmbeddedJsonHeadersMessageMapper（请参阅#toMessage），它无法“解码”消息。它抛出一个com。fasterxml。杰克逊。果心JsonParseException：无法识别的标记“ÿ”：应为（JSON字符串、数字、数组、对象或标记“null”、“true”或“false”），因为嵌入头中包含其他信息（0xff等）。

我需要通过导线发送标头（标头用于在另一侧布线），因此无法在生产者上“关闭”标头。我看不到使用“旧”嵌入头的方法。

我想在制作人方面使用Spring-Cloud-Stream/函数-这太棒了。我希望我可以重做消费者，但是。。。

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

鉴于sping-cloud-stream和sping-集成之间的密切关系，我一定做错了什么。Spring集成是否有一个能够理解新嵌入形式的OutbnMessageMapper？

或者有没有办法强制spring cloud html" target="_blank">stream使用不同的嵌入策略？

我可以在制作端使用Spring集成。（悲伤的脸）。

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

Answer 1

了解新格式

这不是一种“新”格式，而是Spring Cloud Stream创建的格式，最初是为Kafka创建的，它只在0.11中添加了标头支持。

我可以编写自己的嵌入式头映射器来理解新格式（使用嵌入式头映射器），并将其连接到动画片驱动程序（KinesisMessageDrivenChannelAdapter）中。

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