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

@Test
public void testWithDirectEndpoint(TestContext context) throws Exception {
  Async async = context.async();
  Endpoint endpoint = camel.getEndpoint("direct:foo");

  bridge = CamelBridge.create(vertx, new CamelBridgeOptions(camel)
      .addInboundMapping(fromCamel("direct:foo").toVertx("test")));

  vertx.eventBus().consumer("test", message -> {
    context.assertEquals("hello", message.body());
    async.complete();
  });

  camel.start();
  BridgeHelper.startBlocking(bridge);

  ProducerTemplate producer = camel.createProducerTemplate();
  producer.asyncSendBody(endpoint, "hello");
}

@Test
@DirtiesContext
public void testMocksAreValid() throws Exception {
    result.setExpectedCount(1);

    ProducerTemplate producerTemplate = camelContext.createProducerTemplate();
    producerTemplate.sendBody("direct:start", Util.generateMockTwitterStatus());

    MockEndpoint.assertIsSatisfied(camelContext);
    Object body = result.getExchanges().get(0).getIn().getBody();
    assertEquals(String.class, body.getClass());
    ObjectMapper mapper = new ObjectMapper();
    JsonNode outJson = mapper.readTree((String)body);
    assertEquals("Bob", outJson.get("FirstName").asText());
    assertEquals("Vila", outJson.get("LastName").asText());
    assertEquals("bobvila1982", outJson.get("Title").asText());
    assertEquals("Let's build a house!", outJson.get("Description").asText());
}

@Test
@DirtiesContext
public void testMocksAreValid() throws Exception {
    result.setExpectedCount(1);

    ProducerTemplate producerTemplate = camelContext.createProducerTemplate();
    producerTemplate.sendBody("direct:start", Util.generateMockTwitterStatus());

    MockEndpoint.assertIsSatisfied(camelContext);
    Object body = result.getExchanges().get(0).getIn().getBody();
    assertEquals(String.class, body.getClass());
    ObjectMapper mapper = new ObjectMapper();
    JsonNode sfJson = mapper.readTree((String)body);
    assertNotNull(sfJson.get("TwitterScreenName__c"));
    assertEquals("bobvila1982", sfJson.get("TwitterScreenName__c").asText());
}

@Test
@DirtiesContext
public void testMocksAreValid() throws Exception {
    result.setExpectedCount(1);

    ProducerTemplate producerTemplate = camelContext.createProducerTemplate();
    producerTemplate.sendBody("direct:start", Util.generateMockTwitterStatus());

    MockEndpoint.assertIsSatisfied(camelContext);
    Object body = result.getExchanges().get(0).getIn().getBody();
    assertEquals(String.class, body.getClass());
    ObjectMapper mapper = new ObjectMapper();
    JsonNode outJson = mapper.readTree((String)body);
    assertEquals("Bob", outJson.get("FirstName").asText());
    assertEquals("Vila", outJson.get("LastName").asText());
    assertEquals("bobvila1982", outJson.get("Title").asText());
    assertEquals("Let's build a house!", outJson.get("Description").asText());
}

@Test
@DirtiesContext
public void testSeparateNotSucceed() throws Exception {
    result.setExpectedCount(1);

    ProducerTemplate producerTemplate = camelContext.createProducerTemplate();
    Status s = Util.generateMockTwitterStatus();
    when(s.getUser().getName()).thenReturn("BobVila");
    producerTemplate.sendBody("direct:start", s);

    MockEndpoint.assertIsSatisfied(camelContext);
    Object body = result.getExchanges().get(0).getIn().getBody();
    assertEquals(String.class, body.getClass());
    ObjectMapper mapper = new ObjectMapper();
    JsonNode outJson = mapper.readTree((String)body);
    assertEquals("BobVila", outJson.get("FirstName").asText());
    assertNull(outJson.get("LastName"));
    assertEquals("bobvila1982", outJson.get("Title").asText());
    assertEquals("Let's build a house!", outJson.get("Description").asText());
}

