You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by dk...@apache.org on 2011/09/20 00:45:45 UTC
svn commit: r1172871 - in /camel/branches/camel-2.8.x: ./
camel-core/src/main/java/org/apache/camel/component/seda/
camel-core/src/test/java/org/apache/camel/component/seda/
Author: dkulp
Date: Mon Sep 19 22:45:44 2011
New Revision: 1172871
URL: http://svn.apache.org/viewvc?rev=1172871&view=rev
Log:
Merged revisions 1161388 via svnmerge from
https://svn.apache.org/repos/asf/camel/trunk
........
r1161388 | davsclaus | 2011-08-25 01:25:23 -0400 (Thu, 25 Aug 2011) | 1 line
CAMEL-4227: Added blockWhenFull option to seda component. Thanks to Mathieu Lalonde for the patch. Marked CollectionProducer as @deprecated.
........
Added:
camel/branches/camel-2.8.x/camel-core/src/test/java/org/apache/camel/component/seda/SedaBlockWhenFullTest.java
- copied unchanged from r1161388, camel/trunk/camel-core/src/test/java/org/apache/camel/component/seda/SedaBlockWhenFullTest.java
Modified:
camel/branches/camel-2.8.x/ (props changed)
camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/component/seda/CollectionProducer.java
camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/component/seda/SedaEndpoint.java
camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/component/seda/SedaProducer.java
camel/branches/camel-2.8.x/camel-core/src/test/java/org/apache/camel/component/seda/CollectionProducerTest.java
camel/branches/camel-2.8.x/camel-core/src/test/java/org/apache/camel/component/seda/SedaConfigureTest.java
Propchange: camel/branches/camel-2.8.x/
------------------------------------------------------------------------------
Binary property 'svnmerge-integrated' - no diff available.
Modified: camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/component/seda/CollectionProducer.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/component/seda/CollectionProducer.java?rev=1172871&r1=1172870&r2=1172871&view=diff
==============================================================================
--- camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/component/seda/CollectionProducer.java (original)
+++ camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/component/seda/CollectionProducer.java Mon Sep 19 22:45:44 2011
@@ -26,8 +26,10 @@ import org.apache.camel.impl.DefaultAsyn
/**
* A simple {@link org.apache.camel.Producer} which just appends to a {@link Collection} the {@link Exchange} object.
*
+ * @deprecated will be removed in a future Camel release
* @version
*/
+@Deprecated
public class CollectionProducer extends DefaultAsyncProducer {
protected final Collection<Exchange> queue;
Modified: camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/component/seda/SedaEndpoint.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/component/seda/SedaEndpoint.java?rev=1172871&r1=1172870&r2=1172871&view=diff
==============================================================================
--- camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/component/seda/SedaEndpoint.java (original)
+++ camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/component/seda/SedaEndpoint.java Mon Sep 19 22:45:44 2011
@@ -41,8 +41,6 @@ import org.apache.camel.util.ServiceHelp
* An implementation of the <a
* href="http://camel.apache.org/queue.html">Queue components</a> for
* asynchronous SEDA exchanges on a {@link BlockingQueue} within a CamelContext
- *
- * @version
*/
public class SedaEndpoint extends DefaultEndpoint implements BrowsableEndpoint, MultipleConsumersSupport {
private volatile BlockingQueue<Exchange> queue;
@@ -56,6 +54,7 @@ public class SedaEndpoint extends Defaul
private final Set<SedaConsumer> consumers = new CopyOnWriteArraySet<SedaConsumer>();
private volatile MulticastProcessor consumerMulticastProcessor;
private volatile boolean multicastStarted;
+ private boolean blockWhenFull;
public SedaEndpoint() {
}
@@ -81,9 +80,9 @@ public class SedaEndpoint extends Defaul
this.size = queue.remainingCapacity();
this.concurrentConsumers = concurrentConsumers;
}
-
+
public Producer createProducer() throws Exception {
- return new SedaProducer(this, getQueue(), getWaitForTaskToComplete(), getTimeout());
+ return new SedaProducer(this, getQueue(), getWaitForTaskToComplete(), getTimeout(), isBlockWhenFull());
}
public Consumer createConsumer(Processor processor) throws Exception {
@@ -100,7 +99,7 @@ public class SedaEndpoint extends Defaul
}
return queue;
}
-
+
protected synchronized MulticastProcessor getConsumerMulticastProcessor() throws Exception {
if (!multicastStarted && consumerMulticastProcessor != null) {
// only start it on-demand to avoid starting it during stopping
@@ -109,7 +108,7 @@ public class SedaEndpoint extends Defaul
}
return consumerMulticastProcessor;
}
-
+
protected synchronized void updateMulticastProcessor() throws Exception {
if (consumerMulticastProcessor != null) {
ServiceHelper.stopService(consumerMulticastProcessor);
@@ -153,10 +152,18 @@ public class SedaEndpoint extends Defaul
this.size = size;
}
+ public void setBlockWhenFull(boolean blockWhenFull) {
+ this.blockWhenFull = blockWhenFull;
+ }
+
+ public boolean isBlockWhenFull() {
+ return blockWhenFull;
+ }
+
public void setConcurrentConsumers(int concurrentConsumers) {
this.concurrentConsumers = concurrentConsumers;
}
-
+
public int getConcurrentConsumers() {
return concurrentConsumers;
}
@@ -213,7 +220,7 @@ public class SedaEndpoint extends Defaul
public Set<SedaProducer> getProducers() {
return new HashSet<SedaProducer>(producers);
}
-
+
void onStarted(SedaProducer producer) {
producers.add(producer);
}
Modified: camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/component/seda/SedaProducer.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/component/seda/SedaProducer.java?rev=1172871&r1=1172870&r2=1172871&view=diff
==============================================================================
--- camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/component/seda/SedaProducer.java (original)
+++ camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/component/seda/SedaProducer.java Mon Sep 19 22:45:44 2011
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.seda;
+import java.util.Collection;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -25,21 +26,35 @@ import org.apache.camel.Exchange;
import org.apache.camel.ExchangeTimedOutException;
import org.apache.camel.WaitForTaskToComplete;
import org.apache.camel.impl.SynchronizationAdapter;
+import org.apache.camel.impl.DefaultAsyncProducer;
import org.apache.camel.util.ExchangeHelper;
/**
* @version
*/
-public class SedaProducer extends CollectionProducer {
+public class SedaProducer extends DefaultAsyncProducer {
+ protected final Collection<Exchange> queue;
private final SedaEndpoint endpoint;
private final WaitForTaskToComplete waitForTaskToComplete;
private final long timeout;
+ private final boolean blockWhenFull;
+ /**
+ * @deprecated use the other constructor
+ */
+ @Deprecated
public SedaProducer(SedaEndpoint endpoint, BlockingQueue<Exchange> queue, WaitForTaskToComplete waitForTaskToComplete, long timeout) {
- super(endpoint, queue);
+ this(endpoint, queue, waitForTaskToComplete, timeout, false);
+ }
+
+ public SedaProducer(SedaEndpoint endpoint, BlockingQueue<Exchange> queue, WaitForTaskToComplete waitForTaskToComplete,
+ long timeout, boolean blockWhenFull) {
+ super(endpoint);
+ this.queue = queue;
this.endpoint = endpoint;
this.waitForTaskToComplete = waitForTaskToComplete;
this.timeout = timeout;
+ this.blockWhenFull = blockWhenFull;
}
@Override
@@ -98,7 +113,7 @@ public class SedaProducer extends Collec
});
log.trace("Adding Exchange to queue: {}", copy);
- queue.add(copy);
+ addToQueue(copy);
if (timeout > 0) {
if (log.isTraceEnabled()) {
@@ -130,7 +145,7 @@ public class SedaProducer extends Collec
} else {
// no wait, eg its a InOnly then just add to queue and return
log.trace("Adding Exchange to queue: {}", copy);
- queue.add(copy);
+ addToQueue(copy);
}
// we use OnCompletion on the Exchange to callback and wait for the Exchange to be done
@@ -150,4 +165,25 @@ public class SedaProducer extends Collec
endpoint.onStopped(this);
super.doStop();
}
+
+ /**
+ * Strategy method for adding the exchange to the queue.
+ * <p>
+ * Will perform a blocking "put" if blockWhenFull is true, otherwise it will
+ * simply add which will throw exception if the queue is full
+ *
+ * @param exchange the exchange to add to the queue
+ */
+ protected void addToQueue(Exchange exchange) {
+ if (blockWhenFull) {
+ try {
+ ((BlockingQueue<Exchange>)queue).put(exchange);
+ } catch (InterruptedException e) {
+ // ignore
+ }
+ } else {
+ queue.add(exchange);
+ }
+ }
+
}
Modified: camel/branches/camel-2.8.x/camel-core/src/test/java/org/apache/camel/component/seda/CollectionProducerTest.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.8.x/camel-core/src/test/java/org/apache/camel/component/seda/CollectionProducerTest.java?rev=1172871&r1=1172870&r2=1172871&view=diff
==============================================================================
--- camel/branches/camel-2.8.x/camel-core/src/test/java/org/apache/camel/component/seda/CollectionProducerTest.java (original)
+++ camel/branches/camel-2.8.x/camel-core/src/test/java/org/apache/camel/component/seda/CollectionProducerTest.java Mon Sep 19 22:45:44 2011
@@ -28,6 +28,7 @@ import org.apache.camel.impl.DefaultExch
/**
* @version
*/
+@Deprecated
public class CollectionProducerTest extends ContextTestSupport {
private class MyProducer extends CollectionProducer {
Modified: camel/branches/camel-2.8.x/camel-core/src/test/java/org/apache/camel/component/seda/SedaConfigureTest.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.8.x/camel-core/src/test/java/org/apache/camel/component/seda/SedaConfigureTest.java?rev=1172871&r1=1172870&r2=1172871&view=diff
==============================================================================
--- camel/branches/camel-2.8.x/camel-core/src/test/java/org/apache/camel/component/seda/SedaConfigureTest.java (original)
+++ camel/branches/camel-2.8.x/camel-core/src/test/java/org/apache/camel/component/seda/SedaConfigureTest.java Mon Sep 19 22:45:44 2011
@@ -23,19 +23,34 @@ import org.apache.camel.ContextTestSuppo
import org.apache.camel.Exchange;
/**
- * @version
+ * @version
*/
public class SedaConfigureTest extends ContextTestSupport {
+ @SuppressWarnings("unchecked")
public void testBlockingQueueConfigured() throws Exception {
SedaEndpoint endpoint = resolveMandatoryEndpoint("seda:foo?size=2000", SedaEndpoint.class);
BlockingQueue<Exchange> queue = endpoint.getQueue();
- LinkedBlockingQueue blockingQueue = assertIsInstanceOf(LinkedBlockingQueue.class, queue);
+
+ LinkedBlockingQueue<Exchange> blockingQueue = assertIsInstanceOf(LinkedBlockingQueue.class, queue);
assertEquals("remainingCapacity", 2000, blockingQueue.remainingCapacity());
}
-
+
public void testConcurrentConsumersConfigured() {
SedaEndpoint endpoint = resolveMandatoryEndpoint("seda:foo?concurrentConsumers=5", SedaEndpoint.class);
assertEquals("concurrentConsumers", 5, endpoint.getConcurrentConsumers());
}
+
+ public void testBlockWhenFull() {
+ SedaEndpoint endpoint = resolveMandatoryEndpoint("seda:foo?size=2000&blockWhenFull=true", SedaEndpoint.class);
+ assertTrue("blockWhenFull", endpoint.isBlockWhenFull());
+ }
+
+ public void testDefaults() {
+ SedaEndpoint endpoint = resolveMandatoryEndpoint("seda:foo", SedaEndpoint.class);
+ assertFalse("blockWhenFull: wrong default", endpoint.isBlockWhenFull());
+ assertEquals("concurrentConsumers: wrong default", 1, endpoint.getConcurrentConsumers());
+ assertEquals("size (remainingCapacity): wrong default", Integer.MAX_VALUE, endpoint.getSize());
+ assertEquals("timeout: wrong default", 30000L, endpoint.getTimeout());
+ }
}