Spring集成使用和方法验证

1. 下面的代码是使用DSL构造流的正确方法吗?
2. 在下面的“C”中,我可以将结果冒泡到“B”流吗?
3. 使用DSL和XML是更好的方法吗?
4. 我很困惑如何正确地“终止”流?

在下面的代码中,我只是将页面发布到目标。整个流程是这样的。

1. Publisher-flow侦听负载并将其拆分为多个部分。
2. 内容流过滤出页面并将其拆分为多个部分。
   1. AWS flow订阅并处理部件。
   2. 文件流订阅并处理部件。

A) 发布流(publisher.jar):

这是我通过网关启动的“主”流。其目的是作为开始触发所有发布流的入口点。

1. 接收消息
2. 预处理消息并保存。
4. 用其余数据充实每个条目
5. 将每个条目放在输出通道上。

代码如下:

@Bean
IntegrationFlow flowPublish()
{
    return f -> f
        .channel(this.publishingInputChannel())
        //Prepare the payload
        .<Package>handle((p, h) -> this.save(p))
        //Split the artifact resolved items
        .split(Package.class, Package::getItems)
        //Find the artifact associated to each item (if available)
        .enrich(
            e -> e.<PackageEntry>requestPayload(
                m ->
                {
                    final PackageEntry item = m.getPayload();
                    final Publishable publishable = this.findPublishable(item);
                    item.setPublishable(publishable);
                    return item;
                }))
        //Send the results to the output channel
        .channel(this.publishingOutputChannel());
}

该模块的职责是处理传入的“内容”有效负载(即本例中的页面),并将其拆分/路由到适当的订阅者。

1. 收听发布者输出频道
3. 将原始有效载荷添加到收割台以备将来使用
4. 将有效载荷转换为实际类型
6. 将每个元素路由到适当的PubSub通道。

代码如下:

@Bean
@ContentChannel("asset")
MessageChannel contentAssetChannel()
{
    return MessageChannels.publishSubscribe("assetPublisherChannel").get();

    //return MessageChannels.queue(10).get();
}

@Bean
@ContentChannel("page")
MessageChannel contentPageChannel()
{
    return MessageChannels.publishSubscribe("pagePublisherChannel").get();

    //return MessageChannels.queue(10).get();
}

@Bean
IntegrationFlow flowPublishContent()
{
    return flow -> flow
        .channel(this.publishingChannel)
        //Filter for root pages (which contain elements)
        .filter(PackageEntry.class, p -> p.getPublishable() instanceof Page)
        //Put the publishable details in the header
        .enrichHeaders(e -> e.headerFunction("item", Message::getPayload))
        //Transform the item to a Page
        .transform(PackageEntry.class, PackageEntry::getPublishable)
        //Split page into components and put the type in the header
        .split(Page.class, this::splitPageElements)
        //Route content based on type to the subscriber
        .<PageContent, String>route(PageContent::getType, mapping -> mapping
            .resolutionRequired(false)
            .subFlowMapping("page", sf -> sf.channel(this.contentPageChannel()))
            .subFlowMapping("image", sf -> sf.channel(this.contentAssetChannel()))
            .defaultOutputToParentFlow())
        .channel(IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME);
}

该模块是内容特定流的许多潜在订户之一。它根据发布到上面的路由通道分别处理每个元素。

1. 订阅适当的频道。

例如,“contentPageChannel”可以调用下面的flowPageToS3(在aws模块中)和FlowPagetFile(在另一个模块中)。

代码如下:

@Bean
IntegrationFlow flowAssetToS3()
{
    return flow -> flow
        .channel(this.assetChannel)
        .publishSubscribeChannel(c -> c
            .subscribe(s -> s
                .<PageContent>handle((p, h) ->
                                     {
                                         return this.publishS3Asset(p);
                                     })));
}

@Bean
IntegrationFlow flowPageToS3()
{
    return flow -> flow
        .channel(this.pageChannel)
        .publishSubscribeChannel(c -> c
            .subscribe(s -> s
                .<Page>handle((p, h) -> this.publishS3Page(p))
                .enrichHeaders(e -> e.header("s3Command", Command.UPLOAD.name()))
                .handle(this.s3MessageHandler())));
}