Java 类org.apache.camel.ExchangePattern 实例源码

@Override
public void configure() throws Exception {
    from("direct:STATUS")
            .onException(Exception.class)
                .maximumRedeliveries(0)
                .handled(true)
                .process(exchange -> {
                    Exception exception = (Exception) exchange.getProperty(Exchange.EXCEPTION_CAUGHT);
                    exchange.getOut().setBody(new ResponseMessage(ResponseCode.ERROR, exception.getMessage()));
                })
                .to("direct:marshal-response")
            .end()
            .setExchangePattern(ExchangePattern.InOut)
            .bean(hacep, "status()", false)
            .process(exchange -> {
                Object body = exchange.getIn().getBody();
                ResponseMessage output = new ResponseMessage(ResponseCode.SUCCESS, (String) body);
                exchange.getOut().setBody(output);
            })
            .to("direct:marshal-response");
}

/**
 * Generates a Camel route, that listens from any HTTP request made (GET or POST) regardless
 * of the path. The response resolution is delegated towards the response processor.
 */
@Override
public void configure() throws Exception {

    onException(Exception.class)
        .handled(true)
        .setHeader(Exchange.HTTP_RESPONSE_CODE, constant(500))
        .transform(simple("An error occurred: ${exception.message}"));

    from(generateJettyEndpoint())
        .routeId(ROUTE_ID_JETTY)
        .id(ROUTE_ID_JETTY)
        .log(LoggingLevel.DEBUG, LOGGER, "Received request...")
        .setExchangePattern(ExchangePattern.InOut)
        .to(DIRECT_MAIN);

    from(DIRECT_MAIN)
        .routeId(ROUTE_ID_MAIN)
        .id(ROUTE_ID_MAIN)
        .log(LoggingLevel.DEBUG, LOGGER, "Current headers: ${headers}")
        .process(this.responseProcessor);
}

/**
 * Configure the message route workflow.
 */
public void configure() throws Exception {

    // Distribute message based on headers.
    from("{{input.stream}}")
            .routeId("MessageBroadcaster")
            .description("Broadcast messages from one queue/topic to other queues/topics")
            .log(INFO, LOGGER,
                    "Distributing message: ${headers[JMSMessageID]} with timestamp ${headers[JMSTimestamp]}")
            .filter(header("IslandoraExchangePattern"))
                .process(exchange -> {
                    final String patternName = exchange.getIn().getHeader("IslandoraExchangePattern", String.class);
                    try {
                        exchange.setPattern(ExchangePattern.asEnum(patternName));
                    } catch (IllegalArgumentException e) {
                        LOGGER.warn("Ignoring malformed exchange pattern: " + patternName);
                    }
                })
                .end()
            .routingSlip(header("IslandoraBroadcastRecipients")).ignoreInvalidEndpoints();
}

@Override
public boolean apply(Event event) {
    Exchange exchange = endpoint.createExchange(ExchangePattern.InOnly);
    Message in = exchange.getIn();
    in.setBody(event);
    try {
        if (LOG.isTraceEnabled()) {
            LOG.trace("Processing Ignite Event: {}.", event);
        }
        getAsyncProcessor().process(exchange, new AsyncCallback() {
            @Override
            public void done(boolean doneSync) {
                // do nothing
            }
        });
    } catch (Exception e) {
        LOG.error(String.format("Exception while processing Ignite Event: %s.", event), e);
    }
    return true;
}

@Override
public boolean apply(UUID uuid, Object payload) {
    Exchange exchange = endpoint.createExchange(ExchangePattern.InOnly);
    Message in = exchange.getIn();
    in.setBody(payload);
    in.setHeader(IgniteConstants.IGNITE_MESSAGING_TOPIC, endpoint.getTopic());
    in.setHeader(IgniteConstants.IGNITE_MESSAGING_UUID, uuid);
    try {
        if (LOG.isTraceEnabled()) {
            LOG.trace("Processing Ignite message for subscription {} with payload {}.", uuid, payload);
        }
        getProcessor().process(exchange);
    } catch (Exception e) {
        LOG.error(String.format("Exception while processing Ignite Message from topic %s", endpoint.getTopic()), e);
    }
    return true;
}

