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

@Test
public void testSimpleTransform() throws Exception {

    CamelContext camelctx = new DefaultCamelContext();
    camelctx.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("file://{{jboss.server.data.dir}}/" + RouteBuilderA.class.getName() + "?fileName=fileA&doneFileName=fileA.done")
            .convertBodyTo(String.class)
            .to("seda:end");
        }
    });

    camelctx.start();
    try {
        ConsumerTemplate consumer = camelctx.createConsumerTemplate();
        String result = consumer.receiveBody("seda:end", String.class);
        Assert.assertEquals("Hello 1", result);
    } finally {
        camelctx.stop();
    }
}

@Test
public void testSimpleTransform() throws Exception {

    CamelContext camelctx = new DefaultCamelContext();
    camelctx.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("file://{{jboss.server.data.dir}}/" + RouteBuilderA.class.getName() + "?fileName=fileA&doneFileName=fileA.done")
            .convertBodyTo(String.class)
            .to("seda:end");
        }
    });

    camelctx.start();
    try {
        ConsumerTemplate consumer = camelctx.createConsumerTemplate();
        String result = consumer.receiveBody("seda:end", String.class);
        Assert.assertEquals("Hello 1", result);
    } finally {
        camelctx.stop();
    }
}

@Test
public void testSimpleTransform() throws Exception {

    CamelContext camelctx = new DefaultCamelContext();
    camelctx.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("file://{{jboss.server.data.dir}}/" + RouteBuilderA.class.getName() + "?fileName=fileA&doneFileName=fileA.done")
            .convertBodyTo(String.class)
            .to("seda:end");
        }
    });

    camelctx.start();
    try {
        ConsumerTemplate consumer = camelctx.createConsumerTemplate();
        String result = consumer.receiveBody("seda:end", String.class);
        Assert.assertEquals("Hello 1", result);
    } finally {
        camelctx.stop();
    }
}

public void testCacheConsumers() throws Exception {
    ConsumerTemplate template = context.createConsumerTemplate();

    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 < 203; i++) {
        Endpoint e = context.getEndpoint("direct:queue:" + i);
        template.receiveNoWait(e);
    }

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

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

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

public void testCacheConsumers() throws Exception {
    ConsumerTemplate template = new DefaultConsumerTemplate(context);
    template.setMaximumCacheSize(500);
    template.start();

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

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

    // 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());
}

public void testCacheConsumersFromContext() throws Exception {
    ConsumerTemplate template = context.createConsumerTemplate(500);

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

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

    // 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());
}

@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();
}

private static void doStopTemplates(ConsumerTemplate consumer,
                                    ProducerTemplate template) throws Exception {
    if (consumer != null) {
        if (consumer == threadConsumer.get()) {
            threadConsumer.remove();
        }
        consumer.stop();
        consumer = null;
    }
    if (template != null) {
        if (template == threadTemplate.get()) {
            threadTemplate.remove();
        }
        template.stop();
        template = null;
    }
}

@Test
public void testTemplateMaximumCache() throws Exception {
    assertNotNull("Should have injected a consumer template", template);

    ConsumerTemplate lookup = context.getRegistry().lookupByNameAndType("template", ConsumerTemplate.class);
    assertNotNull("Should lookup consumer template", lookup);

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

    // test that we cache at most 50 producers to avoid it eating to much memory
    for (int i = 0; i < 53; i++) {
        Endpoint e = context.getEndpoint("direct:queue:" + i);
        template.receiveNoWait(e);
    }

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

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

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

@Inject
public CamelFinalizer(final GuiceInjector injector,
    final DefaultCamelContext ctx,
    final @Named("camel.routes") Set<Object> routes,
    final RouteBuilder rb,
    final ProducerTemplate producer,
    final ConsumerTemplate consumer) throws Exception {
  this.ctx = ctx;
  this.producer = producer;
  this.consumer = consumer;

  this.ctx.setInjector(injector);

  for (Object route : routes) {
    if (route instanceof RoutesBuilder) {
      this.ctx.addRoutes((RoutesBuilder) route);
    }
  }
  this.ctx.addRoutes(rb);
}

@Test
public void testMonitorMBeanAttribute() throws Exception {
    CamelContext context = contextRegistry.getCamelContext("jmx-context-1");
    Assert.assertNotNull("Camel context jmx-context-1 was null", context);
    final String routeName = context.getRoutes().get(0).getId();

    CamelContext camelctx = new DefaultCamelContext();
    camelctx.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("jmx:platform?format=raw&objectDomain=org.apache.camel&key.context=jmx-context-1&key.type=routes&key.name=\"" + routeName + "\"" +
            "&monitorType=counter&observedAttribute=ExchangesTotal&granularityPeriod=500").
            to("direct:end");
        }
    });

    camelctx.start();
    try {
        ConsumerTemplate consumer = camelctx.createConsumerTemplate();
        MonitorNotification notifcation = consumer.receiveBody("direct:end", MonitorNotification.class);
        Assert.assertEquals("ExchangesTotal", notifcation.getObservedAttribute());
    } finally {
        camelctx.stop();
    }
}

