com.mongodb.MongoGridFSException#org.apache.camel.Exchange源码实例Demo

@Override
@SuppressWarnings("unchecked")
public Pair<String, List<PropagationStatus>> reactivate(
        final StatusR statusR, final boolean nullPriorityAsync, final String updater, final String context) {

    PollingConsumer pollingConsumer = getConsumer("direct:statusPort");

    Map<String, Object> props = new HashMap<>();
    props.put("key", statusR.getKey());
    props.put("statusR", statusR);
    props.put("nullPriorityAsync", nullPriorityAsync);
    props.put("updater", updater);
    props.put("context", context);

    if (statusR.isOnSyncope()) {
        sendMessage("direct:reactivateUser", statusR.getKey(), props);
    } else {
        UserWorkflowResult<String> updated =
                new UserWorkflowResult<>(statusR.getKey(), null, null, statusR.getType().name().toLowerCase());
        sendMessage("direct:userStatusPropagation", updated, props);
    }

    Exchange exchange = pollingConsumer.receive();

    if (exchange.getProperty(Exchange.EXCEPTION_CAUGHT) != null) {
        throw (RuntimeException) exchange.getProperty(Exchange.EXCEPTION_CAUGHT);
    }

    return exchange.getIn().getBody(Pair.class);
}
 

@Test
public void testGetInvalid() throws Exception {
    final int size = getMenuService().getMenuItems().size();

    try {
        // TODO: report camel-undertow not throwing exception on failure
        String out = fluentTemplate().to("netty4-http:http://localhost:" + port1 + "/cafe/menu/items/" + (size + 1))
                .withHeader(Exchange.HTTP_METHOD, "GET")
                .request(String.class);
    } catch (Exception e) {
        // Expect Exception to be thrown since we're requesting an item that does not exist
        //System.out.println("Exception Message = " + e.getMessage());
        return;
    }

    fail("Expected call to fail with exception thrown");
}
 

@Test
public void testIsCamelMessage() throws Exception {
    MockEndpoint mock = getMockEndpoint("mock:quote");
    mock.expectedMessageCount(2);

    template.sendBody("stub:jms:topic:quote", "Hello Camel");
    template.sendBody("stub:jms:topic:quote", "Camel rocks");

    assertMockEndpointsSatisfied();

    List<Exchange> list = mock.getReceivedExchanges();
    String body1 = list.get(0).getIn().getBody(String.class);
    String body2 = list.get(1).getIn().getBody(String.class);
    assertTrue(body1.contains("Camel"));
    assertTrue(body2.contains("Camel"));
}
 

private void beforeProducer(Exchange exchange) {
    final Message in = exchange.getIn();
    final KuduTable model = exchange.getIn().getBody(KuduTable.class);

    if (model != null && ObjectHelper.isNotEmpty(model.getSchema())) {
        schema = model.getSchema();
    }

    KuduTable.ColumnSchema[] columnSchema = schema.getColumns();
    List<ColumnSchema> columns = new ArrayList<>(columnSchema.length);
    List<String> rangeKeys = new ArrayList<>();
    for (int i = 0; i < columnSchema.length; i++) {
        if (columnSchema[i].isKey()) {
            rangeKeys.add(columnSchema[i].getName());
        }

        columns.add(
                new ColumnSchema.ColumnSchemaBuilder(columnSchema[i].getName(), convertType(columnSchema[i].getType()))
                        .key(columnSchema[i].isKey())
                        .build()
        );
    }

    in.setHeader("Schema", new Schema(columns));
    in.setHeader("TableOptions", new CreateTableOptions().setRangePartitionColumns(rangeKeys));
}
 

@Deployment(resources = { "process/example.bpmn20.xml" })
public void testRunProcess() throws Exception {
  CamelContext ctx = applicationContext.getBean(CamelContext.class);
  ProducerTemplate tpl = ctx.createProducerTemplate();
  service1.expectedBodiesReceived("ala");

  Exchange exchange = ctx.getEndpoint("direct:start").createExchange();
  exchange.getIn().setBody(Collections.singletonMap("var1", "ala"));
  tpl.send("direct:start", exchange);
  String instanceId = (String) exchange.getProperty("PROCESS_ID_PROPERTY");

  tpl.sendBodyAndProperty("direct:receive", null, ActivitiProducer.PROCESS_ID_PROPERTY, instanceId);

  assertProcessEnded(instanceId);

  service1.assertIsSatisfied();
  Map<?, ?> m = service2.getExchanges().get(0).getIn().getBody(Map.class);
  assertEquals("ala", m.get("var1"));
  assertEquals("var2", m.get("var2"));
}
 