@Test
public void testInvokingSimpleServerWithParams() throws Exception {
 // START SNIPPET: sending
    Exchange senderExchange = new DefaultExchange(context, ExchangePattern.InOut);
    final List<String> params = new ArrayList<String>();
    // Prepare the request message for the camel-cxf procedure
    params.add(TEST_MESSAGE);
    senderExchange.getIn().setBody(params);
    senderExchange.getIn().setHeader(CxfConstants.OPERATION_NAME, ECHO_OPERATION);

    Exchange exchange = template.send("direct:EndpointA", senderExchange);

    org.apache.camel.Message out = exchange.getOut();
    // The response message's body is an MessageContentsList which first element is the return value of the operation,
    // If there are some holder parameters, the holder parameter will be filled in the reset of List.
    // The result will be extract from the MessageContentsList with the String class type
    MessageContentsList result = (MessageContentsList)out.getBody();
    LOG.info("Received output text: " + result.get(0));
    Map<String, Object> responseContext = CastUtils.cast((Map<?, ?>)out.getHeader(Client.RESPONSE_CONTEXT));
    assertNotNull(responseContext);
    assertEquals("We should get the response context here", "UTF-8", responseContext.get(org.apache.cxf.message.Message.ENCODING));
    assertEquals("Reply body on Camel is wrong", "echo " + TEST_MESSAGE, result.get(0));
 // END SNIPPET: sending
}

@Test
public void createOnAcceptAlertNotificationExchangeWithExchangePattern() {
    AlertNotification alertNotification = createMock(AlertNotification.class);
    SmppMessage message = createMock(SmppMessage.class);
    expect(binding.createSmppMessage(alertNotification)).andReturn(message);
    message.setExchange(isA(Exchange.class));

    replay(alertNotification, binding, message);

    Exchange exchange = endpoint.createOnAcceptAlertNotificationExchange(ExchangePattern.InOut, alertNotification);

    verify(alertNotification, binding, message);

    assertSame(binding, exchange.getProperty(Exchange.BINDING));
    assertSame(message, exchange.getIn());
    assertSame(ExchangePattern.InOut, exchange.getPattern());
}

public static Exchange toExchange(Endpoint endpoint, SessionID sessionID, Message message, QuickfixjEventCategory eventCategory, ExchangePattern exchangePattern) {
    Exchange exchange = endpoint.createExchange(exchangePattern);

    org.apache.camel.Message camelMessage = exchange.getIn();
    camelMessage.setHeader(EVENT_CATEGORY_KEY, eventCategory);
    camelMessage.setHeader(SESSION_ID_KEY, sessionID);

    if (message != null) {
        try {
            camelMessage.setHeader(MESSAGE_TYPE_KEY, message.getHeader().getString(MsgType.FIELD));
        } catch (FieldNotFound e) {
            LOG.warn("Message type field not found in QFJ message: {}, continuing...", message);
        }
    }
    camelMessage.setBody(message);

    return exchange;
}

public void testAsyncProducerWait() throws Exception {
    MockEndpoint mock = getMockEndpoint("mock:result");
    mock.expectedMessageCount(1);

    // using the new async API we can fire a real async message
    Exchange exchange = new DefaultExchange(context);
    exchange.getIn().setBody("Hello World");
    exchange.setPattern(ExchangePattern.InOut);
    exchange.setProperty(Exchange.ASYNC_WAIT, WaitForTaskToComplete.IfReplyExpected);
    template.send("direct:start", exchange);

    // I should not happen before mock
    route = route + "send";

    assertMockEndpointsSatisfied();

    assertEquals("Send should occur before processor", "processsend", route);

    String response = exchange.getOut().getBody(String.class);
    assertEquals("Bye World", response);
}

@Test
public void testXPath() throws Exception {
    Endpoint directEndpoint = context.getEndpoint("direct:input");
    Exchange exchange = directEndpoint.createExchange(ExchangePattern.InOnly);
    Message message = exchange.getIn();
    String str1 = "<person name='David' city='Rome'/>";
    message.setBody(str1, byte[].class);
    Producer producer = directEndpoint.createProducer();
    producer.start();
    producer.process(exchange);
    String str2 = "<person name='James' city='London'/>";
    message.setBody(str2, byte[].class);
    producer.process(exchange);
    latch = new CountDownLatch(1);
    latch.await();
}

