You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2015/07/17 15:22:02 UTC
[3/3] camel git commit: Polished and fixed CS and a few other bits
Polished and fixed CS and a few other bits
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/832a99c5
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/832a99c5
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/832a99c5
Branch: refs/heads/master
Commit: 832a99c546d4fca5510a88136de8fb2002c54d6f
Parents: 2ba152d
Author: Claus Ibsen <da...@apache.org>
Authored: Fri Jul 17 15:25:15 2015 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Fri Jul 17 15:25:15 2015 +0200
----------------------------------------------------------------------
.../camel/component/sjms/SjmsEndpoint.java | 7 +-
.../component/sjms/batch/SessionCompletion.java | 22 ++--
.../sjms/batch/SjmsBatchComponent.java | 14 +--
.../component/sjms/batch/SjmsBatchConsumer.java | 79 +++++++-------
.../component/sjms/batch/SjmsBatchEndpoint.java | 104 +++++++++----------
.../sjms/jms/DestinationNameParser.java | 1 +
.../sjms/batch/ListAggregationStrategy.java | 12 +--
.../sjms/batch/SjmsBatchConsumerTest.java | 43 ++++----
.../sjms/batch/SjmsBatchEndpointTest.java | 11 +-
.../sjms/jms/DestinationNameParserTest.java | 4 +-
.../sjms/producer/QueueProducerQoSTest.java | 2 +-
11 files changed, 145 insertions(+), 154 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/832a99c5/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java
index 0e8d68a..6ffa513 100644
--- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java
+++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java
@@ -22,7 +22,12 @@ import org.apache.camel.ExchangePattern;
import org.apache.camel.MultipleConsumersSupport;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
-import org.apache.camel.component.sjms.jms.*;
+import org.apache.camel.component.sjms.jms.ConnectionResource;
+import org.apache.camel.component.sjms.jms.DefaultDestinationCreationStrategy;
+import org.apache.camel.component.sjms.jms.DestinationCreationStrategy;
+import org.apache.camel.component.sjms.jms.DestinationNameParser;
+import org.apache.camel.component.sjms.jms.KeyFormatStrategy;
+import org.apache.camel.component.sjms.jms.SessionAcknowledgementType;
import org.apache.camel.component.sjms.producer.InOnlyProducer;
import org.apache.camel.component.sjms.producer.InOutProducer;
import org.apache.camel.impl.DefaultEndpoint;
http://git-wip-us.apache.org/repos/asf/camel/blob/832a99c5/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SessionCompletion.java
----------------------------------------------------------------------
diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SessionCompletion.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SessionCompletion.java
index 27f06e6..cae90cb 100644
--- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SessionCompletion.java
+++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SessionCompletion.java
@@ -16,34 +16,32 @@
*/
package org.apache.camel.component.sjms.batch;
+import javax.jms.JMSException;
+import javax.jms.Session;
+
import org.apache.camel.Exchange;
import org.apache.camel.spi.Synchronization;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import javax.jms.JMSException;
-import javax.jms.Session;
-
-/**
- * @author jkorab
- */
class SessionCompletion implements Synchronization {
- private final Logger log = LoggerFactory.getLogger(this.getClass());
+ private static final Logger LOG = LoggerFactory.getLogger(SessionCompletion.class);
private final Session session;
+ // TODO: add more details in the commit/rollback eg such as message id
+
public SessionCompletion(Session session) {
- assert (session != null);
this.session = session;
}
@Override
public void onComplete(Exchange exchange) {
try {
- log.debug("Committing");
+ LOG.debug("Committing");
session.commit();
} catch (JMSException ex) {
- log.error("Exception caught while committing: {}", ex.getMessage());
+ LOG.error("Exception caught while committing: {}", ex.getMessage());
exchange.setException(ex);
}
}
@@ -51,10 +49,10 @@ class SessionCompletion implements Synchronization {
@Override
public void onFailure(Exchange exchange) {
try {
- log.debug("Rolling back");
+ LOG.debug("Rolling back");
session.rollback();
} catch (JMSException ex) {
- log.error("Exception caught while rolling back: {}", ex.getMessage());
+ LOG.error("Exception caught while rolling back: {}", ex.getMessage());
exchange.setException(ex);
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/832a99c5/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SjmsBatchComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SjmsBatchComponent.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SjmsBatchComponent.java
index 04b875d..421fd8a 100644
--- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SjmsBatchComponent.java
+++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SjmsBatchComponent.java
@@ -16,16 +16,13 @@
*/
package org.apache.camel.component.sjms.batch;
+import java.util.Map;
+import javax.jms.ConnectionFactory;
+
import org.apache.camel.Endpoint;
import org.apache.camel.impl.UriEndpointComponent;
import org.apache.camel.util.ObjectHelper;
-import javax.jms.ConnectionFactory;
-import java.util.Map;
-
-/**
- * @author jkorab
- */
public class SjmsBatchComponent extends UriEndpointComponent {
private ConnectionFactory connectionFactory;
@@ -36,7 +33,7 @@ public class SjmsBatchComponent extends UriEndpointComponent {
@Override
protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
- ObjectHelper.notNull(connectionFactory, "connectionFactory is null");
+ ObjectHelper.notNull(connectionFactory, "connectionFactory");
SjmsBatchEndpoint sjmsBatchEndpoint = new SjmsBatchEndpoint(uri, this, remaining);
setProperties(sjmsBatchEndpoint, parameters);
return sjmsBatchEndpoint;
@@ -46,6 +43,9 @@ public class SjmsBatchComponent extends UriEndpointComponent {
return connectionFactory;
}
+ /**
+ * A ConnectionFactory is required to enable the SjmsBatchComponent.
+ */
public void setConnectionFactory(ConnectionFactory connectionFactory) {
this.connectionFactory = connectionFactory;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/832a99c5/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SjmsBatchConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SjmsBatchConsumer.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SjmsBatchConsumer.java
index ca47c7c..ee2b250 100644
--- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SjmsBatchConsumer.java
+++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SjmsBatchConsumer.java
@@ -16,16 +16,6 @@
*/
package org.apache.camel.component.sjms.batch;
-import org.apache.camel.Exchange;
-import org.apache.camel.Processor;
-import org.apache.camel.component.sjms.jms.JmsMessageHelper;
-import org.apache.camel.impl.DefaultConsumer;
-import org.apache.camel.processor.aggregate.AggregationStrategy;
-import org.apache.camel.util.ObjectHelper;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.jms.*;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.Date;
@@ -36,30 +26,47 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.ObjectMessage;
+import javax.jms.Queue;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.component.sjms.jms.JmsMessageHelper;
+import org.apache.camel.impl.DefaultConsumer;
+import org.apache.camel.processor.aggregate.AggregationStrategy;
+import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
-/**
- * @author jkorab
- */
public class SjmsBatchConsumer extends DefaultConsumer {
private static final boolean TRANSACTED = true;
- private final Logger LOG = LoggerFactory.getLogger(SjmsBatchConsumer.class);
+ private static final Logger LOG = LoggerFactory.getLogger(SjmsBatchConsumer.class);
+
+ // global counters, maybe they should be on component instead?
+ private static final AtomicInteger BATCH_COUNT = new AtomicInteger();
+ private static final AtomicLong MESSAGE_RECEIVED = new AtomicLong();
+ private static final AtomicLong MESSAGE_PROCESSED = new AtomicLong();
private final SjmsBatchEndpoint sjmsBatchEndpoint;
private final AggregationStrategy aggregationStrategy;
-
private final int completionSize;
private final int completionTimeout;
private final int consumerCount;
private final int pollDuration;
-
private final ConnectionFactory connectionFactory;
private final String destinationName;
private final Processor processor;
-
- private static AtomicInteger batchCount = new AtomicInteger();
- private static AtomicLong messagesReceived = new AtomicLong();
- private static AtomicLong messagesProcessed = new AtomicLong();
private ExecutorService jmsConsumerExecutors;
+ private final AtomicBoolean running = new AtomicBoolean(true);
+ private final AtomicReference<CountDownLatch> consumersShutdownLatchRef = new AtomicReference<>();
+ private Connection connection;
public SjmsBatchConsumer(SjmsBatchEndpoint sjmsBatchEndpoint, Processor processor) {
super(sjmsBatchEndpoint, processor);
@@ -92,11 +99,6 @@ public class SjmsBatchConsumer extends DefaultConsumer {
return sjmsBatchEndpoint;
}
- private final AtomicBoolean running = new AtomicBoolean(true);
- private final AtomicReference<CountDownLatch> consumersShutdownLatchRef = new AtomicReference<>();
-
- private Connection connection;
-
@Override
protected void doStart() throws Exception {
super.doStart();
@@ -145,8 +147,7 @@ public class SjmsBatchConsumer extends DefaultConsumer {
LOG.error("Exception caught closing connection: {}", getStackTrace(jex));
}
- getEndpoint().getCamelContext().getExecutorServiceManager()
- .shutdown(jmsConsumerExecutors);
+ getEndpoint().getCamelContext().getExecutorServiceManager().shutdown(jmsConsumerExecutors);
}
private String getStackTrace(Exception ex) {
@@ -174,7 +175,6 @@ public class SjmsBatchConsumer extends DefaultConsumer {
consumer.close();
} catch (JMSException ex2) {
log.error("Exception caught closing consumer: {}", ex2.getMessage());
-
}
}
} finally {
@@ -197,7 +197,7 @@ public class SjmsBatchConsumer extends DefaultConsumer {
private void consumeBatchesOnLoop(Session session, MessageConsumer consumer) throws JMSException {
final boolean usingTimeout = completionTimeout > 0;
- batchConsumption:
+ batchConsumption:
while (running.get()) {
int messageCount = 0;
@@ -206,7 +206,7 @@ public class SjmsBatchConsumer extends DefaultConsumer {
long startTime = 0;
Exchange aggregatedExchange = null;
- batch:
+ batch:
while ((completionSize <= 0) || (messageCount < completionSize)) {
// check periodically to see whether we should be shutting down
long waitTime = (usingTimeout && (timeElapsed > 0))
@@ -219,7 +219,7 @@ public class SjmsBatchConsumer extends DefaultConsumer {
// timed out, no message received
LOG.trace("No message received");
} else {
- if ((usingTimeout) && (messageCount == 0)) { // this is the first message
+ if (usingTimeout && messageCount == 0) { // this is the first message
startTime = new Date().getTime(); // start counting down the period for this batch
}
messageCount++;
@@ -230,12 +230,11 @@ public class SjmsBatchConsumer extends DefaultConsumer {
aggregatedExchange = aggregationStrategy.aggregate(aggregatedExchange, exchange);
aggregatedExchange.setProperty(SjmsBatchEndpoint.PROPERTY_BATCH_SIZE, messageCount);
} else {
- throw new IllegalArgumentException("Unexpected message type: "
- + message.getClass().toString());
+ throw new IllegalArgumentException("Unexpected message type: " + message.getClass().toString());
}
}
- if ((usingTimeout) && (startTime > 0)) {
+ if (usingTimeout && startTime > 0) {
// a batch has been started, check whether it should be timed out
long currentTime = new Date().getTime();
timeElapsed = currentTime - startTime;
@@ -252,13 +251,13 @@ public class SjmsBatchConsumer extends DefaultConsumer {
break batchConsumption;
}
} // batch
- assert (aggregatedExchange != null);
process(aggregatedExchange, session);
}
}
/**
* Determine the time that a call to {@link MessageConsumer#receive()} should wait given the time that has elapsed for this batch.
+ *
* @param timeElapsed The time that has elapsed.
* @return The shorter of the time remaining or poll duration.
*/
@@ -277,25 +276,25 @@ public class SjmsBatchConsumer extends DefaultConsumer {
private long getTimeRemaining(long timeElapsed) {
long timeRemaining = completionTimeout - timeElapsed;
- if (LOG.isDebugEnabled() && (timeElapsed > 0)) {
+ if (LOG.isDebugEnabled() && timeElapsed > 0) {
LOG.debug("Time remaining this batch: {}", timeRemaining);
}
return timeRemaining;
}
private void process(Exchange exchange, Session session) {
- assert (exchange != null);
- int id = batchCount.getAndIncrement();
+ int id = BATCH_COUNT.getAndIncrement();
int batchSize = exchange.getProperty(SjmsBatchEndpoint.PROPERTY_BATCH_SIZE, Integer.class);
if (LOG.isDebugEnabled()) {
- LOG.debug("Processing batch[" + id + "]:size=" + batchSize + ":total=" + messagesReceived.addAndGet(batchSize));
+ LOG.debug("Processing batch[" + id + "]:size=" + batchSize + ":total=" + MESSAGE_RECEIVED.addAndGet(batchSize));
}
SessionCompletion sessionCompletion = new SessionCompletion(session);
exchange.addOnCompletion(sessionCompletion);
try {
processor.process(exchange);
- LOG.debug("Completed processing[{}]:total={}", id, messagesProcessed.addAndGet(batchSize));
+ long total = MESSAGE_PROCESSED.addAndGet(batchSize);
+ LOG.debug("Completed processing[{}]:total={}", id, total);
} catch (Exception e) {
LOG.error("Error processing exchange: {}", e.getMessage());
}
http://git-wip-us.apache.org/repos/asf/camel/blob/832a99c5/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SjmsBatchEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SjmsBatchEndpoint.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SjmsBatchEndpoint.java
index 5d307a7..b4c052f 100644
--- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SjmsBatchEndpoint.java
+++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/batch/SjmsBatchEndpoint.java
@@ -28,12 +28,7 @@ import org.apache.camel.spi.UriEndpoint;
import org.apache.camel.spi.UriParam;
import org.apache.camel.spi.UriPath;
-/**
- * @author jkorab
- */
-@UriEndpoint(scheme = "sjmsBatch",
- title = "Simple JMS Batch Component",
- syntax = "sjms-batch:destinationName?aggregationStrategy=#aggStrategy",
+@UriEndpoint(scheme = "sjms-batch", title = "Simple JMS Batch Component", syntax = "sjms-batch:destinationName",
consumerClass = SjmsBatchComponent.class, label = "messaging")
public class SjmsBatchEndpoint extends DefaultEndpoint {
@@ -41,41 +36,29 @@ public class SjmsBatchEndpoint extends DefaultEndpoint {
public static final int DEFAULT_COMPLETION_TIMEOUT = 500;
public static final String PROPERTY_BATCH_SIZE = "CamelSjmsBatchSize";
- @UriPath
- @Metadata(required = "true",
- description = "The destination name. Only queues are supported, names may be prefixed by 'queue:'.")
+ @UriPath(label = "consumer") @Metadata(required = "true")
private String destinationName;
-
- @UriParam(label = "consumer", defaultValue = "1", description = "The number of JMS sessions to consume from")
- private Integer consumerCount = 1;
-
- @UriParam(label = "consumer", defaultValue = "200",
- description = "The number of messages consumed at which the batch will be completed")
- private Integer completionSize = DEFAULT_COMPLETION_SIZE;
-
- @UriParam(label = "consumer", defaultValue = "500",
- description = "The timeout from receipt of the first first message when the batch will be completed")
- private Integer completionTimeout = DEFAULT_COMPLETION_TIMEOUT;
-
- @UriParam(label = "consumer", defaultValue = "1000",
- description = "The duration in milliseconds of each poll for messages. " +
- "completionTimeOut will be used if it is shorter and a batch has started.")
- private Integer pollDuration = 1000;
-
- @Metadata(required = "true")
- @UriParam(label = "consumer", description = "A #-reference to an AggregationStrategy visible to Camel")
+ @UriParam(label = "consumer", defaultValue = "1")
+ private int consumerCount = 1;
+ @UriParam(label = "consumer", defaultValue = "200")
+ private int completionSize = DEFAULT_COMPLETION_SIZE;
+ @UriParam(label = "consumer", defaultValue = "500")
+ private int completionTimeout = DEFAULT_COMPLETION_TIMEOUT;
+ @UriParam(label = "consumer", defaultValue = "1000")
+ private int pollDuration = 1000;
+ @UriParam(label = "consumer") @Metadata(required = "true")
private AggregationStrategy aggregationStrategy;
- private boolean topic;
-
- public SjmsBatchEndpoint() {}
+ public SjmsBatchEndpoint() {
+ }
public SjmsBatchEndpoint(String endpointUri, Component component, String remaining) {
super(endpointUri, component);
+
DestinationNameParser parser = new DestinationNameParser();
if (parser.isTopic(remaining)) {
- throw new IllegalArgumentException("Only batch consumption from queues is supported. For topics you " +
- "should use a regular JMS consumer with an aggregator.");
+ throw new IllegalArgumentException("Only batch consumption from queues is supported. For topics you "
+ + "should use a regular JMS consumer with an aggregator.");
}
this.destinationName = parser.getShortName(remaining);
}
@@ -99,47 +82,62 @@ public class SjmsBatchEndpoint extends DefaultEndpoint {
return aggregationStrategy;
}
+ /**
+ * The aggregation strategy to use, which merges all the batched messages into a single message
+ */
public void setAggregationStrategy(AggregationStrategy aggregationStrategy) {
this.aggregationStrategy = aggregationStrategy;
}
- public Integer getCompletionSize() {
- return completionSize;
- }
-
- public void setCompletionSize(Integer completionSize) {
- this.completionSize = completionSize;
+ /**
+ * The destination name. Only queues are supported, names may be prefixed by 'queue:'.
+ */
+ public String getDestinationName() {
+ return destinationName;
}
- public Integer getCompletionTimeout() {
- return completionTimeout;
+ public int getConsumerCount() {
+ return consumerCount;
}
- public void setCompletionTimeout(Integer completionTimeout) {
- this.completionTimeout = completionTimeout;
+ /**
+ * The number of JMS sessions to consume from
+ */
+ public void setConsumerCount(int consumerCount) {
+ this.consumerCount = consumerCount;
}
- public String getDestinationName() {
- return destinationName;
+ public int getCompletionSize() {
+ return completionSize;
}
- public void setDestinationName(String destinationName) {
- this.destinationName = destinationName;
+ /**
+ * The number of messages consumed at which the batch will be completed
+ */
+ public void setCompletionSize(int completionSize) {
+ this.completionSize = completionSize;
}
- public Integer getConsumerCount() {
- return consumerCount;
+ public int getCompletionTimeout() {
+ return completionTimeout;
}
- public void setConsumerCount(Integer consumerCount) {
- this.consumerCount = consumerCount;
+ /**
+ * The timeout from receipt of the first first message when the batch will be completed
+ */
+ public void setCompletionTimeout(int completionTimeout) {
+ this.completionTimeout = completionTimeout;
}
- public Integer getPollDuration() {
+ public int getPollDuration() {
return pollDuration;
}
- public void setPollDuration(Integer pollDuration) {
+ /**
+ * The duration in milliseconds of each poll for messages.
+ * completionTimeOut will be used if it is shorter and a batch has started.
+ */
+ public void setPollDuration(int pollDuration) {
this.pollDuration = pollDuration;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/832a99c5/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/DestinationNameParser.java
----------------------------------------------------------------------
diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/DestinationNameParser.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/DestinationNameParser.java
index d248350..ddc213d 100644
--- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/DestinationNameParser.java
+++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/DestinationNameParser.java
@@ -20,6 +20,7 @@ package org.apache.camel.component.sjms.jms;
* @author jkorab
*/
public class DestinationNameParser {
+
public boolean isTopic(String destinationName) {
if (destinationName == null) {
throw new IllegalArgumentException("destinationName is null");
http://git-wip-us.apache.org/repos/asf/camel/blob/832a99c5/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/ListAggregationStrategy.java
----------------------------------------------------------------------
diff --git a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/ListAggregationStrategy.java b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/ListAggregationStrategy.java
index 39774b2..d1eb6e5 100644
--- a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/ListAggregationStrategy.java
+++ b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/ListAggregationStrategy.java
@@ -16,16 +16,14 @@
*/
package org.apache.camel.component.sjms.batch;
-import org.apache.camel.Exchange;
-import org.apache.camel.processor.aggregate.AggregationStrategy;
-
import java.util.ArrayList;
import java.util.List;
-/**
- * @author jkorab
- */
+import org.apache.camel.Exchange;
+import org.apache.camel.processor.aggregate.AggregationStrategy;
+
public class ListAggregationStrategy implements AggregationStrategy {
+
@Override
public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
String body = newExchange.getIn().getBody(String.class);
@@ -34,7 +32,7 @@ public class ListAggregationStrategy implements AggregationStrategy {
list.add(body);
newExchange.getIn().setBody(list);
return newExchange;
- } else {
+ } else {
List<String> list = oldExchange.getIn().getBody(List.class);
list.add(body);
return oldExchange;
http://git-wip-us.apache.org/repos/asf/camel/blob/832a99c5/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/SjmsBatchConsumerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/SjmsBatchConsumerTest.java b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/SjmsBatchConsumerTest.java
index 642a38f..76c739b 100644
--- a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/SjmsBatchConsumerTest.java
+++ b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/SjmsBatchConsumerTest.java
@@ -16,8 +16,12 @@
*/
package org.apache.camel.component.sjms.batch;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.List;
+import javax.jms.ConnectionFactory;
+
import org.apache.activemq.ActiveMQConnectionFactory;
-import org.apache.activemq.broker.jmx.DestinationViewMBean;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.LoggingLevel;
@@ -34,16 +38,8 @@ import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import javax.jms.ConnectionFactory;
-import java.text.SimpleDateFormat;
-import java.util.Date;
-import java.util.List;
-
-/**
- * @author jkorab
- */
public class SjmsBatchConsumerTest extends CamelTestSupport {
- private final Logger LOG = LoggerFactory.getLogger(SjmsBatchConsumerTest.class);
+ private static final Logger LOG = LoggerFactory.getLogger(SjmsBatchConsumerTest.class);
@Rule
public EmbeddedActiveMQBroker broker = new EmbeddedActiveMQBroker("localhost");
@@ -76,10 +72,10 @@ public class SjmsBatchConsumerTest extends CamelTestSupport {
@Override
public void configure() throws Exception {
from("direct:in").routeId("harness").startupOrder(20)
- .split(body())
+ .split(body())
.toF("sjms:queue:%s?transacted=true", queueName)
.to("mock:before")
- .end();
+ .end();
}
}
@@ -102,12 +98,11 @@ public class SjmsBatchConsumerTest extends CamelTestSupport {
int completionTimeout = 1000;
int completionSize = 200;
- fromF("sjms-batch:%s?completionTimeout=%s&completionSize=%s" +
- "&consumerCount=%s&aggregationStrategy=#testStrategy",
+ fromF("sjms-batch:%s?completionTimeout=%s&completionSize=%s&consumerCount=%s&aggregationStrategy=#testStrategy",
queueName, completionTimeout, completionSize, consumerCount)
.routeId("batchConsumer").startupOrder(10).autoStartup(false)
- .split(body())
- .to("mock:split");
+ .split(body())
+ .to("mock:split");
}
});
context.start();
@@ -132,7 +127,7 @@ public class SjmsBatchConsumerTest extends CamelTestSupport {
}
@Test
- public void testConsumption_completionSize() throws Exception {
+ public void testConsumptionCompletionSize() throws Exception {
final int completionSize = 5;
final int completionTimeout = -1; // size-based only
@@ -157,7 +152,7 @@ public class SjmsBatchConsumerTest extends CamelTestSupport {
}
@Test
- public void testConsumption_completionTimeout() throws Exception {
+ public void testConsumptionCompletionTimeout() throws Exception {
final int completionTimeout = 2000;
final int completionSize = -1; // timeout-based only
@@ -186,7 +181,7 @@ public class SjmsBatchConsumerTest extends CamelTestSupport {
* Checks whether multiple consumer endpoints can operate in parallel.
*/
@Test
- public void testConsumption_multipleConsumerEndpoints() throws Exception {
+ public void testConsumptionMultipleConsumerEndpoints() throws Exception {
final int completionTimeout = 2000;
final int completionSize = 5;
@@ -196,10 +191,10 @@ public class SjmsBatchConsumerTest extends CamelTestSupport {
from("direct:in")
.split().body()
- .multicast()
- .toF("sjms:%s", queueName + "A")
- .toF("sjms:%s", queueName + "B")
- .end();
+ .multicast()
+ .toF("sjms:%s", queueName + "A")
+ .toF("sjms:%s", queueName + "B")
+ .end();
fromF("sjms-batch:%s?completionTimeout=%s&completionSize=%s&aggregationStrategy=#testStrategy",
queueName + "A", completionTimeout, completionSize).routeId("batchConsumerA")
@@ -229,7 +224,7 @@ public class SjmsBatchConsumerTest extends CamelTestSupport {
}
@Test
- public void testConsumption_rollback() throws Exception {
+ public void testConsumptionRollback() throws Exception {
final int completionTimeout = 2000;
final int completionSize = 5;
http://git-wip-us.apache.org/repos/asf/camel/blob/832a99c5/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/SjmsBatchEndpointTest.java
----------------------------------------------------------------------
diff --git a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/SjmsBatchEndpointTest.java b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/SjmsBatchEndpointTest.java
index c97d0dd..45fe324 100644
--- a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/SjmsBatchEndpointTest.java
+++ b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/batch/SjmsBatchEndpointTest.java
@@ -30,9 +30,6 @@ import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
-/**
- * @author jkorab
- */
public class SjmsBatchEndpointTest extends CamelTestSupport {
// Create one embedded broker instance for the entire test, as we aren't actually
@@ -92,19 +89,19 @@ public class SjmsBatchEndpointTest extends CamelTestSupport {
}
@Test(expected = IllegalArgumentException.class)
- public void testConsumer_negativePollDuration() throws Exception {
+ public void testConsumerNegativePollDuration() throws Exception {
context.addRoutes(new RouteBuilder() {
@Override
public void configure() throws Exception {
from("sjms-batch:in?aggregationStrategy=#aggStrategy&pollDuration=-1")
- .to("mock:out");
+ .to("mock:out");
}
});
context.start();
}
@Test(expected = IllegalArgumentException.class)
- public void testConsumer_negativeConsumerCount() throws Exception {
+ public void testConsumerNegativeConsumerCount() throws Exception {
context.addRoutes(new RouteBuilder() {
@Override
public void configure() throws Exception {
@@ -116,7 +113,7 @@ public class SjmsBatchEndpointTest extends CamelTestSupport {
}
@Test(expected = FailedToCreateRouteException.class)
- public void testConsumer_topic() throws Exception {
+ public void testConsumerTopic() throws Exception {
context.addRoutes(new RouteBuilder() {
@Override
public void configure() throws Exception {
http://git-wip-us.apache.org/repos/asf/camel/blob/832a99c5/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/jms/DestinationNameParserTest.java
----------------------------------------------------------------------
diff --git a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/jms/DestinationNameParserTest.java b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/jms/DestinationNameParserTest.java
index 3fb4968..7f25239 100644
--- a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/jms/DestinationNameParserTest.java
+++ b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/jms/DestinationNameParserTest.java
@@ -34,7 +34,7 @@ public class DestinationNameParserTest {
}
@Test(expected = IllegalArgumentException.class)
- public void testIsTopic_nullDestinationName() throws Exception {
+ public void testIsTopicNullDestinationName() throws Exception {
DestinationNameParser parser = new DestinationNameParser();
parser.isTopic(null);
}
@@ -48,7 +48,7 @@ public class DestinationNameParserTest {
}
@Test(expected = IllegalArgumentException.class)
- public void testGetShortName_nullDestinationName() throws Exception {
+ public void testGetShortNameNullDestinationName() throws Exception {
DestinationNameParser parser = new DestinationNameParser();
parser.getShortName(null);
}
http://git-wip-us.apache.org/repos/asf/camel/blob/832a99c5/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/producer/QueueProducerQoSTest.java
----------------------------------------------------------------------
diff --git a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/producer/QueueProducerQoSTest.java b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/producer/QueueProducerQoSTest.java
index beef0a9..2a925f0 100644
--- a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/producer/QueueProducerQoSTest.java
+++ b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/producer/QueueProducerQoSTest.java
@@ -33,7 +33,7 @@ public class QueueProducerQoSTest extends JmsTestSupport {
private static final String TEST_INOUT_DESTINATION_NAME = "queue.producer.test.qos.inout";
private static final String EXPIRED_MESSAGE_ROUTE_ID = "expiredAdvisoryRoute";
- public static final String MOCK_EXPIRED_ADVISORY = "mock:expiredAdvisory";
+ private static final String MOCK_EXPIRED_ADVISORY = "mock:expiredAdvisory";
@EndpointInject(uri = MOCK_EXPIRED_ADVISORY)
MockEndpoint mockExpiredAdvisory;