代码之家 › 专栏 › 技术社区 › sansari

Spring集成:带标题的转换和路由

spring-integration-dsl spring-integration routing

sansari · 技术社区 · 8 年前

我正在构建一个基于Spring的库,在将消息转换为正确的类型之后,它应该使用消息并将其传递到配置的通道。我的库可以通过“streamToConsume:finalChannelDestination”对列表进行配置。

streams:
    source: destinationChannel

我想要一个 IntegrationFlow 如下所示:

IntegrationFlows
        .from(kinesisInboundChannelAdapter(amazonKinesis(), streamNames))
        .transform(new IssuanceTransformer())
        .route(router())
        .get();

public HeaderValueRouter router() {
    HeaderValueRouter router = new HeaderValueRouter(AwsHeaders.STREAM);
    consumerClientProperties.getKinesis().getStreams().forEach((k, v) ->
        router.setChannelMapping(k, v)
    );
    return router;
  }

转换事件,然后将其传递到配置中映射到流的通道。如何在转换后保留事件头以便将其发送到正确的通道?

谢谢你

1 回复 | 直到 8 年前

Artem Bilan 8 年前

我相信你的担心 IssuanceTransformer 没有一个愿望 AwsHeaders.STREAM 标题。开发自定义转换器时,需要确保将所有头从请求消息传输到回复消息:与许多其他组件不同,Transformer不会修改来自POJO的回复消息。

为此,您可以使用如下内容:

MessageBuilder.withPayload(myPayload).copyHeadersIfAbsent(requestMessage.getHeaders()).build();

注意:您可以使用 AwsHeaders.RECEIVED_STREAM 因为这个是从 KinesisMessageDrivenChannelAdapter :

private void performSend(AbstractIntegrationMessageBuilder<?> messageBuilder, Object rawRecord) {
        messageBuilder.setHeader(AwsHeaders.RECEIVED_STREAM, this.shardOffset.getStream())
                .setHeader(AwsHeaders.SHARD, this.shardOffset.getShard());

        if (CheckpointMode.manual.equals(KinesisMessageDrivenChannelAdapter.this.checkpointMode)) {
            messageBuilder.setHeader(AwsHeaders.CHECKPOINTER, this.checkpointer);
        }

推荐文章

BÅaÅ¼ej Karwowski · 如何处理JmsChannelFactoryBean错误,是否有可能使用自定义错误通道?

8 年前

m.gavr · 在Spring集成中,如何处理拆分后的消息(即使会出现错误)?

8 年前

Klaus Stadler · 在spring集成应用程序中禁用jackson的时区调整

8 年前

Roman · Spring集成:刷新/重新加载动态注册的IntegrationFlow

8 年前

Nick Tsitlakidis · 如何推迟spring集成中的消息消耗

8 年前

kiran reddy · split&aggregate flow中的Spring集成验证

8 年前

Tom · int http的请求-响应延迟时间:出站网关

8 年前

escos · Spring集成-哪种反序列化器用于只有一个起始字节的无限字节流(作为消息的一部分)

8 年前

user55926 · 如何验证使用Spring SFTP下载的文件的校验和

8 年前

user1516815 · 使用带有spring rest模板的Oauth2(授权码)rest api

8 年前