@Test
public void testDlqMultistep() throws Exception {
    final MockEndpoint mockResult = getMockEndpoint("mock:result");
    mockResult.expectedMessageCount(1);
    mockResult.expectedBodiesReceived("Foo");
    mockResult.message(0).exchangeProperty(Exchange.EXCEPTION_CAUGHT).isNull();
    mockResult.message(0).header("myHeader").isEqualTo("changed");

    final MockEndpoint mockError = getMockEndpoint("mock:error");
    mockError.expectedMessageCount(1);
    mockError.expectedBodiesReceived("KaBoom");
    mockError.message(0).exchangeProperty(Exchange.EXCEPTION_CAUGHT).isNotNull();
    mockError.message(0).exchangeProperty(Exchange.FAILURE_ROUTE_ID).isEqualTo("myFlakyRoute");
    mockError.message(0).header("myHeader").isEqualTo("flaky");

    template.sendBodyAndHeader("direct:multiroute", "Foo", "myHeader", "original");
    template.sendBodyAndHeader("direct:multiroute", "KaBoom", "myHeader", "original");

    assertMockEndpointsSatisfied();
}
 

@Override
public void onExchangeDone(Route route, Exchange exchange) {
    // decide which route to stop and start
    // basically we should flip the two routes
    String stop = route.getId().equals(name1) ? name1 : name2;
    String start = route.getId().equals(name1) ? name2 : name1;

    CamelContext context = exchange.getContext();
    try {
        // force stopping this route while we are routing an Exchange
        // requires two steps:
        // 1) unregister from the inflight registry
        // 2) stop the route
        context.getInflightRepository().remove(exchange);
        context.stopRoute(stop);
        // then we can start the other route
        context.startRoute(start);
    } catch (Exception e) {
        // let the exception handle handle it, which is often just to log it
        getExceptionHandler().handleException("Error flipping routes", e);
    }
}
 

@Test
public void testNoTenantId() throws Exception {

    RangedWorkNote workNote = RangedWorkNote.createSimpleWorkNote(BATCHJOBID);

    Exchange ex = Mockito.mock(Exchange.class);
    Message message = Mockito.mock(Message.class);
    Mockito.when(ex.getIn()).thenReturn(message);
    Mockito.when(message.getBody(WorkNote.class)).thenReturn(workNote);

    NewBatchJob job = new NewBatchJob();
    Mockito.when(mockBatchJobDAO.findBatchJobById(BATCHJOBID)).thenReturn(job);

    AbstractMessageReport messageReport = Mockito.mock(AbstractMessageReport.class);
    purgeProcessor.setMessageReport(messageReport);

    purgeProcessor.process(ex);
    Mockito.verify(messageReport, Mockito.atLeastOnce()).error(Matchers.any(ReportStats.class),
            Matchers.any(Source.class), Matchers.eq(CoreMessageCode.CORE_0035));
}
 

@Test
public void testNormalizeJson() throws Exception {
    final InputStream resource = getClass().getClassLoader().getResourceAsStream("bookstore.json");
    final String request = context().getTypeConverter().convertTo(String.class, resource);

    getMockEndpoint("mock:unknown").setExpectedMessageCount(0);
    getMockEndpoint("mock:csv").setExpectedMessageCount(0);
    getMockEndpoint("mock:xml").setExpectedMessageCount(0);

    getMockEndpoint("mock:json").expectedBodiesReceived(getExpectedBookstore());
    getMockEndpoint("mock:normalized").expectedBodiesReceived(getExpectedBookstore());

    template.sendBodyAndHeader("direct:start", request, Exchange.FILE_NAME, "bookstore.json");

    assertMockEndpointsSatisfied();
}
 

@Test
public void testUndertowLoadBalance() throws Exception {

    // Obtain the endpoint URL from the management model
    EndpointRegistryClient registryClient = new EndpointRegistryClient(managementClient.getControllerClient());
    final String[] endpoints = wrapEndpoints(registryClient.getRegisteredEndpoints("/undertow-load-balance"));

    CamelContext camelctx = new DefaultCamelContext();
    camelctx.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("direct:start").loadBalance().roundRobin().to(endpoints);
        }
    });

    camelctx.start();
    try {
        ProducerTemplate producer = camelctx.createProducerTemplate();
        String result = producer.requestBodyAndHeader("direct:start", null, Exchange.HTTP_QUERY, "name=Kermit", String.class);
        Assert.assertEquals("Hello Kermit", result);
    } finally {
        camelctx.close();
    }
}
 