@Test
public void testConsumer() throws Exception {
    if (MtomTestHelper.isAwtHeadless(logger, null)) {
        return;
    }

    context.createProducerTemplate().send("cxf:bean:consumerEndpoint", new Processor() {

        public void process(Exchange exchange) throws Exception {
            exchange.setPattern(ExchangePattern.InOut);
            assertEquals("Get a wrong Content-Type header", "application/xop+xml", exchange.getIn().getHeader("Content-Type"));
            List<Source> elements = new ArrayList<Source>();
            elements.add(new DOMSource(StaxUtils.read(new StringReader(getRequestMessage())).getDocumentElement()));
            CxfPayload<SoapHeader> body = new CxfPayload<SoapHeader>(new ArrayList<SoapHeader>(),
                elements, null);
            exchange.getIn().setBody(body);
            exchange.getIn().addAttachment(MtomTestHelper.REQ_PHOTO_CID, 
                new DataHandler(new ByteArrayDataSource(MtomTestHelper.REQ_PHOTO_DATA, "application/octet-stream")));

            exchange.getIn().addAttachment(MtomTestHelper.REQ_IMAGE_CID, 
                new DataHandler(new ByteArrayDataSource(MtomTestHelper.requestJpeg, "image/jpeg")));
        }
    });
}

private RouteBuilder createRouteBuilder() {
    return new RouteBuilder() {
        public void configure() {
            CometdComponent component = (CometdComponent) context.getComponent("cometds");
            component.setSslPassword(pwd);
            component.setSslKeyPassword(pwd);
            File file = new File("./src/test/resources/jsse/localhost.ks");
            URI keyStoreUrl = file.toURI();
            component.setSslKeystore(keyStoreUrl.getPath());

            from(URI, URIS).setExchangePattern(ExchangePattern.InOut).process(new Processor() {
                public void process(Exchange exchange) throws Exception {
                    Message out = new DefaultMessage();
                    out.setBody("reply: " + exchange.getIn().getBody());
                    exchange.setOut(out);
                }
            });
        }
    };
}

public Exchange dispatchAsync(RouteboxEndpoint endpoint, Exchange exchange) throws Exception {
    URI dispatchUri;
    Exchange reply;

    if (LOG.isDebugEnabled()) {
        LOG.debug("Dispatching exchange {} to endpoint {}", exchange, endpoint.getEndpointUri());
    }

    dispatchUri = selectDispatchUri(endpoint, exchange);

    if (exchange.getPattern() == ExchangePattern.InOnly) {
        producer.asyncSend(dispatchUri.toASCIIString(), exchange);
        reply = exchange;
    } else {
        Future<Exchange> future = producer.asyncCallback(dispatchUri.toASCIIString(), exchange, new SynchronizationAdapter());
        reply = future.get(endpoint.getConfig().getConnectionTimeout(), TimeUnit.MILLISECONDS);
    }

    return reply;
}

@Test
public void execute() throws Exception {
    Exchange exchange = new DefaultExchange(new DefaultCamelContext(), ExchangePattern.InOut);
    exchange.getIn().setHeader(SmppConstants.COMMAND, "DataSm");
    exchange.getIn().setHeader(SmppConstants.SERVICE_TYPE, "XXX");
    exchange.getIn().setHeader(SmppConstants.SOURCE_ADDR_TON, TypeOfNumber.NATIONAL.value());
    exchange.getIn().setHeader(SmppConstants.SOURCE_ADDR_NPI, NumberingPlanIndicator.NATIONAL.value());
    exchange.getIn().setHeader(SmppConstants.SOURCE_ADDR, "1818");
    exchange.getIn().setHeader(SmppConstants.DEST_ADDR_TON, TypeOfNumber.INTERNATIONAL.value());
    exchange.getIn().setHeader(SmppConstants.DEST_ADDR_NPI, NumberingPlanIndicator.INTERNET.value());
    exchange.getIn().setHeader(SmppConstants.DEST_ADDR, "1919");
    exchange.getIn().setHeader(SmppConstants.REGISTERED_DELIVERY, new RegisteredDelivery(SMSCDeliveryReceipt.FAILURE).value());
    expect(session.dataShortMessage(eq("XXX"), eq(TypeOfNumber.NATIONAL), eq(NumberingPlanIndicator.NATIONAL), eq("1818"),
            eq(TypeOfNumber.INTERNATIONAL), eq(NumberingPlanIndicator.INTERNET), eq("1919"), eq(new ESMClass()),
            eq(new RegisteredDelivery((byte) 2)), eq(DataCodings.newInstance((byte) 0))))
        .andReturn(new DataSmResult(new MessageId("1"), null));

    replay(session);

    command.execute(exchange);

    verify(session);

    assertEquals("1", exchange.getOut().getHeader(SmppConstants.ID));
    assertNull(exchange.getOut().getHeader(SmppConstants.OPTIONAL_PARAMETERS));
}