public ProducerTemplate getProducerTemplate() {
    ProducerTemplate result = producerTemplate;

    // https://en.wikipedia.org/wiki/Double-checked_locking#Usage_in_Java
    if (result == null) {
        synchronized (this) {
            result = producerTemplate;
            if (result == null) {
                result = camelContext.getRegistry().lookupByNameAndType(PRODUCER_TEMPLATE, ProducerTemplate.class);
            }

            if (result == null) {
                // Create a new ProducerTemplate when there is none in the Camel registry.
                result = camelContext.createProducerTemplate();
                producerTemplateCreatedManually = true;
            }

            producerTemplate = result;
        }
    }

    return result;
}

@Test
public void testCamelProducer() throws Exception {
    // Starting Spring context.
    try (GenericApplicationContext context = new AnnotationConfigApplicationContext(ExampleConfiguration.class)) {
        context.start();

        // Sending Camel message.
        CamelContext camel = context.getBean(CamelContext.class);
        ProducerTemplate producerTemplate = camel.createProducerTemplate();
        producerTemplate.sendBody("direct:start", "Send me to the Sponge");

        // Waiting for the engine to process an event.
        Engine engine = context.getBean(Engine.class);
        await().atMost(60, TimeUnit.SECONDS)
                .until(() -> engine.getOperations().getVariable(AtomicBoolean.class, "sentCamelMessage").get());

        assertFalse(engine.isError());
        context.stop();
    }
}

public void xxxTestForkAndJoin() throws InterruptedException {
    MockEndpoint mock = getMockEndpoint("mock:result");
    mock.expectedMessageCount(NUMBER_OF_MESSAGES);

    ProducerTemplate template = context.createProducerTemplate();
    for (int i = 0; i < NUMBER_OF_MESSAGES; i++) {
        template.sendBodyAndHeader("seda:fork", "Test Message: " + i,
                "seqnum", new Long(i));
    }

    long expectedTime = NUMBER_OF_MESSAGES
            * (RandomSleepProcessor.MAX_PROCESS_TIME + RandomSleepProcessor.MIN_PROCESS_TIME)
            / 2 / CONCURRENCY + TIMEOUT;
    Thread.sleep(expectedTime);

    assertMockEndpointsSatisfied();
}

@Test
public void testMarshalViaDozer() throws Exception {

    CamelContext context = new DefaultCamelContext();
    context.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("direct:start").convertBodyTo(HashMap.class);
        }
    });

    DozerBeanMapperConfiguration mconfig = new DozerBeanMapperConfiguration();
    mconfig.setMappingFiles(Arrays.asList("bean-to-map-dozer-mappings.xml"));
    new DozerTypeConverterLoader(context, mconfig);

    context.start();
    try {
        ProducerTemplate producer = context.createProducerTemplate();
        Map<?, ?> result = producer.requestBody("direct:start", new Customer("John", "Doe", null), Map.class);
        Assert.assertEquals("John", result.get("firstName"));
        Assert.assertEquals("Doe", result.get("lastName"));
    } finally {
        context.stop();
    }
}

@Test
public void testReplyWithCustomType() throws Exception {
  Endpoint endpoint = camel.getEndpoint("direct:stuff");

  vertx.eventBus().registerDefaultCodec(Person.class, new PersonCodec());

  bridge = CamelBridge.create(vertx, new CamelBridgeOptions(camel)
      .addInboundMapping(new InboundMapping().setAddress("test-reply").setEndpoint(endpoint)));

  vertx.eventBus().consumer("test-reply", message -> {
    message.reply(new Person().setName("alice"));
  });

  camel.start();
  BridgeHelper.startBlocking(bridge);

  ProducerTemplate template = camel.createProducerTemplate();
  Future<Object> future = template.asyncRequestBody(endpoint, new Person().setName("bob"));
  Person response = template.extractFutureBody(future, Person.class);
  assertThat(response.getName()).isEqualTo("alice");
}

