Author: davsclaus
Date: Thu Aug 25 05:25:23 2011
New Revision: 1161388
URL: http://svn.apache.org/viewvc?rev=1161388&view=rev
Log:
CAMEL-4227: Added blockWhenFull option to seda component. Thanks to Mathieu
Lalonde for the patch. Marked CollectionProducer as @deprecated.
Added:
camel/trunk/camel-core/src/test/java/org/apache/camel/component/seda/SedaBlockWhenFullTest.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/CollectionProducer.java
camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/SedaEndpoint.java
camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/SedaProducer.java
camel/trunk/camel-core/src/test/java/org/apache/camel/component/seda/CollectionProducerTest.java
camel/trunk/camel-core/src/test/java/org/apache/camel/component/seda/SedaConfigureTest.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/CollectionProducer.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/CollectionProducer.java?rev=1161388&r1=1161387&r2=1161388&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/CollectionProducer.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/CollectionProducer.java
Thu Aug 25 05:25:23 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/trunk/camel-core/src/main/java/org/apache/camel/component/seda/SedaEndpoint.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/SedaEndpoint.java?rev=1161388&r1=1161387&r2=1161388&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/SedaEndpoint.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/SedaEndpoint.java
Thu Aug 25 05:25:23 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/trunk/camel-core/src/main/java/org/apache/camel/component/seda/SedaProducer.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/SedaProducer.java?rev=1161388&r1=1161387&r2=1161388&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/SedaProducer.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/SedaProducer.java
Thu Aug 25 05:25:23 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;
@@ -24,22 +25,36 @@ import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangeTimedOutException;
import org.apache.camel.WaitForTaskToComplete;
+import org.apache.camel.impl.DefaultAsyncProducer;
import org.apache.camel.support.SynchronizationAdapter;
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/trunk/camel-core/src/test/java/org/apache/camel/component/seda/CollectionProducerTest.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/component/seda/CollectionProducerTest.java?rev=1161388&r1=1161387&r2=1161388&view=diff
==============================================================================
---
camel/trunk/camel-core/src/test/java/org/apache/camel/component/seda/CollectionProducerTest.java
(original)
+++
camel/trunk/camel-core/src/test/java/org/apache/camel/component/seda/CollectionProducerTest.java
Thu Aug 25 05:25:23 2011
@@ -28,6 +28,7 @@ import org.apache.camel.impl.DefaultExch
/**
* @version
*/
+@Deprecated
public class CollectionProducerTest extends ContextTestSupport {
private static class MyProducer extends CollectionProducer {
Added:
camel/trunk/camel-core/src/test/java/org/apache/camel/component/seda/SedaBlockWhenFullTest.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/component/seda/SedaBlockWhenFullTest.java?rev=1161388&view=auto
==============================================================================
---
camel/trunk/camel-core/src/test/java/org/apache/camel/component/seda/SedaBlockWhenFullTest.java
(added)
+++
camel/trunk/camel-core/src/test/java/org/apache/camel/component/seda/SedaBlockWhenFullTest.java
Thu Aug 25 05:25:23 2011
@@ -0,0 +1,82 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.seda;
+
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.builder.RouteBuilder;
+
+/**
+ * Tests that a Seda producer supports the blockWhenFull option by blocking
+ * when a message is sent while the queue is full.
+ */
+public class SedaBlockWhenFullTest extends ContextTestSupport {
+ private static final int QUEUE_SIZE = 1;
+ private static final int DELAY = 10;
+ private static final String MOCK_URI = "mock:blockWhenFullOutput";
+ private static final String SIZE_PARAM = "?size=%d";
+ private static final String BLOCK_WHEN_FULL_URI = "seda:blockingFoo" +
String.format(SIZE_PARAM, QUEUE_SIZE) + "&blockWhenFull=true";
+ private static final String DEFAULT_URI = "seda:foo" +
String.format(SIZE_PARAM, QUEUE_SIZE);
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from(BLOCK_WHEN_FULL_URI).delay(DELAY).to(MOCK_URI);
+
+ // use same delay as above on purpose
+ from(DEFAULT_URI).delay(DELAY).to("mock:whatever");
+ }
+ };
+ }
+
+ public void testSedaDefaultWhenFull() throws Exception {
+ try {
+ SedaEndpoint seda = context.getEndpoint(DEFAULT_URI,
SedaEndpoint.class);
+ assertFalse("Seda Endpoint is not setting the correct default
(should be false) for \"blockWhenFull\"", seda.isBlockWhenFull());
+
+ sendTwoOverCapacity(DEFAULT_URI, QUEUE_SIZE);
+
+ fail("The route didn't fill the queue beyond capacity: test class
isn't working as intended");
+ } catch (Exception e) {
+ assertIsInstanceOf(IllegalStateException.class, e.getCause());
+ }
+ }
+
+ public void testSedaBlockingWhenFull() throws Exception {
+ getMockEndpoint(MOCK_URI).setExpectedMessageCount(QUEUE_SIZE + 2);
+
+ SedaEndpoint seda = context.getEndpoint(BLOCK_WHEN_FULL_URI,
SedaEndpoint.class);
+ assertEquals(QUEUE_SIZE, seda.getQueue().remainingCapacity());
+
+ sendTwoOverCapacity(BLOCK_WHEN_FULL_URI, QUEUE_SIZE);
+ assertMockEndpointsSatisfied();
+ }
+
+ /**
+ * This method make sure that we hit the limit by sending two msg over the
+ * given capacity which allows the delayer to kick in, leaving the 2nd msg
+ * in the queue, blocking/throwing on the third one.
+ */
+ private void sendTwoOverCapacity(String uri, int capacity) {
+ for (int i = 0; i < (capacity + 2); i++) {
+ template.sendBody(uri, "Message " + i);
+ }
+ }
+
+}
Modified:
camel/trunk/camel-core/src/test/java/org/apache/camel/component/seda/SedaConfigureTest.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/component/seda/SedaConfigureTest.java?rev=1161388&r1=1161387&r2=1161388&view=diff
==============================================================================
---
camel/trunk/camel-core/src/test/java/org/apache/camel/component/seda/SedaConfigureTest.java
(original)
+++
camel/trunk/camel-core/src/test/java/org/apache/camel/component/seda/SedaConfigureTest.java
Thu Aug 25 05:25:23 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());
+ }
}