@Override
public void process(Exchange exchange) throws Exception {
    exchange.setPattern(ExchangePattern.InOut);
    Message inMessage = exchange.getIn();
    setupDestinationURL(inMessage);
    // using the http central client API
    inMessage.setHeader(CxfConstants.CAMEL_CXF_RS_USING_HTTP_API, Boolean.TRUE);
    // set the Http method
    inMessage.setHeader(Exchange.HTTP_METHOD, "GET");
    // set the relative path
    inMessage.setHeader(Exchange.HTTP_PATH, "/customerservice/customers/123");
    // Specify the response class , cxfrs will use InputStream as the response object type
    inMessage.setHeader(CxfConstants.CAMEL_CXF_RS_RESPONSE_CLASS, Customer.class);
    // set a customer header
    inMessage.setHeader("key", "value");
    // since we use the Get method, so we don't need to set the message body
    inMessage.setBody(null);
}

public void testExpectedExchangePattern() throws Exception {
    MockEndpoint mock = getMockEndpoint("mock:result");
    mock.expectedMessageCount(1);
    mock.expectedExchangePattern(ExchangePattern.InOnly);

    template.sendBody("direct:a", "Hello World");

    assertMockEndpointsSatisfied();

    // reset and try with InOut this time
    resetMocks();
    mock.expectedMessageCount(1);
    mock.expectedExchangePattern(ExchangePattern.InOut);

    template.requestBody("direct:a", "Bye World");

    assertMockEndpointsSatisfied();
}

private org.apache.camel.Exchange prepareExchange(Exchange cxfExchange, Method method,
        Object[] paramArray, Object response) {
    ExchangePattern ep = ExchangePattern.InOut;
    if (method.getReturnType() == Void.class) {
        ep = ExchangePattern.InOnly;
    } 
    final org.apache.camel.Exchange camelExchange = endpoint.createExchange(ep);
    if (response != null) {
        camelExchange.getOut().setBody(response);
    }
    CxfRsBinding binding = endpoint.getBinding();
    binding.populateExchangeFromCxfRsRequest(cxfExchange, camelExchange, method, paramArray);

    // REVISIT: It can be done inside a binding but a propagateContext would need to be passed along as
    // the CXF in message property. Question: where should this property name be set up ? 
    if (endpoint.isPropagateContexts()) {
        camelExchange.setProperty(UriInfo.class.getName(), new UriInfoImpl(cxfExchange.getInMessage()));
        camelExchange.setProperty(Request.class.getName(), new RequestImpl(cxfExchange.getInMessage()));
        camelExchange.setProperty(HttpHeaders.class.getName(), new HttpHeadersImpl(cxfExchange.getInMessage()));
        camelExchange.setProperty(SecurityContext.class.getName(), new SecurityContextImpl(cxfExchange.getInMessage()));
    }

    return camelExchange;
}

@Before
public void setUp() throws ConfigError, FieldConvertError, IOException, JMException {
    mockExchange = Mockito.mock(Exchange.class);
    mockEndpoint = Mockito.mock(QuickfixjEndpoint.class);
    mockCamelMessage = Mockito.mock(org.apache.camel.Message.class);
    Mockito.when(mockExchange.getIn()).thenReturn(mockCamelMessage);
    Mockito.when(mockExchange.getPattern()).thenReturn(ExchangePattern.InOnly);

    quickfixjEngine = TestSupport.createEngine();
    Mockito.when(mockEndpoint.getEngine()).thenReturn(quickfixjEngine);

    inboundFixMessage = new Message();
    inboundFixMessage.getHeader().setString(BeginString.FIELD, FixVersions.BEGINSTRING_FIX44);
    inboundFixMessage.getHeader().setString(SenderCompID.FIELD, "SENDER");
    inboundFixMessage.getHeader().setString(TargetCompID.FIELD, "TARGET");
    sessionID = MessageUtils.getSessionID(inboundFixMessage);

    Mockito.when(mockCamelMessage.getBody(Message.class)).thenReturn(inboundFixMessage);

    Mockito.when(mockEndpoint.getSessionID()).thenReturn(sessionID);     

    producer = Mockito.spy(new QuickfixjProducer(mockEndpoint));
}