public void testRequestUsingDefaultEndpoint() throws Exception {
    ProducerTemplate producer = new DefaultProducerTemplate(context, context.getEndpoint("direct:out"));
    producer.start();

    Object out = producer.requestBody("Hello");
    assertEquals("Bye Bye World", out);

    out = producer.requestBodyAndHeader("Hello", "foo", 123);
    assertEquals("Bye Bye World", out);

    Map<String, Object> headers = new HashMap<String, Object>();
    out = producer.requestBodyAndHeaders("Hello", headers);
    assertEquals("Bye Bye World", out);

    out = producer.requestBodyAndHeaders("Hello", null);
    assertEquals("Bye Bye World", out);

    producer.stop();
}

@Override
protected void doPost(final HttpServletRequest request, final HttpServletResponse response) throws ServletException, IOException {
    /**
     * Get message and name parameters sent on the POST request
     */
    String message = request.getParameter("message");
    String name = request.getParameter("name");

    /**
     * Create a ProducerTemplate to invoke the direct:start endpoint, which will
     * result in the greeting web service 'greet' method being invoked.
     *
     * The web service parameters are sent to camel as an object array which is
     * set as the request message body.
     *
     * The web service result string is returned back for display on the UI.
     */
    ProducerTemplate producer = camelContext.createProducerTemplate();
    Object[] serviceParams = new Object[] { message, name };
    String result = producer.requestBody("direct:start", serviceParams, String.class);

    request.setAttribute("greeting", result);
    request.getRequestDispatcher("/greeting.jsp").forward(request, response);
}

public void testCacheProducersFromContext() throws Exception {
    ProducerTemplate template = context.createProducerTemplate(500);

    assertEquals("Size should be 0", 0, template.getCurrentCacheSize());

    // test that we cache at most 500 producers to avoid it eating to much memory
    for (int i = 0; i < 503; i++) {
        Endpoint e = context.getEndpoint("seda:queue:" + i);
        template.sendBody(e, "Hello");
    }

    // the eviction is async so force cleanup
    template.cleanUp();

    assertEquals("Size should be 500", 500, template.getCurrentCacheSize());
    template.stop();

    // should be 0
    assertEquals("Size should be 0", 0, template.getCurrentCacheSize());
}

@Override
protected void doPost(final HttpServletRequest request, final HttpServletResponse response) throws ServletException, IOException {
    /**
     * Get message and name parameters sent on the POST request
     */
    String message = request.getParameter("message");
    String name = request.getParameter("name");

    /**
     * Create a ProducerTemplate to invoke the direct:start endpoint, which will
     * result in the greeting web service 'greet' method being invoked.
     *
     * The web service parameters are sent to camel as an object array which is
     * set as the request message body.
     *
     * The web service result string is returned back for display on the UI.
     */
    ProducerTemplate producer = camelContext.createProducerTemplate();
    Object[] serviceParams = new Object[] { message, name };
    String result = producer.requestBody("direct:start", serviceParams, String.class);

    request.setAttribute("greeting", result);
    request.getRequestDispatcher("/greeting.jsp").forward(request, response);
}

@Test
public void testJettyMulticastJmsFile() throws Exception {
    TestSupport.deleteDirectory("target/jetty");

    ProducerTemplate template = camelContext.createProducerTemplate();

    String out = template.requestBody(URL, "Hello World", String.class);
    assertEquals("Bye World", out);

    template.stop();

    ConsumerTemplate consumer = camelContext.createConsumerTemplate();
    String in = consumer.receiveBody("jms:queue:foo", 5000, String.class);
    assertEquals("Hello World", in);

    String in2 = consumer.receiveBody("file://target/jetty?noop=true&readLock=none", 5000, String.class);
    assertEquals("Hello World", in2);

    consumer.stop();
}