@Test
public void testGet() throws Exception {
    CamelContext camelctx = createCamelContext();
    camelctx.start();
    try {
        ProducerTemplate template = camelctx.createProducerTemplate();
        Mockito.when(map.get("4711")).thenReturn("my-foo");
        template.sendBodyAndHeader("direct:get", null, HazelcastConstants.OBJECT_ID, "4711");

        ConsumerTemplate consumer = camelctx.createConsumerTemplate();
        String body = consumer.receiveBody("seda:out", 5000, String.class);
        Mockito.verify(map).get("4711");
        Assert.assertEquals("my-foo", body);
    } finally {
        camelctx.stop();
    }
}

@Test
public void testGetAllEmptySet() throws Exception {
    CamelContext camelctx = createCamelContext();
    camelctx.start();
    try {
        Set<Object> l = new HashSet<Object>();
        Map t = new HashMap();
        t.put("key1", "value1");
        t.put("key2", "value2");
        t.put("key3", "value3");
        Mockito.when(map.getAll(Mockito.anySet())).thenReturn(t);

        ProducerTemplate template = camelctx.createProducerTemplate();
        template.sendBodyAndHeader("direct:getAll", null, HazelcastConstants.OBJECT_ID, l);

        ConsumerTemplate consumer = camelctx.createConsumerTemplate();
        String body = consumer.receiveBody("seda:out", 5000, String.class);
        Mockito.verify(map).getAll(l);
        Assert.assertTrue(body.contains("key1=value1"));
        Assert.assertTrue(body.contains("key2=value2"));
        Assert.assertTrue(body.contains("key3=value3"));
    } finally {
        camelctx.stop();
    }
}

@Test
public void testGetAllOnlyOneKey() throws Exception {
    CamelContext camelctx = createCamelContext();
    camelctx.start();
    try {
        Set<Object> l = new HashSet<Object>();
        l.add("key1");
        Map t = new HashMap();
        t.put("key1", "value1");
        Mockito.when(map.getAll(l)).thenReturn(t);

        ProducerTemplate template = camelctx.createProducerTemplate();
        template.sendBodyAndHeader("direct:getAll", null, HazelcastConstants.OBJECT_ID, l);

        ConsumerTemplate consumer = camelctx.createConsumerTemplate();
        String body = consumer.receiveBody("seda:out", 5000, String.class);
        Mockito.verify(map).getAll(l);
        Assert.assertEquals("{key1=value1}", body);
    } finally {
        camelctx.stop();
    }
}

@Test
public void testQuery() throws Exception {
    CamelContext camelctx = createCamelContext();
    camelctx.start();
    try {
        String sql = "bar > 1000";
        Mockito.when(map.values(Mockito.any(SqlPredicate.class))).thenReturn(Arrays.<Object>asList(new Dummy("beta", 2000), new Dummy("gamma", 3000)));

        ProducerTemplate template = camelctx.createProducerTemplate();
        template.sendBodyAndHeader("direct:queue", null, HazelcastConstants.QUERY, sql);
        Mockito.verify(map).values(Mockito.any(SqlPredicate.class));

        ConsumerTemplate consumer = camelctx.createConsumerTemplate();
        Collection<?> b1 = consumer.receiveBody("seda:out", 5000, Collection.class);

        Assert.assertNotNull(b1);
        Assert.assertEquals(2, b1.size());
    } finally {
        camelctx.stop();
    }
}

@Test
public void testEmptyQuery() throws Exception {
    CamelContext camelctx = createCamelContext();
    camelctx.start();
    try {
        Mockito.when(map.values()).thenReturn(Arrays.<Object>asList(new Dummy("beta", 2000), new Dummy("gamma", 3000), new Dummy("delta", 4000)));

        ProducerTemplate template = camelctx.createProducerTemplate();
        template.sendBody("direct:queue", null);
        Mockito. verify(map).values();

        ConsumerTemplate consumer = camelctx.createConsumerTemplate();
        Collection<?> b1 = consumer.receiveBody("seda:out", 5000, Collection.class);

        Assert.assertNotNull(b1);
        Assert.assertEquals(3, b1.size());
    } finally {
        camelctx.stop();
    }
}