@Test
public void execute() throws Exception {
    Exchange exchange = new DefaultExchange(new DefaultCamelContext(), ExchangePattern.InOut);
    exchange.getIn().setHeader(SmppConstants.COMMAND, "CancelSm");
    exchange.getIn().setHeader(SmppConstants.ID, "1");
    exchange.getIn().setHeader(SmppConstants.SERVICE_TYPE, "XXX");
    exchange.getIn().setHeader(SmppConstants.SOURCE_ADDR_TON, TypeOfNumber.NATIONAL.value());
    exchange.getIn().setHeader(SmppConstants.SOURCE_ADDR_NPI, NumberingPlanIndicator.NATIONAL.value());
    exchange.getIn().setHeader(SmppConstants.SOURCE_ADDR, "1818");
    exchange.getIn().setHeader(SmppConstants.DEST_ADDR_TON, TypeOfNumber.INTERNATIONAL.value());
    exchange.getIn().setHeader(SmppConstants.DEST_ADDR_NPI, NumberingPlanIndicator.INTERNET.value());
    exchange.getIn().setHeader(SmppConstants.DEST_ADDR, "1919");
    session.cancelShortMessage("XXX", "1", TypeOfNumber.NATIONAL, NumberingPlanIndicator.NATIONAL, "1818", TypeOfNumber.INTERNATIONAL, NumberingPlanIndicator.INTERNET, "1919");

    replay(session);

    command.execute(exchange);

    verify(session);

    assertEquals("1", exchange.getOut().getHeader(SmppConstants.ID));
}

@Test
public void testSchematronFileReadFromFileSystem()throws Exception {

    String payload = IOUtils.toString(ClassLoader.getSystemResourceAsStream("xml/article-2.xml"));
    String path = ClassLoader.getSystemResource("sch/schematron-1.sch").getPath();
    Endpoint endpoint = context().getEndpoint("schematron://" + path);
    Producer producer = endpoint.createProducer();
    Exchange exchange = new DefaultExchange(context, ExchangePattern.InOut);

    exchange.getIn().setBody(payload);

    // invoke the component.
    producer.process(exchange);

    String report = exchange.getOut().getHeader(Constants.VALIDATION_REPORT, String.class);
    assertNotNull(report);
}

public void testPreserveOldMEPInOut() throws Exception {
    // the mock should get an InOut MEP
    getMockEndpoint("mock:result").expectedMessageCount(1);
    getMockEndpoint("mock:result").message(0).exchangePattern().isEqualTo(ExchangePattern.InOut);

    // we send an InOnly
    Exchange out = template.send("direct:testInOut", new Processor() {
        @Override
        public void process(Exchange exchange) throws Exception {
            exchange.getIn().setBody("Hello World");
            exchange.setPattern(ExchangePattern.InOnly);
        }
    });

    // the MEP should be preserved
    assertNotNull(out);
    assertEquals(ExchangePattern.InOnly, out.getPattern());

    assertMockEndpointsSatisfied();
}

@Test
public void createOnAcceptDeliverSmWithExchangePattern() throws Exception {
    DeliverSm deliverSm = createMock(DeliverSm.class);
    SmppMessage message = createMock(SmppMessage.class);
    expect(binding.createSmppMessage(deliverSm)).andReturn(message);
    message.setExchange(isA(Exchange.class));

    replay(deliverSm, binding, message);

    Exchange exchange = endpoint.createOnAcceptDeliverSmExchange(ExchangePattern.InOut, deliverSm);

    verify(deliverSm, binding, message);

    assertSame(binding, exchange.getProperty(Exchange.BINDING));
    assertSame(message, exchange.getIn());
    assertSame(ExchangePattern.InOut, exchange.getPattern());
}

@Test
public void sendLongSubmitSM() throws Exception {
    result.expectedMessageCount(2);

    Exchange exchange = start.createExchange(ExchangePattern.InOnly);
    exchange.getIn().setBody("Hello SMPP World! Hello SMPP World! Hello SMPP World! Hello SMPP World! Hello SMPP World! "
            + "Hello SMPP World! Hello SMPP World! Hello SMPP World! Hello SMPP World! Hello SMPP World! "
            + "Hello SMPP World! Hello SMPP World! Hello SMPP World! Hello SMPP World! Hello SMPP World! "); // 270 chars

    template.send(start, exchange);

    assertMockEndpointsSatisfied();
    assertEquals(SmppMessageType.DeliveryReceipt.toString(), result.getExchanges().get(0).getIn().getHeader(SmppConstants.MESSAGE_TYPE));
    assertEquals(SmppMessageType.DeliveryReceipt.toString(), result.getExchanges().get(1).getIn().getHeader(SmppConstants.MESSAGE_TYPE));

    assertNotNull(exchange.getIn().getHeader(SmppConstants.ID));
    assertEquals(2, exchange.getIn().getHeader(SmppConstants.SENT_MESSAGE_COUNT));
}