@Override
protected void doPost(final HttpServletRequest request, final HttpServletResponse response) throws ServletException, IOException {
  Map<String, Object> headers = new HashMap<>();

  Enumeration<String> parameterNames = request.getParameterNames();
  while(parameterNames.hasMoreElements()) {
    String parameterName = parameterNames.nextElement();
    String parameterValue = request.getParameter(parameterName);
    headers.put(parameterName, parameterValue);
  }

  try {
    ProducerTemplate producer = camelContext.createProducerTemplate();
    producer.sendBodyAndHeaders("direct:sendmail", request.getParameter("message"), headers);
    request.getRequestDispatcher("/success.jsp").forward(request, response);
  } catch (CamelExecutionException e) {
    request.setAttribute("error", e);
    request.getRequestDispatcher("/error.jsp").forward(request, response);
  }
}

@Override
protected void doPost(final HttpServletRequest request, final HttpServletResponse response) throws ServletException, IOException {
    /**
     * Get message and name parameters sent on the POST request
     */
    String message = request.getParameter("message");
    String name = request.getParameter("name");

    /**
     * Create a ProducerTemplate to invoke the direct:start endpoint, which will
     * result in the greeting web service 'greet' method being invoked.
     *
     * The web service parameters are sent to camel as an object array which is
     * set as the request message body.
     *
     * The web service result string is returned back for display on the UI.
     */
    ProducerTemplate producer = camelContext.createProducerTemplate();
    Object[] serviceParams = new Object[] {message, name};
    String result = producer.requestBody("direct:start", serviceParams, String.class);

    request.setAttribute("greeting", result);
    request.getServletContext().getRequestDispatcher("/greeting.jsp").forward(request, response);
}

@Test
public void testStatelessSessionBean() throws Exception {

    CamelContext camelctx = new DefaultCamelContext();
    camelctx.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("direct:start").to("ejb:java:module/HelloBean");
        }
    });

    camelctx.start();
    try {
        ProducerTemplate producer = camelctx.createProducerTemplate();
        String result = producer.requestBody("direct:start", "Kermit", String.class);
        Assert.assertEquals("Hello Kermit", result);
    } finally {
        camelctx.stop();
    }
}

public void testStandalone() throws Exception {
    CamelContext context = new DefaultCamelContext();
    context.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("direct:start").to("mock:result");
        }
    });
    context.start();

    MockEndpoint mock = context.getEndpoint("mock:result", MockEndpoint.class);
    mock.expectedMessageCount(1);

    ProducerTemplate template = context.createProducerTemplate();
    template.sendBody("direct:start", "Hello World");

    mock.assertIsSatisfied();

    template.stop();
    context.stop();
}

public void testAutoStartupFalse() throws Exception {
    ac = new ClassPathXmlApplicationContext("org/apache/camel/spring/config/RouteAutoStartupFalseTest.xml");

    SpringCamelContext camel = ac.getBeansOfType(SpringCamelContext.class).values().iterator().next();

    assertEquals(false, camel.getRouteStatus("foo").isStarted());

    // now starting route manually
    camel.startRoute("foo");
    assertEquals(true, camel.getRouteStatus("foo").isStarted());

    // and now we can send a message to the route and see that it works
    MockEndpoint mock = camel.getEndpoint("mock:result", MockEndpoint.class);
    mock.expectedMessageCount(1);

    ProducerTemplate template = camel.createProducerTemplate();
    template.start();
    template.sendBody("direct:start", "Hello World");
    template.stop();

    mock.assertIsSatisfied();
}

public User getUser(String username, ProducerTemplate producerTemplate) throws Exception
    {
        Map<String, String> cond = new HashMap<>(2);
//        cond.put("status", "active");
        cond.put("username", username);
        DBObject result = findOneByCondition(producerTemplate, cond);
        if (result == null)
            throw new IllegalArgumentException("User " + username + " has not been found");

        result.removeField("_id");
        User user = subscriberFactory.convertJson2Pojo(User.class, subscriberFactory.convertPojo2Json(result));
        BasicDBList subscribers = (BasicDBList) result.get("subscribers");
        logger.info("GET: User: {} with status {} \n {} {}", user.getUsername(), user.getStatus(), subscribers.getClass().toString(), subscribers);
        return user;

    }

