-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Christian Ribeaud
committed
Mar 10, 2017
1 parent
1825fd3
commit d90323f
Showing
6 changed files
with
220 additions
and
45 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
93 changes: 93 additions & 0 deletions
93
src/main/java/org/apache/camel/component/dataprovider/DataProviderPollingConsumer.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,93 @@ | ||
| package org.apache.camel.component.dataprovider; | ||
|
|
||
| import com.google.common.collect.Iterables; | ||
| import com.google.common.collect.Range; | ||
| import org.apache.camel.CamelContext; | ||
| import org.apache.camel.Exchange; | ||
| import org.apache.camel.Message; | ||
| import org.apache.camel.impl.PollingConsumerSupport; | ||
| import org.apache.camel.util.ObjectHelper; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import java.util.concurrent.*; | ||
|
|
||
| /** | ||
| * {@link PollingConsumerSupport} extension for {@link IDataProvider}. | ||
| * | ||
| * @author <a href="mailto:[email protected]">Christian Ribeaud</a> | ||
| */ | ||
| public class DataProviderPollingConsumer extends PollingConsumerSupport { | ||
|
|
||
| private static final Logger LOG = LoggerFactory.getLogger(DataProviderPollingConsumer.class); | ||
|
|
||
| private ExecutorService executorService; | ||
|
|
||
| DataProviderPollingConsumer(DataProviderEndpoint dataProviderEndpoint) { | ||
| super(dataProviderEndpoint); | ||
| } | ||
|
|
||
| @Override | ||
| public Exchange receive() { | ||
| Range<Integer> range = createRange(); | ||
| LogUtils.info(LOG, () -> String.format("Handling range '%s'.", range)); | ||
| Iterable<?> partition = getEndpoint().getDataProvider().partition(range); | ||
| Exchange exchange = getEndpoint().createExchange(); | ||
| int size = Iterables.size(partition); | ||
| Message in = exchange.getIn(); | ||
| switch (size) { | ||
| case 1: | ||
| in.setBody(Iterables.get(partition, 0)); | ||
| break; | ||
| default: | ||
| in.setBody(partition); | ||
| } | ||
| return exchange; | ||
| } | ||
|
|
||
| @Override | ||
| public Exchange receiveNoWait() { | ||
| return receive(); | ||
| } | ||
|
|
||
| @Override | ||
| public Exchange receive(long timeout) { | ||
| // the task is the receive method | ||
| Future<Exchange> future = executorService.submit((Callable<Exchange>) this::receive); | ||
| try { | ||
| return future.get(timeout, TimeUnit.MILLISECONDS); | ||
| } catch (ExecutionException | InterruptedException e) { | ||
| throw ObjectHelper.wrapRuntimeCamelException(e); | ||
| } catch (TimeoutException e) { | ||
| // ignore as we hit timeout then return null | ||
| } | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| protected void doStart() throws Exception { | ||
| executorService = getCamelContext().getExecutorServiceManager().newDefaultThreadPool(this, getClass().getSimpleName()); | ||
| } | ||
|
|
||
| @Override | ||
| protected void doStop() throws Exception { | ||
| getCamelContext().getExecutorServiceManager().shutdown(executorService); | ||
| } | ||
|
|
||
| private CamelContext getCamelContext() { | ||
| return getEndpoint().getCamelContext(); | ||
| } | ||
|
|
||
| @Override | ||
| public DataProviderEndpoint getEndpoint() { | ||
| return (DataProviderEndpoint) super.getEndpoint(); | ||
| } | ||
|
|
||
| private Range<Integer> createRange() { | ||
| int size = getEndpoint().getDataProvider().getSize(); | ||
| int rangeSize = getEndpoint().getRangeSize(); | ||
| rangeSize = rangeSize < 0 ? size : rangeSize; | ||
| int lower = getEndpoint().getStartingIndex(); | ||
| return Range.closedOpen(Math.min(lower, size - 1), Math.min(size, lower + rangeSize)); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
48 changes: 48 additions & 0 deletions
48
src/test/java/org/apache/camel/component/dataprovider/DataProviderPollingConsumerTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,48 @@ | ||
| package org.apache.camel.component.dataprovider; | ||
|
|
||
| import org.apache.camel.impl.JndiRegistry; | ||
| import org.apache.camel.testng.CamelTestSupport; | ||
| import org.testng.annotations.Test; | ||
|
|
||
| import java.util.Arrays; | ||
| import java.util.List; | ||
|
|
||
| /** | ||
| * Test cases for corresponding class {@link DataProviderPollingConsumer}. | ||
| * | ||
| * @author <a href="mailto:[email protected]">Christian Ribeaud</a> | ||
| */ | ||
| public class DataProviderPollingConsumerTest extends CamelTestSupport { | ||
|
|
||
| @Test | ||
| public void testReceiveBodyWithRangeSizeAndStartingIndex() { | ||
| List<String> body = consumer.receiveBody("dataprovider://foo?rangeSize=2&startingIndex=3", List.class); | ||
| assertEquals(body.size(), 2); | ||
| assertEquals(body.get(0), "Four"); | ||
| } | ||
|
|
||
| @Test | ||
| public void testReceiveBody() { | ||
| List<String> body = consumer.receiveBody("dataprovider://foo", List.class); | ||
| assertEquals(body.size(), 10); | ||
| } | ||
|
|
||
| @Test | ||
| public void testReceiveBodyWithOneString() { | ||
| String body = consumer.receiveBody("dataprovider://foo?rangeSize=1&startingIndex=12345", String.class); | ||
| assertEquals(body, "Ten"); | ||
| } | ||
|
|
||
| @Test | ||
| public void testReceiveBodyWithZeroRangeSize() { | ||
| List<String> body = consumer.receiveBody("dataprovider://foo?rangeSize=0&startingIndex=4", List.class); | ||
| assertEquals(body.size(), 0); | ||
| } | ||
|
|
||
| @Override | ||
| protected JndiRegistry createRegistry() throws Exception { | ||
| JndiRegistry registry = super.createRegistry(); | ||
| registry.bind("foo", new StaticDataProvider<>(Arrays.asList("One", "Two", "Three", "Four", "Five", "Six", "Seven", "Eight", "Nine", "Ten"))); | ||
| return registry; | ||
| } | ||
| } |