@Test
public void testContainsKey() throws Exception {
    CamelContext camelctx = createCamelContext();
    camelctx.start();
    try {
        Mockito.when(map.containsKey("testOk")).thenReturn(true);
        Mockito.when(map.containsKey("testKo")).thenReturn(false);

        ProducerTemplate template = camelctx.createProducerTemplate();
        template.sendBodyAndHeader("direct:containsKey", null, HazelcastConstants.OBJECT_ID, "testOk");

        ConsumerTemplate consumer = camelctx.createConsumerTemplate();
        Boolean body = consumer.receiveBody("seda:out", 5000, Boolean.class);
        Mockito.verify(map).containsKey("testOk");
        Assert.assertEquals(true, body);
        template.sendBodyAndHeader("direct:containsKey", null, HazelcastConstants.OBJECT_ID, "testKo");
        body = consumer.receiveBody("seda:out", 5000, Boolean.class);
        Mockito.verify(map).containsKey("testKo");
        Assert.assertEquals(false, body);
    } finally {
        camelctx.stop();
    }
}

@Test
public void testContainsValue() throws Exception {
    CamelContext camelctx = createCamelContext();
    camelctx.start();
    try {
        Mockito.when(map.containsValue("testOk")).thenReturn(true);
        Mockito.when(map.containsValue("testKo")).thenReturn(false);

        ProducerTemplate template = camelctx.createProducerTemplate();
        template.sendBody("direct:containsValue", "testOk");

        ConsumerTemplate consumer = camelctx.createConsumerTemplate();
        Boolean body = consumer.receiveBody("seda:out", 5000, Boolean.class);
        Mockito.verify(map).containsValue("testOk");
        Assert.assertEquals(true, body);
        template.sendBody("direct:containsValue", "testKo");
        body = consumer.receiveBody("seda:out", 5000, Boolean.class);
        Mockito.verify(map).containsValue("testKo");
        Assert.assertEquals(false, body);
    } finally {
        camelctx.stop();
    }
}

@Test
public void testConsumeAll() throws Exception {

    CamelContext camelctx = new DefaultCamelContext();
    camelctx.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("cql://localhost/camel_ks?cql=" + CQL).to("seda:end");
        }
    });

    camelctx.start();
    try {
        ConsumerTemplate consumer = camelctx.createConsumerTemplate();
        List<?> result = consumer.receiveBody("seda:end", 3000, List.class);
        Assert.assertNotNull("Result not null", result);
        Assert.assertEquals("Two records selected", 2, result.size());
    } finally {
        camelctx.stop();
    }
}

public static Object receive(final String locator, final String endpoint, final long timeout, final Class<?> expected) {
    final BeanLocator.LocatorInstance<CamelTemplateLocator> locatorInstance = locator(locator);
    try {
        final ConsumerTemplate consumerTemplate = locatorInstance.getValue().findConsumerTemplate();
        if (timeout > 0) {
            if (expected != null) {
                return consumerTemplate.receiveBody(endpoint, expected);
            }
            return consumerTemplate.receiveBody(endpoint);
        }

        if (expected != null) {
            return consumerTemplate.receiveBody(endpoint, timeout, expected);
        }
        return consumerTemplate.receiveBody(endpoint, timeout);
    } finally {
        locatorInstance.release();
    }
}

@Test
public void testMonitorMBeanAttribute() throws Exception {
    Context context = new InitialContext();
    CamelContextRegistry contextRegistry = (CamelContextRegistry) context.lookup("java:jboss/camel/CamelContextRegistry");

    CamelContext sysctx = contextRegistry.getCamelContext("camel-1");
    Assert.assertEquals(ServiceStatus.Started, sysctx.getStatus());
    final String routeName = sysctx.getRoutes().get(0).getId();

    CamelContext camelctx = new DefaultCamelContext();
    camelctx.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("jmx:platform?format=raw&objectDomain=org.apache.camel&key.context=camel-1&key.type=routes&key.name=\"" + routeName + "\"" +
            "&monitorType=counter&observedAttribute=ExchangesTotal&granularityPeriod=500").
            to("direct:end");
        }
    });

    camelctx.start();
    try {
        ConsumerTemplate consumer = camelctx.createConsumerTemplate();
        MonitorNotification notifcation = consumer.receiveBody("direct:end", MonitorNotification.class);
        Assert.assertEquals("ExchangesTotal", notifcation.getObservedAttribute());
    } finally {
        camelctx.stop();
    }
}