public void run() {
    try {
        LOG.debug("Waiting for latch");
        latch.await();

        // wait 1 sec after latch before sending he late replay
        Thread.sleep(1000);
    } catch (Exception e) {
        // ignore
    }

    LOG.debug("Sending late reply");
    template.send(componentName + ":" + myReplyTo, new Processor() {
        public void process(Exchange exchange) throws Exception {
            exchange.setPattern(ExchangePattern.InOnly);
            exchange.getIn().setBody("Late reply");
        }
    });
}

/**
 * When the setting useMessageIdAsCorrelationid is true and
 * a correlation id is set on the message then we expect the reply
 * to contain the messageId of the sent message. Here we test only that
 * it is not the correlation id given as the messageId is not know
 * beforehand.
 */
@Test
public void testRequestReplyCorrelationByMessageId() throws Exception {
    MockEndpoint result = getMockEndpoint("mock:result");
    result.expectedMessageCount(1);

    Exchange out = template.send("jms2:queue:hello2", ExchangePattern.InOut, new Processor() {
        public void process(Exchange exchange) throws Exception {
            Message in = exchange.getIn();
            in.setBody("Hello World");
            in.setHeader("JMSCorrelationID", "a");
        }
    });

    result.assertIsSatisfied();

    assertNotNull(out);

    assertEquals(REPLY_BODY, out.getOut().getBody(String.class));
    assertEquals("a", out.getOut().getHeader("JMSCorrelationID"));
}

@SuppressWarnings("el-syntax")
@Override
public void configure() throws Exception {
    from("netty4-http:http://0.0.0.0:9090/produceJmsMessage")
        .routeId("jms-message-producer-route")
        .streamCaching()
        .setProperty("start", simple("${date:now:YYYYMMDD HH:mm:ss.SSS}"))
        .filter(header("destination").isNull())
            .transform().method("informationBean","transactedUsage()")
            .stop()
        .end()
        .process("jmsSendPropertiesProcessor") // Prepare JMS Send Properties
        .log(LoggingLevel.INFO, 
                "Publishing ${exchangeProperty.messageCount} messages...")
        .setExchangePattern(ExchangePattern.InOnly)
        .loopDoWhile(simple("${exchangeProperty.messageCount} > 0"))
            .process("jmsTransactionBatchProcessor") // Prepare Transaction Batch to send
            .log(LoggingLevel.INFO, 
                    "Batch size: ${exchangeProperty.transactionBatchSize} - [${exchangeProperty.sjmsUri}]")
            .recipientList(exchangeProperty("sjmsUri")).end()
        .end()
        .setProperty("end", simple("${date:now:YYYYMMDD HH:mm:ss.SSS}"))
        .setProperty("duration").method("informationBean",
                     "getDuration(${exchangeProperty.start},${exchangeProperty.end})")
        .log(LoggingLevel.INFO, 
                "Publishing ${exchangeProperty.messageToSend} messages DONE in ${exchangeProperty.duration}")
        .setExchangePattern(ExchangePattern.InOut)
        .process("prepareHttpResponseProcessor");
}

@Override
public Task add(TaskOptions task) {
  long etaMillis = task.getEtaMillis() - System.currentTimeMillis();
  if (etaMillis > 0) {
    DelayScheduler.getInstance().schedule(task, Constants.DEFAULT);
  } else {
    // get the camel template for Spring template style sending of messages (= producer)
    ProducerTemplate
        camelTemplate =
        AppFactory.get().getTaskService().getContext()
            .getBean("camel-client", ProducerTemplate.class);
    camelTemplate.sendBody("direct:tasks", ExchangePattern.InOnly, task);
  }
  return new SimpleTask(task);
}

public static void generate(Long domainId, int eventId, Map<String, Object> params, String objectType, String objectId,
                     CustomOptions customOptions, Object eventObject) throws EventGenerationException {
  EventData eventData = new EventData(domainId,eventId,params, objectType, objectId,
      customOptions, eventObject);
  try {
    ProducerTemplate
        camelTemplate =
        getContext()
            .getBean("camel-events-gen", ProducerTemplate.class);
    camelTemplate.sendBody("direct:events", ExchangePattern.InOnly, eventData);
  }catch (Exception e){
    throw new EventGenerationException(e);
  }
}

