使用骆驼 pollEnrich 合并 xml 输出

Question

从需要为 XSLT 处理器组合的各种来源创建 XML 输出，但在使用 pollEnrich 时得到“端点上没有可用的消费者”。 pollEnrich 聚合器未通过 pollEnrich Exchange。

取出聚合器并使用默认聚合器。得到同样的问题。添加日志显示有 XML 输出来自之前的路由到 pollEnrich 端点。

package com.hitrust.route;

import com.hitrust.aggregator.AddToOutput;
import com.hitrust.processor.ConvertResultToRecords;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.processor.aggregate.AggregationStrategy;

public class AuthoritativeSourceDocument extends RouteBuilder {

   public void configure() throws Exception {

    AggregationStrategy addToOutput = new AddToOutput();

    restConfiguration()
           .component("restlet")
           .host("localhost").port("18082");

    rest("/authoritativesourcedocument/{authoritativesourcedocumentid}")
            .consumes("application/json").produces("application/json")
            .get()
            .to("direct:Start");

    from("direct:Start")
            .multicast()
            .to("direct:GetSections")
            .to("direct:GetTransactions")
            .to("direct:MergeSections");

    from("direct:GetSections")
            .setBody(simple("SELECT * " +
                    "  FROM [dbo].[Section] AS S" +
                    "  WHERE [Id] = ${header.id}"))
            .to("jdbc:dataSource")
            .setProperty("paramName", simple("Sections"))
            .process(new ConvertResultToRecords())
            .to("direct:GetSectionsOutput");

    from("direct:GetTransactions")
            .setBody(simple("SELECT * " +
                    "  FROM [dbo].[SectionTransaction] AS ST" +
                    "  WHERE [Id] = ${header.id}"))
            .to("jdbc:dataSource")
            .setProperty("paramName", simple("SectionTransactions"))
            .process(new ConvertResultToRecords())
            .to("direct:GetTransactionsOutput");

    from("direct:MergeSections")
            .setBody(simple("<param><id>${header.id}</id></param>"))
            .convertBodyTo(org.w3c.dom.Document.class)
            .pollEnrich("direct:GetSectionsOutput", 500, addToOutput)
            .pollEnrich("direct:GetTransactionsOutput", 500, addToOutput)
            .to("xslt:file:src/main/resources/xslts/MergeSections.xsl")


  }
}

执行并合并最后一条路由的 xml 输出。

Answer 1

请尝试使用“seda:”而不是“direct:”从您的 Get* 路由向 MergeSections 路由中的 pollEnrich 发送消息：

[...]
from("direct:Start")
        .multicast()
        .to("direct:GetSections")
        .to("direct:GetTransactions")
        .to("direct:MergeSections");

from("direct:GetSections")
        .setBody(simple("SELECT * " +
                "  FROM [dbo].[Section] AS S" +
                "  WHERE [Id] = ${header.id}"))
        .to("jdbc:dataSource")
        .setProperty("paramName", simple("Sections"))
        .process(new ConvertResultToRecords())
        .to("seda:GetSectionsOutput");

from("direct:GetTransactions")
        .setBody(simple("SELECT * " +
                "  FROM [dbo].[SectionTransaction] AS ST" +
                "  WHERE [Id] = ${header.id}"))
        .to("jdbc:dataSource")
        .setProperty("paramName", simple("SectionTransactions"))
        .process(new ConvertResultToRecords())
        .to("seda:GetTransactionsOutput");

from("direct:MergeSections")
        .setBody(simple("<param><id>${header.id}</id></param>"))
        .convertBodyTo(org.w3c.dom.Document.class)
        .pollEnrich("seda:GetSectionsOutput", 500, addToOutput)
        .pollEnrich("seda:GetTransactionsOutput", 500, addToOutput)
        .to("xslt:file:src/main/resources/xslts/MergeSections.xsl")

您的代码看起来还假设默认情况下多播正在并行调度到端点 - 它没有。您需要为此添加 parallelProcessing() 选项。