public RssPollingTask(ConsumerTemplate consumer, String rssURI, String path, String fileName)
{
    this.consumer = consumer;
    this.rssURI = rssURI;
    this.path = path;
    this.fileName = fileName;
}

@Test
public void testMonitorMBeanAttribute() throws Exception {
    Context context = new InitialContext();
    CamelContextRegistry contextRegistry = (CamelContextRegistry) context.lookup("java:jboss/camel/CamelContextRegistry");

    CamelContext sysctx = contextRegistry.getCamelContext("camel-1");
    Assert.assertEquals(ServiceStatus.Started, sysctx.getStatus());
    final String routeName = sysctx.getRoutes().get(0).getId();

    CamelContext camelctx = new DefaultCamelContext();
    camelctx.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("jmx:platform?format=raw&objectDomain=org.apache.camel&key.context=camel-1&key.type=routes&key.name=\"" + routeName + "\"" +
                    "&monitorType=counter&observedAttribute=ExchangesTotal&granularityPeriod=500").
                    to("direct:end");
        }
    });

    camelctx.start();
    try {
        ConsumerTemplate consumer = camelctx.createConsumerTemplate();
        MonitorNotification notifcation = consumer.receiveBody("direct:end", MonitorNotification.class);
        Assert.assertEquals("ExchangesTotal", notifcation.getObservedAttribute());
    } finally {
        camelctx.stop();
    }
}

public ConsumerTemplate createConsumerTemplate(int maximumCacheSize) {
    DefaultConsumerTemplate answer = new DefaultConsumerTemplate(this);
    answer.setMaximumCacheSize(maximumCacheSize);
    // start it so its ready to use
    try {
        startService(answer);
    } catch (Exception e) {
        throw ObjectHelper.wrapRuntimeCamelException(e);
    }
    return answer;
}

/**
 * Creates the object to be injected for an {@link org.apache.camel.EndpointInject} or {@link org.apache.camel.Produce} injection point
 */
public Object getInjectionValue(Class<?> type, String endpointUri, String endpointRef, String endpointProperty,
                                String injectionPointName, Object bean, String beanName) {
    if (type.isAssignableFrom(ProducerTemplate.class)) {
        return createInjectionProducerTemplate(endpointUri, endpointRef, endpointProperty, injectionPointName, bean);
    } else if (type.isAssignableFrom(ConsumerTemplate.class)) {
        return createInjectionConsumerTemplate(endpointUri, endpointRef, endpointProperty, injectionPointName);
    } else {
        Endpoint endpoint = getEndpointInjection(bean, endpointUri, endpointRef, endpointProperty, injectionPointName, true);
        if (endpoint != null) {
            if (type.isInstance(endpoint)) {
                return endpoint;
            } else if (type.isAssignableFrom(Producer.class)) {
                return createInjectionProducer(endpoint, bean, beanName);
            } else if (type.isAssignableFrom(PollingConsumer.class)) {
                return createInjectionPollingConsumer(endpoint, bean, beanName);
            } else if (type.isInterface()) {
                // lets create a proxy
                try {
                    return ProxyHelper.createProxy(endpoint, type);
                } catch (Exception e) {
                    throw createProxyInstantiationRuntimeException(type, endpoint, e);
                }
            } else {
                throw new IllegalArgumentException("Invalid type: " + type.getName()
                        + " which cannot be injected via @EndpointInject/@Produce for: " + endpoint);
            }
        }
        return null;
    }
}

/**
 * Factory method to create a {@link org.apache.camel.ConsumerTemplate} to be injected into a POJO
 */
protected ConsumerTemplate createInjectionConsumerTemplate(String endpointUri, String endpointRef, String endpointProperty,
                                                           String injectionPointName) {
    ConsumerTemplate answer = new DefaultConsumerTemplate(getCamelContext());
    // start the template so its ready to use
    try {
        startService(answer, null, null, null);
    } catch (Exception e) {
        throw ObjectHelper.wrapRuntimeCamelException(e);
    }
    return answer;
}