@Override
public void process(Exchange exchange) throws Exception {
    System.out.println("Process the bean MnpRouting!");
    MnpRouting mnpRouting = (MnpRouting) exchange.getIn().getBody();
    if(mnpRouting != null){
        // validate phone numbers of format "1234567890"
        if (!mnpRouting.getTelephone().matches("\\d{11}") || mnpRouting.getCredit() < CREDIT_LIMIT){
            exchange.getOut().setBody("Message doesn't pass validation");
            exchange.getOut().setHeader("key", mnpRouting.getTelephone());
        } else{
            exchange.getOut().setBody(mnpRouting.toString());
            exchange.getOut().setHeader("key", mnpRouting.getTelephone());

        }

    }
    //System.out.println(mnpRouting);
}
 

@Deployment(resources = {"org/activiti/camel/variables/CamelVariableTransferTest.testCamelPropertiesAll.bpmn20.xml"})
public void testCamelPropertiesNone() throws Exception {
  ProducerTemplate tpl = camelContext.createProducerTemplate();
  Exchange exchange = camelContext.getEndpoint("direct:startNoProperties").createExchange();
  tpl.send("direct:startNoProperties", exchange);
  
  assertNotNull(taskService);
  assertNotNull(runtimeService);
  assertEquals(1, taskService.createTaskQuery().count());
  Task task = taskService.createTaskQuery().singleResult();
  assertNotNull(task);
  Map<String, Object> variables = runtimeService.getVariables(task.getExecutionId());
  assertNull(variables.get("property1"));
  assertNull(variables.get("property2"));
  assertNull(variables.get("property3"));
}
 

@Test
public void testNormalizeXml() throws Exception {
    final InputStream resource = getClass().getClassLoader().getResourceAsStream("bookstore.xml");
    final String request = context().getTypeConverter().convertTo(String.class, resource);

    getMockEndpoint("mock:unknown").setExpectedMessageCount(0);
    getMockEndpoint("mock:csv").setExpectedMessageCount(0);
    getMockEndpoint("mock:json").setExpectedMessageCount(0);

    getMockEndpoint("mock:xml").expectedBodiesReceived(getExpectedBookstore());
    getMockEndpoint("mock:normalized").expectedBodiesReceived(getExpectedBookstore());

    template.sendBodyAndHeader("direct:start", request, Exchange.FILE_NAME, "bookstore.xml");

    assertMockEndpointsSatisfied();
}
 

@Test
public void testOrderActiveMQFail() throws Exception {
    // we expect the order to fail and end up in the dead letter queue
    getMockEndpoint("mock:dead").expectedMessageCount(1);

    // we expect the file to be converted to csv and routed to the 2nd route
    MockEndpoint file = getMockEndpoint("mock:file");
    file.expectedMessageCount(1);

    // we do not expect the 2nd route to complete
    MockEndpoint mock = getMockEndpoint("mock:queue.order");
    mock.expectedMessageCount(0);

    template.sendBodyAndHeader("file://target/orders", "amount=1#name=ActiveMQ in Action", Exchange.FILE_NAME, "order.txt");

    // wait 5 seconds to let this test run
    Thread.sleep(5000);

    assertMockEndpointsSatisfied();
}
 

@Deployment(resources = {"org/activiti5/camel/variables/CamelVariableTransferTest.testCamelPropertiesAll.bpmn20.xml"})
public void testCamelHeadersNone() throws Exception {
  ProducerTemplate tpl = camelContext.createProducerTemplate();
  Exchange exchange = camelContext.getEndpoint("direct:startNoHeaders").createExchange();
  tpl.send("direct:startNoHeaders", exchange);
  
  assertNotNull(taskService);
  assertNotNull(runtimeService);
  assertEquals(1, taskService.createTaskQuery().count());
  Task task = taskService.createTaskQuery().singleResult();
  assertNotNull(task);
  Map<String, Object> variables = runtimeService.getVariables(task.getExecutionId());
  assertNull(variables.get("property1"));
  assertNull(variables.get("property2"));
  assertNull(variables.get("property3"));
}
 