@Read
public EpisodeOfCare getEpisodeOfCareById(HttpServletRequest theRequest, @IdParam IdType internalId) {

    ProducerTemplate template = context.createProducerTemplate();



    EpisodeOfCare episode = null;
    IBaseResource resource = null;
    try {
        InputStream inputStream = (InputStream)  template.sendBody("direct:FHIREpisodeOfCare",
                ExchangePattern.InOut,theRequest);


        Reader reader = new InputStreamReader(inputStream);
        resource = ctx.newJsonParser().parseResource(reader);
    } catch(Exception ex) {
        log.error("JSON Parse failed " + ex.getMessage());
        throw new InternalErrorException(ex.getMessage());
    }
    if (resource instanceof EpisodeOfCare) {
        episode = (EpisodeOfCare) resource;
    }else if (resource instanceof OperationOutcome)
    {

        OperationOutcome operationOutcome = (OperationOutcome) resource;
        log.info("Sever Returned: "+ctx.newJsonParser().encodeResourceToString(operationOutcome));

        OperationOutcomeFactory.convertToException(operationOutcome);
    } else {
        throw new InternalErrorException("Unknown Error");
    }


    return episode;
}

@Read
public PractitionerRole getPractitionerRoleById(HttpServletRequest theRequest, @IdParam IdType internalId) {

    ProducerTemplate template = context.createProducerTemplate();

    PractitionerRole practitionerRole = null;
    IBaseResource resource = null;
    try {
        InputStream inputStream = (InputStream)  template.sendBody("direct:FHIRPractitionerRole",
                ExchangePattern.InOut,theRequest);

        Reader reader = new InputStreamReader(inputStream);
        resource = ctx.newJsonParser().parseResource(reader);

    } catch(Exception ex) {
        log.error("JSON Parse failed " + ex.getMessage());
        throw new InternalErrorException(ex.getMessage());
    }
    if (resource instanceof PractitionerRole) {
        practitionerRole = (PractitionerRole) resource;
    } else if (resource instanceof OperationOutcome)
    {

        OperationOutcome operationOutcome = (OperationOutcome) resource;
        log.info("Sever Returned: "+ctx.newJsonParser().encodeResourceToString(operationOutcome));

        OperationOutcomeFactory.convertToException(operationOutcome);
    } else {
        throw new InternalErrorException("Unknown Error");
    }

    return practitionerRole;
}

@Override
public void configure() throws Exception {

    String uri = "log:c";

    from("direct:start")
        .toD("log:a", true)
        .to(ExchangePattern.InOnly, "log:b")
        .to(uri);
}

@Override
public void configure() throws Exception {
    from("jms:" + queueName)
            .setExchangePattern(ExchangePattern.InOut)
            .to("log:it.redhat.hacep.camel.Router?level=INFO&showAll=true&multiline=true")
            .unmarshal().json(JsonLibrary.Jackson, Command.class)
            .to("direct:execute-command");

    from("direct:execute-command")
            .setExchangePattern(ExchangePattern.InOut)
            .recipientList()
            .simple("direct:${body.command}");
}

public void testTypeConstant() throws Exception {
    assertExpression("${type:org.apache.camel.Exchange.FILE_NAME}", Exchange.FILE_NAME);
    assertExpression("${type:org.apache.camel.ExchangePattern.InOut}", ExchangePattern.InOut);

    // non existing fields
    assertExpression("${type:org.apache.camel.ExchangePattern.}", null);
    assertExpression("${type:org.apache.camel.ExchangePattern.UNKNOWN}", null);
}

@Override
public Exchange createExchange(ExchangePattern pattern) {
    if (delegate == null) {
        throw new IllegalStateException("Not started");
    }
    return delegate.createExchange(pattern);
}

@Test
public void testUsingJMStoJMStoBean() throws Exception {
    // the big one from jms to jms to test that we do not lost the bean method name
    MockEndpoint mock = getMockEndpoint("mock:approve");
    mock.expectedBodiesReceived("No");

    template.sendBodyAndHeader("activemq:queue", ExchangePattern.InOut, "James",
        Exchange.BEAN_METHOD_NAME, "approveSuperLoan");

    mock.assertIsSatisfied();
}