@Test
public void shouldNotBreakRedeliveriesOfSecondContextAfterFirstBeingStopped() throws Exception {
    DefaultCamelContext context1 = createContext();
    ProducerTemplate producer1 = context1.createProducerTemplate();
    ConsumerTemplate consumer1 = context1.createConsumerTemplate();
    context1.start();
    producer1.sendBody("seda://input", "Hey1");
    Exchange ex1 = consumer1.receive("seda://output", 5000);

    DefaultCamelContext context2 = createContext();
    ProducerTemplate producer2 = context2.createProducerTemplate();
    ConsumerTemplate consumer2 = context2.createConsumerTemplate();
    context2.start();

    // now stop 1, and see that 2 is still working
    consumer1.stop();
    producer1.stop();
    context1.stop();

    producer2.sendBody("seda://input", "Hey2");
    Exchange ex2 = consumer2.receive("seda://output", 5000);

    Assert.assertNotNull(ex1);
    Assert.assertEquals("Hey1", ex1.getIn().getBody());
    Assert.assertNotNull(ex2);
    Assert.assertEquals("Hey2", ex2.getIn().getBody());

    consumer2.stop();
    producer2.stop();
    context2.stop();
}

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("file://target/enrich?move=.done")
                .process(new Processor() {
                    public void process(Exchange exchange) throws Exception {
                        String name = exchange.getIn().getHeader(Exchange.FILE_NAME_ONLY, String.class);
                        name = FileUtil.stripExt(name) + ".dat";

                        // use a consumer template to get the data file
                        Exchange data = null;
                        ConsumerTemplate con = exchange.getContext().createConsumerTemplate();
                        try {
                            // try to get the data file
                            data = con.receive("file://target/enrichdata?move=.done&fileName=" + name, 5000);
                        } finally {
                            // stop the consumer as it does not need to poll for files anymore
                            con.stop();
                        }

                        // if we found the data file then process it by sending it to the direct:data endpoint
                        if (data != null) {
                            template.send("direct:data", data);
                        } else {
                            // otherwise do a rollback
                            throw new CamelExchangeException("Cannot find the data file " + name, exchange);
                        }
                    }
                }).to("mock:start");

            from("direct:data")
                .to("mock:result");
        }
    };
}

private static void doStopTemplates(ConsumerTemplate consumer, ProducerTemplate template) throws Exception {
    if (consumer != null) {
        if (consumer == threadConsumer.get()) {
            threadConsumer.remove();
        }
        consumer.stop();
    }
    if (template != null) {
        if (template == threadTemplate.get()) {
            threadTemplate.remove();
        }
        template.stop();
    }
}

public ConsumerTemplate getObject() throws Exception {
    template = new DefaultConsumerTemplate(getCamelContext());

    // set custom cache size if provided
    if (maximumCacheSize != null) {
        template.setMaximumCacheSize(maximumCacheSize);
    }

    // must start it so its ready to use
    ServiceHelper.startService(template);
    return template;
}

/**
 * Default consumer template for the bootstrapped Camel context.
 */
@Bean(initMethod = "", destroyMethod = "")
// Camel handles the lifecycle of this bean
@ConditionalOnMissingBean(ConsumerTemplate.class)
ConsumerTemplate consumerTemplate(CamelContext camelContext,
                                  CamelConfigurationProperties config) {
    return camelContext.createConsumerTemplate(config.getConsumerTemplateCacheSize());
}

@Test
public void testHasTwoTemplates() {
    ConsumerTemplate lookup = context.getRegistry().lookupByNameAndType("myTemplate", ConsumerTemplate.class);
    assertNotNull("Should lookup producer template", lookup);

    ConsumerTemplate lookup2 = context.getRegistry().lookupByNameAndType("myOtherTemplate", ConsumerTemplate.class);
    assertNotNull("Should lookup producer template", lookup2);

    assertNotSame("Should not be same", lookup, lookup2);
}

@Test
public void testHasTemplate() {
    assertNotNull("Should have injected a consumer template", template);
    assertNotNull("The template context should not be null", ((DefaultConsumerTemplate)template).getCamelContext());

    ConsumerTemplate lookup = context.getRegistry().lookupByNameAndType("consumerTemplate", ConsumerTemplate.class);
    assertNotNull("Should lookup consumer template", lookup);
}

@Test
public void testHasExistingTemplate() {
    assertNotNull("Should have injected a consumer template", template);

    ConsumerTemplate lookup = context.getRegistry().lookupByNameAndType("myConsumerTemplate", ConsumerTemplate.class);
    assertNotNull("Should lookup consumer template", lookup);

    ConsumerTemplate lookup2 = context.getRegistry().lookupByNameAndType("consumerTemplate", ConsumerTemplate.class);
    assertNull("Should not be able to lookup consumer template", lookup2);
}