@SuppressWarnings("try")
@Override
public boolean process(final Exchange exchange, final AsyncCallback callback) {
    final Message in = exchange.getIn();
    final Span activitySpan = exchange.getProperty(IntegrationLoggingConstants.ACTIVITY_SPAN, Span.class);
    try (Scope activityScope = tracer.scopeManager().activate(activitySpan)) {
        Span span = tracer.buildSpan(stepId).withTag(Tags.SPAN_KIND.getKey(), "step").start();
        in.setHeader(IntegrationLoggingConstants.STEP_SPAN, span);
        return super.process(exchange, doneSync -> {
            String failure = failure(exchange);
            if (failure != null) {
                span.setTag(Tags.ERROR.getKey(), true);
                span.log(failure);
            }
            span.finish();
            callback.done(doneSync);
        });
    }
}
 

@Test
public void testNormalizeJson() throws Exception {
    final InputStream resource = getClass().getClassLoader().getResourceAsStream("bookstore.json");
    final String request = context().getTypeConverter().convertTo(String.class, resource);

    getMockEndpoint("mock:unknown").setExpectedMessageCount(0);
    getMockEndpoint("mock:csv").setExpectedMessageCount(0);
    getMockEndpoint("mock:xml").setExpectedMessageCount(0);

    getMockEndpoint("mock:json").expectedBodiesReceived(getExpectedBookstore());
    getMockEndpoint("mock:normalized").expectedBodiesReceived(getExpectedBookstore());

    template.sendBodyAndHeader("direct:start", request, Exchange.FILE_NAME, "bookstore.json");

    assertMockEndpointsSatisfied();
}
 

private void assertBooksByCategory(Exchange exchange) {
    Message in = exchange.getIn();
    @SuppressWarnings("unchecked")
    Set<String> books = Collections.checkedSet(in.getBody(Set.class), String.class);
    String category = in.getHeader("category", String.class);
    switch (category) {
        case "Tech":
            assertTrue(books.containsAll(Collections.singletonList("Apache Camel Developer's Cookbook")));
            break;
        case "Cooking":
            assertTrue(books.containsAll(Arrays.asList("Camel Cookbook",
                "Double decadence with extra cream", "Cooking with Butter")));
            break;
        default:
            fail();
            break;
    }
}
 

@Deployment(resources = { "process/custom.bpmn20.xml" })
public void testRunProcess() throws Exception {
  CamelContext ctx = applicationContext.getBean(CamelContext.class);
  ProducerTemplate tpl = ctx.createProducerTemplate();
  service1.expectedBodiesReceived("ala");

  Exchange exchange = ctx.getEndpoint("direct:start").createExchange();
  exchange.getIn().setBody(Collections.singletonMap("var1", "ala"));
  tpl.send("direct:start", exchange);

  String instanceId = (String) exchange.getProperty("PROCESS_ID_PROPERTY");

  tpl.sendBodyAndProperty("direct:receive", null, ActivitiProducer.PROCESS_ID_PROPERTY, instanceId);

  assertProcessEnded(instanceId);

  service1.assertIsSatisfied();

  @SuppressWarnings("rawtypes")
  Map m = service2.getExchanges().get(0).getIn().getBody(Map.class);
  assertEquals("ala", m.get("var1"));
  assertEquals("var2", m.get("var2"));
}
 

@Test
public void testAsyncWssRoute() throws Exception {

    CamelContext camelctx = new DefaultCamelContext();
    camelctx.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("direct:start").to("ahc-wss:" + WEBSOCKET_ENDPOINT);
            from("ahc-wss:" + WEBSOCKET_ENDPOINT).to("seda:end");
        }
    });

    WsComponent wsComponent = (WsComponent) camelctx.getComponent("ahc-wss");
    wsComponent.setSslContextParameters(defineSSLContextClientParameters());

    PollingConsumer consumer = camelctx.getEndpoint("seda:end").createPollingConsumer();
    consumer.start();

    camelctx.start();
    try {
        ProducerTemplate producer = camelctx.createProducerTemplate();
        producer.sendBody("direct:start", "Kermit");

        Exchange exchange = consumer.receive(1000);
        Assert.assertEquals("Hello Kermit", exchange.getIn().getBody(String.class));

    } finally {
        camelctx.close();
    }
}
 