public OperationResult addUser(User user, ProducerTemplate producerTemplate) throws Exception
{
    Map<String, String> cond= new HashMap<>(1);
    cond.put("username", user.getUsername());
    DBObject r=findOneByCondition(producerTemplate,cond);
    if (r != null)
        throw new IllegalArgumentException("Creation is impossible. User " + user.getUsername() + " already exists");

    Object result = producerTemplate.requestBody(
            getQuery(getDefaultMongoDatabase(), getDefaulMongoCollection(), MongoDbOperation.insert),
            subscriberFactory.convertPojo2Json(user));

    logger.info("INSERT: New user: {} has been inserted into Mongo with the result: {}", user.getUsername(), result);
    logger.debug("INSERT: result: {}", result);
    return OperationResult.SUCCESS;
}

public void testTwo() throws Exception {
    int start = myInterceptor.getCount();

    MockEndpoint result = camel2.getEndpoint("mock:result", MockEndpoint.class);
    result.expectedBodiesReceived("Bye World");

    ProducerTemplate template = camel2.createProducerTemplate();
    template.start();
    template.sendBody("direct:two", "Bye World");
    template.stop();

    result.assertIsSatisfied();

    // lets see if the counter is +2 since last (has 2 steps in the route)
    int delta = myInterceptor.getCount() - start;
    assertEquals("Should have been counted +2", 2, delta);
}

public void testXMLRouteLoading() throws Exception {
    applicationContext = createApplicationContext();

    SpringCamelContext context = applicationContext.getBean("camel-A", SpringCamelContext.class);
    assertValidContext(context);

    MockEndpoint resultEndpoint = (MockEndpoint) resolveMandatoryEndpoint(context, "mock:result");
    resultEndpoint.expectedBodiesReceived(body);

    // now lets send a message
    ProducerTemplate template = context.createProducerTemplate();
    template.start();
    template.send("seda:start", new Processor() {
        public void process(Exchange exchange) {
            Message in = exchange.getIn();
            in.setHeader("name", "James");
            in.setBody(body);
        }
    });
    template.stop();

    resultEndpoint.assertIsSatisfied();
}

public Exchange doSignatureRouteTest(RouteBuilder builder, Exchange e, Map<String, Object> headers) throws Exception {
    CamelContext context = new DefaultCamelContext();
    try {
        context.addRoutes(builder);
        context.start();

        MockEndpoint mock = context.getEndpoint("mock:result", MockEndpoint.class);
        mock.setExpectedMessageCount(1);

        ProducerTemplate template = context.createProducerTemplate();
        if (e != null) {
            template.send("direct:in", e);
        } else {
            template.sendBodyAndHeaders("direct:in", payload, headers);
        }
        assertMockEndpointsSatisfied();
        return mock.getReceivedExchanges().get(0);
    } finally {
        context.stop();
    }
}

public void testXMLRouteLoading() throws Exception {
    applicationContext = createApplicationContext();

    SpringCamelContext context = applicationContext.getBeansOfType(SpringCamelContext.class).values().iterator().next();
    assertValidContext(context);

    // now lets send a message
    ProducerTemplate template = context.createProducerTemplate();
    template.start();
    template.send("direct:start", new Processor() {
        public void process(Exchange exchange) {
            Message in = exchange.getIn();
            in.setHeader("name", "James");
            in.setBody(body);
        }
    });
    template.stop();

    MyProcessor myProcessor = applicationContext.getBean("myProcessor", MyProcessor.class);
    List<Exchange> list = myProcessor.getExchanges();
    assertEquals("Should have received a single exchange: " + list, 1, list.size());
}