@Test
public void testShouldBeSingleton() {
    ConsumerTemplate lookup = context.getRegistry().lookupByNameAndType("myConsumerTemplate", ConsumerTemplate.class);
    assertNotNull("Should lookup producer template", lookup);

    ConsumerTemplate lookup2 = context.getRegistry().lookupByNameAndType("myConsumerTemplate", ConsumerTemplate.class);
    assertNotNull("Should lookup producer template", lookup);

    assertSame("Should be same instances (singleton)", lookup, lookup2);
}

/**
 * Creates and starts an application scope ConsumeTemplate that has a default cache size of 1000 and is not associated with any endpoint.
 *
 * @return an application scope ConsumeTemplate that has a default cache size of 1000 and is not associated with any endpoint.
 */
@Produces
@Omakase
@ApplicationScoped
public ConsumerTemplate getConsumerTemplate() {
    ConsumerTemplate consumerTemplate = camelContext.createConsumerTemplate(1);
    LOGGER.info("Created Camel Consumer Template with max cache size " + consumerTemplate.getMaximumCacheSize());
    return consumerTemplate;
}

/**
 * Creates and starts an application scope ConsumeTemplate that has a default cache size of 1000 and is not associated with any endpoint.
 *
 * @return an application scope ConsumeTemplate that has a default cache size of 1000 and is not associated with any endpoint.
 */
@Produces
@Omakase
@ApplicationScoped
public ConsumerTemplate getConsumerTemplate() {
    ConsumerTemplate consumerTemplate = camelContext.createConsumerTemplate();
    LOGGER.info("Created Camel Consumer Template with max cache size " + consumerTemplate.getMaximumCacheSize());
    return consumerTemplate;
}

@Test
public void testSeda() throws Exception {
    // send to the seda inbox queue
    producer.sendBody("Hello Swarm");

    ConsumerTemplate consumer = camelContext.createConsumerTemplate();
    // use 5 second timeout to receive the message from outbox
    Object body = consumer.receiveBody("seda:outbox", 5000);

    // expect it was the message we sent
    assertEquals("Hello Swarm", body);
}

@Test
public void defaults() throws Exception {
  Set<Object> routes = Sets.newHashSet(rb, new Object());
  new MockUnit(GuiceInjector.class, DefaultCamelContext.class, RouteBuilder.class,
      ProducerTemplate.class, ConsumerTemplate.class)
      .expect(ctx)
      .run(unit -> {
        new CamelFinalizer(unit.get(GuiceInjector.class),
            unit.get(DefaultCamelContext.class), routes,
            unit.get(RouteBuilder.class), unit.get(ProducerTemplate.class), unit
                .get(ConsumerTemplate.class));
      });
}

@Test
public void testSAPNetweaverEndpoint() throws Exception {

    Assume.assumeTrue("[#1675] Enable SAP testing in Jenkins", SAP_USERNAME != null && SAP_PASSWORD != null);

    CamelContext camelctx = new DefaultCamelContext();
    camelctx.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("direct:start")
            .toF("sap-netweaver:%s?username=%s&password=%s", SAP_GATEWAY_URL, SAP_USERNAME, SAP_PASSWORD);
        }
    });

    camelctx.start();
    try {
        ProducerTemplate producer = camelctx.createProducerTemplate();
        ConsumerTemplate consumer = camelctx.createConsumerTemplate();

        // Flight data is constantly updated, so fetch a valid flight from the flight collection feed
        String sapRssFeedUri = String.format("rss:%s/%s?username=%s&password=%s", SAP_GATEWAY_URL.replace("https4", "https"),
            "FlightCollection", SAP_USERNAME, SAP_PASSWORD);
        SyndFeed feed = consumer.receiveBody(sapRssFeedUri, SyndFeed.class);
        Assert.assertNotNull(feed);
        Assert.assertTrue(feed.getEntries().size() > 0);

        SyndEntry entry = (SyndEntry) feed.getEntries().get(0);
        String sapCommand = entry.getTitle();
        String result = producer.requestBodyAndHeader("direct:start", null, NetWeaverConstants.COMMAND, sapCommand, String.class);
        Assert.assertFalse(result.isEmpty());
    } finally {
        camelctx.stop();
    }
}