@Test
public void testReceive() throws Exception {
    Exchange exchange =  new DefaultExchange(new DefaultCamelContext());
    IngestionFileEntry entry = new IngestionFileEntry("/", FileFormat.EDFI_XML, FileType.XML_STUDENT_PROGRAM, "fileName", "111");
    FileEntryWorkNote workNote = new FileEntryWorkNote("batchJobId", entry, false);
    boolean fileEntryLatchOpened;

    exchange.getIn().setBody(workNote, FileEntryWorkNote.class);

    fileEntryLatchOpened = fileEntryLatch.lastFileProcessed(exchange);

    Assert.assertFalse(fileEntryLatchOpened);

    fileEntryLatchOpened = fileEntryLatch.lastFileProcessed(exchange);

    Assert.assertTrue(fileEntryLatchOpened);
}
 

/**
 * Strategy to create the unit of work to be used for the sub route
 *
 * @param routeContext the route context
 * @param processor    the processor
 * @param exchange     the exchange
 * @return the unit of work processor
 */
protected Processor createUnitOfWorkProcessor(RouteContext routeContext, Processor processor, Exchange exchange) {
    String routeId = routeContext != null ? routeContext.getRoute().idOrCreate(routeContext.getCamelContext().getNodeIdFactory()) : null;
    CamelInternalProcessor internal = new CamelInternalProcessor(processor);

    // and wrap it in a unit of work so the UoW is on the top, so the entire route will be in the same UoW
    UnitOfWork parent = exchange.getProperty(Exchange.PARENT_UNIT_OF_WORK, UnitOfWork.class);
    if (parent != null) {
        internal.addAdvice(new CamelInternalProcessor.ChildUnitOfWorkProcessorAdvice(routeId, parent));
    } else {
        internal.addAdvice(new CamelInternalProcessor.UnitOfWorkProcessorAdvice(routeId));
    }

    // and then in route context so we can keep track which route this is at runtime
    if (routeContext != null) {
        internal.addAdvice(new CamelInternalProcessor.RouteContextAdvice(routeContext));
    }
    return internal;
}
 

@Test
public void testOrderOk() throws Exception {
    // we do not expect any errors and therefore no messages in the dead letter queue
    getMockEndpoint("mock:dead").expectedMessageCount(0);

    // we expect the file to be converted to csv and routed to the 2nd route
    MockEndpoint file = getMockEndpoint("mock:file");
    file.expectedMessageCount(1);

    // we expect the 2nd route to complete
    MockEndpoint mock = getMockEndpoint("mock:queue.order");
    mock.expectedBodiesReceived("amount=1,name=Camel in Action,id=123,status=OK");

    template.sendBodyAndHeader("file://target/orders", "amount=1#name=Camel in Action", Exchange.FILE_NAME, "order.txt");

    assertMockEndpointsSatisfied();
}
 

@Override
public void process(final Exchange exchange) throws Exception {

    IsFinishedResponseProcessor.LOG.debug("Processing IsFinished response....");

    final String requestID = exchange.getIn().getHeader(Route.ID, String.class);

    IsFinishedResponseProcessor.LOG.debug("RequestID: {}", requestID);

    final Response response = exchange.getIn().getHeader(RestletConstants.RESTLET_RESPONSE, Response.class);

    if (exchange.getIn().getBody() instanceof Exception) {

        response.setStatus(Status.CLIENT_ERROR_NOT_FOUND);
        response.setEntity(exchange.getIn().getBody(String.class), MediaType.TEXT_ALL);
    } else {

        final Boolean isFinished = exchange.getIn().getBody(Boolean.class);

        if (isFinished) {
            IsFinishedResponseProcessor.LOG.debug("Invocation has finished, send location of result.");

            response.setStatus(Status.REDIRECTION_SEE_OTHER);
            response.setLocationRef(Route.GET_RESULT_ENDPOINT.replace(Route.ID_PLACEHODLER, requestID));
        } else {
            IsFinishedResponseProcessor.LOG.debug("Invocation has not finished yet.");

            final JSONObject obj = new JSONObject();
            obj.put("status", "PENDING");

            response.setStatus(Status.SUCCESS_OK);
            response.setEntity(obj.toJSONString(), MediaType.APPLICATION_JSON);
        }
        exchange.getOut().setBody(response);
    }
}
 

public void beforeProducer(final Exchange exchange) throws IOException {
    final Message in = exchange.getIn();
    in.setHeader(S3Constants.S3_OPERATION, S3Operations.deleteObject);

    if (filenameKey != null) {
        in.setHeader(S3Constants.KEY, filenameKey);
    }
}
 