@Test
public void testJaxrsProducer() throws Exception {
    CamelContext camelctx = new DefaultCamelContext();
    camelctx.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("direct:start")
            .setHeader(Exchange.HTTP_METHOD, constant("GET")).
            to("cxfrs://" + getEndpointAddress("/") + "?resourceClasses=" + GreetingService.class.getName());
        }
    });

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

private void sendExchangesThroughDroppingThrottler(List<Exchange> sentExchanges, int messages) throws Exception {
    ProducerTemplate myTemplate = context.createProducerTemplate();

    DirectEndpoint targetEndpoint = resolveMandatoryEndpoint("direct:sample", DirectEndpoint.class);
    for (int i = 0; i < messages; i++) {
        Exchange e = targetEndpoint.createExchange();
        e.getIn().setBody("<message>" + i + "</message>");
        // only send if we are still started
        if (context.getStatus().isStarted()) {
            myTemplate.send(targetEndpoint, e);
            sentExchanges.add(e);
            Thread.sleep(100);
        }
    }
    myTemplate.stop();
}

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

    tpl.sendBodyAndProperty("direct:start", Collections.singletonMap("var1", "ala"), FlowableProducer.PROCESS_KEY_PROPERTY, "key1");

    String instanceId = runtimeService.createProcessInstanceQuery().processInstanceBusinessKey("key1").singleResult().getProcessInstanceId();
    tpl.sendBodyAndProperty("direct:receive", null, FlowableProducer.PROCESS_KEY_PROPERTY, "key1");

    assertProcessEnded(instanceId);

    me.assertIsSatisfied();
}

@Test
public void testCamelContext() throws Exception {
    CamelContext context = getCamelContext();
    assertNotNull(context);

    assertEquals("MyCamel", context.getName());

    ProducerTemplate template = context.createProducerTemplate();

    MockEndpoint mock = context.getEndpoint("mock:foo", MockEndpoint.class);
    mock.expectedMessageCount(1);

    template.sendBody("seda:foo", "Hello World");

    mock.assertIsSatisfied();
    template.stop();
}

@Deployment
public void testCamelPropertiesAll() throws Exception {
    ProducerTemplate tpl = camelContext.createProducerTemplate();
    Exchange exchange = camelContext.getEndpoint("direct:startAllProperties").createExchange();
    tpl.send("direct:startAllProperties", 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());
    assertEquals("sampleValueForProperty1", variables.get("property1"));
    assertEquals("sampleValueForProperty2", variables.get("property2"));
    assertEquals("sampleValueForProperty3", variables.get("property3"));
}

@Deployment(resources = { "org/flowable/camel/variables/CamelVariableTransferTest.testCamelPropertiesAll.bpmn20.xml" })
public void testCamelPropertiesAndBody() throws Exception {
    ProducerTemplate tpl = camelContext.createProducerTemplate();
    Exchange exchange = camelContext.getEndpoint("direct:startAllProperties").createExchange();

    tpl.send("direct:startAllProperties", 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());
    assertEquals("sampleValueForProperty1", variables.get("property1"));
    assertEquals("sampleValueForProperty2", variables.get("property2"));
    assertEquals("sampleValueForProperty3", variables.get("property3"));
    assertEquals("sampleBody", variables.get("camelBody"));
}

@Deployment(resources = { "org/flowable/camel/variables/CamelVariableTransferTest.testCamelPropertiesAll.bpmn20.xml" })
public void testCamelPropertiesFiltered() throws Exception {
    ProducerTemplate tpl = camelContext.createProducerTemplate();
    Exchange exchange = camelContext.getEndpoint("direct:startFilteredProperties").createExchange();
    tpl.send("direct:startFilteredProperties", 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());
    assertEquals("sampleValueForProperty1", variables.get("property1"));
    assertEquals("sampleValueForProperty2", variables.get("property2"));
    assertNull(variables.get("property3"));
}

@Deployment(resources = { "org/flowable/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"));
}

