Отправить ответ на предыдущий маршрут отдыха

Question

У меня ниже верблюжий маршрут, в настоящее время работает нормально.Маршрут reset-route вызывается нажатием rest url.

reset-route работает нормально, но как правильно отправить ответ службе покоя?

Ответ, который я хочу отправить на URL отдыха, - это Количество сообщений, отправленных в MQService, и если маршрут переходит в doCatch, я хотел бы отправить «ОШИБКУ» в виде строки.

Яв настоящее время не имеет ни малейшего представления, потому что при запуске reset-route он вызывает Java-бин и получает список объектов, которые разделяются и обрабатываются параллельно.Как считать сообщение, успешно отправленное в MQ?Где конец сплита?Должен ли я использовать транзакции для постобработки после разделения?

@Override
public void configure() throws Exception {
    logger.info("ResetRoute.configure() : tp started configuring camel route.");

    restConfiguration()
            .enableCORS(false)
            .apiContextPath("/api-doc")
            .apiProperty("api.title", "REST Api to Trigger Swap Reset Tranformation.")
            .apiProperty("cors", "false")
            .apiContextRouteId("doc-api")
            .component("servlet")
            .dataFormatProperty("prettyPrint", "true");

    rest("/api/")
    .id("reset-api-route")
    .get("/reset")
    .to("direct:resetRoute");

    from("direct:resetRoute").routeId("reset-route")
            .log("Starting reset route")
            .doTry()
                .process(resetProcessor)
                .bean(transactionManager, "beginTransaction()")
                .bean(clientApi, "getResetTransactions(${body},${header.region})")
                .setHeader("transactionSize", simple("${body.size()}"))
                .choice()
                    .when(header("transactionSize").isLessThan(1))
                    .bean(transactionManager, "markSuccess")
                    .stop()
                .end()  //End of choose()
                    .wireTap("{{tp.reset.auditDir}}" + "${header.inXmlFileName}").onPrepare(jacksonProcessor)
                .split(simple("${body}"))
                    .parallelProcessing()
                    .bean(transformationService, "tranform")
                        .wireTap("{{tp.reset.auditDir}}" + "${header.outXmlFileName}").onPrepare(jacksonProcessor)
                    .process(jacksonProcessor).convertBodyTo(String.class)
                    .bean(mqService, "send")
                .end() //End of Split()
                .bean(transactionManager, "markSuccess")
                .transform().constant("SUCCESS")
            .endDoTry() //End of doTry()
            .doCatch(Exception.class)
                .bean(transactionManager, "markFailure").transform().constant("ERROR")
                .log(LoggingLevel.ERROR, "EXCEPTION: ${exception.stacktrace}")
            .end(); //End of doCatch()
}


This page contains the following errors:
error on line 1 at column 1: Document is empty
Below is a rendering of the page up to the first error.

Конец разделения

.split(simple("${body}"))
    .parallelProcessing()
    .bean(transformationService, "tranform")
        .wireTap("{{trade-publisher.reset.auditDir}}" + "${header.outXmlFileName}").onPrepare(jacksonProcessor)
    .process(jacksonProcessor).convertBodyTo(String.class)
    .bean(mqService, "send")
.end() //End of Split()
.bean(transactionManager, "markSuccess")
.transform().constant("SUCCESS")

У меня сложилось впечатление, что после строки .end() //End of Split() все, что я вызываю, будет вызываться только один раз, но в моем случае .bean(transactionManager, "markSuccess") вызывается несколько раз.

Answer 1

Проблема связана с прослушиванием телефонных разговоров, которое превращает ваше SplitDefinition в WireTapDefinition. Когда вы делаете .end (), вы не заканчиваете разделение, а прослушиваете. Это довольно запутанно, но часто бывает с верблюдом Вы должны проверить, какое у вас сейчас "xxxDefinition". В вашем случае удвоение ".end ()" должно сработать.

Вы можете поиграть с этим небольшим примером:

package com.example.camel;

import org.apache.camel.RoutesBuilder;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.language.tokenizer.TokenizeLanguage;
import org.apache.camel.test.junit4.CamelTestSupport;
import org.junit.Test;


public class CamelApplicationTests extends CamelTestSupport {

    @Test
    public void endingSplitTest() throws InterruptedException {
        template().sendBody("direct:start", "1,2,3");
        getMockEndpoint("mock:inner").expectedMessageCount(3);
        getMockEndpoint("mock:outer").expectedMessageCount(1);
        getMockEndpoint("mock:outer").setAssertPeriod(1000);
        assertMockEndpointsSatisfied();
    }

    @Override
    protected RoutesBuilder createRouteBuilder() throws Exception {
        return new RouteBuilder() {
            @Override
            public void configure() throws Exception {
                from("direct:start")
                    .split(TokenizeLanguage.tokenize(","))
                    .wireTap("mock:wiretap")
                    .end() // Ending WireTapDefinition
                    .log("Nested route : ${body}")
                    .to("mock:inner")
                    .end() // Ending SplitDefinition
                    .log("Outer route : ${body}")
                    .to("mock:outer");
            }
        };
    }
}

Вы также можете выполнять оба метода .end () последовательно, но для меня это выглядит немного хуже.

 @Override
    protected RoutesBuilder createRouteBuilder() throws Exception {
        return new RouteBuilder() {
            @Override
            public void configure() throws Exception {
                from("direct:start")
                    .split(TokenizeLanguage.tokenize(","))
                    .wireTap("mock:wiretap")
                    .log("Nested route : ${body}")
                    .to("mock:inner")
                    .end()
                    .end()
                    .log("Outer route : ${body}")
                    .to("mock:outer");
            }
        };
    }