@Test
public void eightBitDataCodingOverridesDefaultAlphabet() throws Exception {
    final byte binDataCoding = (byte)0x04; /* SMPP 8-bit */
    byte[] body = {(byte)0xFF, 'A', 'B', (byte)0x00, (byte)0xFF, (byte)0x7F, 'C', (byte)0xFF};

    Exchange exchange = new DefaultExchange(new DefaultCamelContext(), ExchangePattern.InOut);
    exchange.getIn().setHeader(SmppConstants.COMMAND, "SubmitSm");
    exchange.getIn().setHeader(SmppConstants.ALPHABET, Alphabet.ALPHA_DEFAULT.value());
    exchange.getIn().setHeader(SmppConstants.DATA_CODING, binDataCoding);
    exchange.getIn().setBody(body);
    expect(session.submitShortMessage(eq("CMT"),
                                      eq(TypeOfNumber.UNKNOWN),
                                      eq(NumberingPlanIndicator.UNKNOWN),
                                      eq("1616"),
                                      eq(TypeOfNumber.UNKNOWN),
                                      eq(NumberingPlanIndicator.UNKNOWN),
                                      eq("1717"),
                                      eq(new ESMClass()),
                                      eq((byte) 0),
                                      eq((byte) 1),
                                      (String) isNull(),
                                      (String) isNull(),
                                      eq(new RegisteredDelivery(SMSCDeliveryReceipt.SUCCESS_FAILURE)),
                                      eq(ReplaceIfPresentFlag.DEFAULT.value()),
                                      eq(DataCodings.newInstance(binDataCoding)),
                                      eq((byte) 0),
                                      aryEq(body)))
        .andReturn("1");

    replay(session);

    command.execute(exchange);

    verify(session);
}

@Test
public void testCamelEndpointInvocation() throws Exception {
    // get the endpoint from the camel context
    Endpoint endpoint = context.getEndpoint("jms:queue:numbers");

    // create the exchange used for the communication
    // we use the in out pattern for a synchronized exchange where we expect a response
    Exchange exchange = endpoint.createExchange(ExchangePattern.InOut);
    // set the input on the in body
    // must you correct type to match the expected type of an Integer object
    exchange.getIn().setBody(11);

    // to send the exchange we need an producer to do it for us
    Producer producer = endpoint.createProducer();
    // start the producer so it can operate
    producer.start();

    // let the producer process the exchange where it does all the work in this one line of code
    producer.process(exchange);

    // get the response from the out body and cast it to an integer
    int response = exchange.getOut().getBody(Integer.class);

    assertEquals("Get a wrong response.", 33, response);

    // stop the producer after usage
    producer.stop();
}

@Test
public void testInvokingSimpleServerWithMessageDataFormat() throws Exception {
    Exchange senderExchange = new DefaultExchange(context, ExchangePattern.InOut);
    senderExchange.getIn().setBody(REQUEST_MESSAGE);
    Exchange exchange = template.send("direct:EndpointB", senderExchange);

    org.apache.camel.Message out = exchange.getOut();
    String response = out.getBody(String.class);
    assertTrue("It should has the echo message", response.indexOf("echo " + TEST_MESSAGE) > 0);
    assertTrue("It should has the echoResponse tag", response.indexOf("echoResponse") > 0);

}

@Test
public void latin1DataCodingOverridesEightBitAlphabet() throws Exception {
    final byte latin1DataCoding = (byte)0x03; /* ISO-8859-1 (Latin1) */
    byte[] body = {(byte)0xFF, 'A', 'B', (byte)0x00, (byte)0xFF, (byte)0x7F, 'C', (byte)0xFF};
    byte[] bodyNarrowed = {'?', 'A', 'B', '\0', '?', (byte)0x7F, 'C', '?'};

    Exchange exchange = new DefaultExchange(new DefaultCamelContext(), ExchangePattern.InOut);
    exchange.getIn().setHeader(SmppConstants.COMMAND, "SubmitSm");
    exchange.getIn().setHeader(SmppConstants.ALPHABET, Alphabet.ALPHA_8_BIT.value());
    exchange.getIn().setHeader(SmppConstants.DATA_CODING, latin1DataCoding);
    exchange.getIn().setBody(body);
    expect(session.submitShortMessage(eq("CMT"),
                                      eq(TypeOfNumber.UNKNOWN),
                                      eq(NumberingPlanIndicator.UNKNOWN),
                                      eq("1616"),
                                      eq(TypeOfNumber.UNKNOWN),
                                      eq(NumberingPlanIndicator.UNKNOWN),
                                      eq("1717"),
                                      eq(new ESMClass()),
                                      eq((byte) 0),
                                      eq((byte) 1),
                                      (String) isNull(),
                                      (String) isNull(),
                                      eq(new RegisteredDelivery(SMSCDeliveryReceipt.SUCCESS_FAILURE)),
                                      eq(ReplaceIfPresentFlag.DEFAULT.value()),
                                      eq(DataCodings.newInstance(latin1DataCoding)),
                                      eq((byte) 0),
                                      aryEq(bodyNarrowed)))
        .andReturn("1");

    replay(session);

    command.execute(exchange);

    verify(session);
}