CamelEndpoint extends AbstractEndpoint { @Override public Consumer createConsumer() { if (camelConsumer == null) { camelConsumer = new CamelConsumer(getConsumerName(), getEndpointConfiguration()); } return camelConsumer; } CamelEndpoint();  CamelEndpoint(CamelEndpointConfiguration endpointConfiguration); @Override Producer createProducer(); @Override Consumer createConsumer(); @Override CamelEndpointConfiguration getEndpointConfiguration();  }
@Test public void testCamelEndpointConsumer() { String endpointUri = "direct:news-feed"; CamelEndpointConfiguration endpointConfiguration = new CamelEndpointConfiguration(); endpointConfiguration.setCamelContext(camelContext); endpointConfiguration.setEndpointUri(endpointUri); CamelEndpoint camelEndpoint = new CamelEndpoint(endpointConfiguration); reset(camelContext, consumerTemplate); when(camelContext.createConsumerTemplate()).thenReturn(consumerTemplate); when(camelContext.getHeadersMapFactory()).thenReturn(new DefaultHeadersMapFactory()); when(camelContext.getUuidGenerator()).thenReturn(new SimpleUuidGenerator()); DefaultMessage message = new DefaultMessage(camelContext); message.setBody("Hello from Camel!"); message.setHeader("operation", "newsFeed"); Exchange exchange = new DefaultExchange(camelContext); exchange.setIn(message); when(consumerTemplate.receive(endpointUri, endpointConfiguration.getTimeout())).thenReturn(exchange); Message receivedMessage = camelEndpoint.createConsumer().receive(context, endpointConfiguration.getTimeout()); Assert.assertEquals(receivedMessage.getPayload(), "Hello from Camel!"); Assert.assertEquals(receivedMessage.getHeader("operation"), "newsFeed"); Assert.assertNotNull(receivedMessage.getHeader(CitrusCamelMessageHeaders.EXCHANGE_ID)); Assert.assertNotNull(receivedMessage.getHeader(CitrusCamelMessageHeaders.EXCHANGE_PATTERN)); Assert.assertNotNull(receivedMessage.getHeader(CitrusCamelMessageHeaders.EXCHANGE_FAILED)); }