@Test public void testProcess() throws Exception { Map<String, Object> exchangeProps = new HashMap<String, Object>(); Exchange exchange = mock(Exchange.class, RETURNS_MOCKS); when(exchange.getPattern()).thenReturn(ExchangePattern.InOut); when(exchange.getProperties()).thenReturn(exchangeProps); Processor processor = mock(Processor.class); doAnswer(new Answer<Object>() { @Override public Object answer(InvocationOnMock invocation) throws Throwable { Exchange exchange = (Exchange) invocation.getArguments()[0]; exchange.setProperty("Hello", "World"); return null; } }).when(processor).process(any(Exchange.class)); OsgiDefaultEndpoint endpoint = mock(OsgiDefaultEndpoint.class); OsgiDefaultConsumer consumer = new OsgiDefaultConsumer(endpoint, processor, Collections.<String, Object>emptyMap()); consumer.process(exchange); verify(processor).process(any(Exchange.class)); assertThat((String) exchangeProps.get("Hello"), equalTo("World")); }
@Override public void process(final Exchange exchange) throws Exception { Exchange copy = copyExchange(exchange); try { processor.process(copy); } finally { ExchangeHelper.copyResults(exchange, copy); } }
OsgiDefaultConsumer extends ServiceSupport implements Consumer, SuspendableService, Processor { @Override public void process(final Exchange exchange) throws Exception { Exchange copy = copyExchange(exchange); try { processor.process(copy); } finally { ExchangeHelper.copyResults(exchange, copy); } } }
OsgiDefaultConsumer extends ServiceSupport implements Consumer, SuspendableService, Processor { @Override public void process(final Exchange exchange) throws Exception { Exchange copy = copyExchange(exchange); try { processor.process(copy); } finally { ExchangeHelper.copyResults(exchange, copy); } } OsgiDefaultConsumer(OsgiDefaultEndpoint endpoint, Processor processor, Map<String, Object> props); }
OsgiDefaultConsumer extends ServiceSupport implements Consumer, SuspendableService, Processor { @Override public void process(final Exchange exchange) throws Exception { Exchange copy = copyExchange(exchange); try { processor.process(copy); } finally { ExchangeHelper.copyResults(exchange, copy); } } OsgiDefaultConsumer(OsgiDefaultEndpoint endpoint, Processor processor, Map<String, Object> props); @Override OsgiDefaultEndpoint getEndpoint(); @Override void process(final Exchange exchange); }
OsgiDefaultConsumer extends ServiceSupport implements Consumer, SuspendableService, Processor { @Override public void process(final Exchange exchange) throws Exception { Exchange copy = copyExchange(exchange); try { processor.process(copy); } finally { ExchangeHelper.copyResults(exchange, copy); } } OsgiDefaultConsumer(OsgiDefaultEndpoint endpoint, Processor processor, Map<String, Object> props); @Override OsgiDefaultEndpoint getEndpoint(); @Override void process(final Exchange exchange);  }