@Override
public void configure() throws Exception {
    Processor markProcessed = new Processor() {
        @Override
        public void process(Exchange exchange) throws Exception {
            exchange.getIn().setHeader("processorAction", "triggered");
        }
    };

    from("direct:start")
        .wireTap("mock:tapped").onPrepare(markProcessed)
        .to("mock:out");
}
 

public boolean process(final Exchange exchange, final AsyncCallback callback) {
    // simulate async communication using a thread pool in which will return a reply in 5 seconds.
    executor.submit(new ERPTask(exchange, callback));

    // return false to tell Camel that we process asynchronously
    // which enables the Camel routing engine to know this and act accordingly
    // notice the ERPTask must invoke the callback.done(false) because what
    // we return here must match the boolean in the callback.done method.
    log.info("Returning false (processing will continue asynchronously)");
    return false;
}
 

private void beforeProducer(Exchange exchange) {
    JiraIssue bodyIssue;
    if (exchange.getIn().getBody() != null) {
        bodyIssue = (JiraIssue) exchange.getIn().getBody();
    } else {
        bodyIssue = new JiraIssue();
    }
    bodyIssue.replaceNullValues(jiraIssue);
    exchange.getIn().setHeader(ISSUE_KEY, bodyIssue.getIssueKey());
    if (bodyIssue.getIssueTypeId() != null) {
        exchange.getIn().setHeader(ISSUE_TYPE_ID, bodyIssue.getIssueTypeId());
    } else {
        exchange.getIn().setHeader(ISSUE_TYPE_NAME, bodyIssue.getIssueTypeName());
    }
    if (bodyIssue.getPriorityId() != null) {
        exchange.getIn().setHeader(ISSUE_PRIORITY_ID, bodyIssue.getPriorityId());
    } else {
        exchange.getIn().setHeader(ISSUE_PRIORITY_NAME, bodyIssue.getPriorityName());
    }
    if (bodyIssue.getSummary() != null) {
        exchange.getIn().setHeader(ISSUE_SUMMARY, bodyIssue.getSummary());
    }
    if (!bodyIssue.getComponents().isEmpty()) {
        exchange.getIn().setHeader(ISSUE_COMPONENTS, bodyIssue.getComponents());
    }
    if (!bodyIssue.getWatchers().isEmpty()) {
        exchange.getIn().setHeader(ISSUE_WATCHERS_ADD, bodyIssue.getWatchers());
    }
    if (bodyIssue.getAssignee() != null) {
        exchange.getIn().setHeader(ISSUE_ASSIGNEE, bodyIssue.getAssignee());
    }
    if (bodyIssue.getDescription() != null) {
        exchange.getIn().setBody(bodyIssue.getDescription());
    }
}
 

/**
 * Aggregates the messages.
 *
 * @param oldExchange  the existing aggregated message. Is <tt>null</tt> the
 *                     very first time as there are no existing message.
 * @param newExchange  the incoming message. This is never <tt>null</tt>.
 * @return the aggregated message.
 */
public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
    // the first time there are no existing message and therefore
    // the oldExchange is null. In these cases we just return
    // the newExchange
    if (oldExchange == null) {
        return newExchange;
    }

    // now we have both an existing message (oldExchange)
    // and a incoming message (newExchange)
    // we want to merge together.

    // in this example we add their bodies
    String oldBody = oldExchange.getIn().getBody(String.class);
    String newBody = newExchange.getIn().getBody(String.class);

    // if its the END message then do not aggregate
    if ("END".equals(newBody)) {
        return oldExchange;
    }

    // the body should be the two bodies added together
    String body = oldBody + newBody;

    // update the existing message with the added body
    oldExchange.getIn().setBody(body);
    // and return it
    return oldExchange;
}
 

@Override
public void process(final Exchange exchange) throws Exception {
	final String xmlRecord = exchange.getIn().getBody(String.class);
	try {
		final Document product =  builders.get().parse(new InputSource(new StringReader(xmlRecord)));
		exchange.getIn().setBody(product);
	} catch (final Exception exception) {
		logger.error(MessageCatalog._00039_STRING_TO_XML_FAILURE, exception);
		logger.debug(MessageCatalog._00039_STRING_TO_XML_FAILURE_DEBUG_MESSAGE, xmlRecord);
		throw exception;
	}
}