@Deployment(resources = { "org/flowable/camel/variables/CamelVariableTransferTest.testCamelPropertiesAll.bpmn20.xml" })
public void testCamelHeadersAll() throws Exception {
    ProducerTemplate tpl = camelContext.createProducerTemplate();
    Exchange exchange = camelContext.getEndpoint("direct:startAllProperties").createExchange();
    tpl.send("direct:startAllProperties", 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());
    assertEquals("sampleValueForProperty1", variables.get("property1"));
    assertEquals("sampleValueForProperty2", variables.get("property2"));
    assertEquals("sampleValueForProperty3", variables.get("property3"));
}

@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, FlowableProducer.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"));
}

@Deployment
public void testInitiatorCamelCall() throws Exception {
    CamelContext ctx = applicationContext.getBean(CamelContext.class);
    ProducerTemplate tpl = ctx.createProducerTemplate();
    String body = "body text";

    Exchange exchange = ctx.getEndpoint("direct:startWithInitiatorHeader").createExchange();
    exchange.getIn().setBody(body);
    tpl.send("direct:startWithInitiatorHeader", exchange);

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

    String initiator = (String) runtimeService.getVariable(instanceId, "initiator");
    assertEquals("kermit", initiator);

    Object camelInitiatorHeader = runtimeService.getVariable(instanceId, "CamelProcessInitiatorHeader");
    assertNull(camelInitiatorHeader);
}

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

    tpl.sendBodyAndProperty("direct:start", Collections.singletonMap("var1", "ala"), FlowableProducer.PROCESS_KEY_PROPERTY, "key1");

    String instanceId = runtimeService.createProcessInstanceQuery().processInstanceBusinessKey("key1").singleResult().getProcessInstanceId();
    tpl.sendBodyAndProperty("direct:receive", null, FlowableProducer.PROCESS_KEY_PROPERTY, "key1");

    assertProcessEnded(instanceId);

    me.assertIsSatisfied();
}

@Deployment(resources = { "process/empty.bpmn20.xml" })
public void testRunProcessWithHeader() throws Exception {
    CamelContext ctx = applicationContext.getBean(CamelContext.class);
    ProducerTemplate tpl = camelContext.createProducerTemplate();
    String body = "body text";
    Exchange exchange = ctx.getEndpoint("direct:startEmptyWithHeader").createExchange();
    exchange.getIn().setBody(body);
    tpl.send("direct:startEmptyWithHeader", exchange);

    String instanceId = (String) exchange.getProperty("PROCESS_ID_PROPERTY");
    assertProcessEnded(instanceId);
    HistoricVariableInstance var = processEngine.getHistoryService().createHistoricVariableInstanceQuery().variableName("camelBody").singleResult();
    assertNotNull(var);
    assertEquals(body, var.getValue());
    var = processEngine.getHistoryService().createHistoricVariableInstanceQuery().variableName("MyVar").singleResult();
    assertNotNull(var);
    assertEquals("Foo", var.getValue());
}

@Test
@Ignore("This issue will be revisit when we have chance to rewrite bindy parser")
public void testBindySeparatorsAround() throws Exception {
    CamelContext ctx = new DefaultCamelContext();
    ctx.addRoutes(createRoute()); // new ReconciliationRoute()
    ctx.start();

    // TODO The separator in the beginning of the quoted field is still not handled.
    // We may need to convert the separators in the quote into some kind of safe code 
    String addressLine1 = ",8506 SIX FORKS ROAD,";

    MockEndpoint mock = ctx.getEndpoint("mock:result", MockEndpoint.class);
    mock.expectedPropertyReceived("addressLine1", addressLine1);

    String csvLine = "\"PROBLEM SOLVER\",\"" + addressLine1
                     + "\",\"SUITE 104\",\"RALEIGH\",\"NC\",\"27615\",\"US\"";
    ProducerTemplate template = ctx.createProducerTemplate();
    template.sendBody("direct:fromCsv", csvLine.trim());

    mock.assertIsSatisfied();

}