This is an automated email from the ASF dual-hosted git repository.
lburgazzoli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/master by this push:
new 96fda7d CAMEL-12466: camel-seda : support configuring endpoint
defaults on component
96fda7d is described below
commit 96fda7db4ee017629d210702dbcd8b8d4ec77e8e
Author: lburgazzoli <lburgazz...@gmail.com>
AuthorDate: Mon Apr 23 11:17:57 2018 +0200
CAMEL-12466: camel-seda : support configuring endpoint defaults on component
---
camel-core/src/main/docs/seda-component.adoc | 3 +-
camel-core/src/main/docs/stub-component.adoc | 3 +-
camel-core/src/main/docs/vm-component.adoc | 3 +-
.../apache/camel/component/seda/SedaComponent.java | 20 ++++
.../seda/SedaDefaultBlockWhenFullTest.java | 117 +++++++++++++++++++++
.../springboot/SedaComponentConfiguration.java | 16 +++
.../springboot/StubComponentConfiguration.java | 16 +++
.../vm/springboot/VmComponentConfiguration.java | 16 +++
8 files changed, 191 insertions(+), 3 deletions(-)
diff --git a/camel-core/src/main/docs/seda-component.adoc
b/camel-core/src/main/docs/seda-component.adoc
index 03e942f..bb2ff0d 100644
--- a/camel-core/src/main/docs/seda-component.adoc
+++ b/camel-core/src/main/docs/seda-component.adoc
@@ -39,7 +39,7 @@ You can append query options to the URI in the following
format:
=== Options
// component options: START
-The SEDA component supports 4 options which are listed below.
+The SEDA component supports 5 options which are listed below.
@@ -49,6 +49,7 @@ The SEDA component supports 4 options which are listed below.
| *queueSize* (advanced) | Sets the default maximum capacity of the SEDA queue
(i.e., the number of messages it can hold). | | int
| *concurrentConsumers* (consumer) | Sets the default number of concurrent
threads processing exchanges. | 1 | int
| *defaultQueueFactory* (advanced) | Sets the default queue factory. | |
Exchange>
+| *defaultBlockWhenFull* (producer) | Whether a thread that sends messages to
a full SEDA queue will block until the queue's capacity is no longer exhausted.
By default, an exception will be thrown stating that the queue is full. By
enabling this option, the calling thread will instead block and wait until the
message can be accepted. | false | boolean
| *resolveProperty Placeholders* (advanced) | Whether the component should
resolve property placeholders on itself when starting. Only properties which
are of String type can use property placeholders. | true | boolean
|===
// component options: END
diff --git a/camel-core/src/main/docs/stub-component.adoc
b/camel-core/src/main/docs/stub-component.adoc
index 32d5a7e..9923274 100644
--- a/camel-core/src/main/docs/stub-component.adoc
+++ b/camel-core/src/main/docs/stub-component.adoc
@@ -29,7 +29,7 @@ Where *`someUri`* can be any URI with any query parameters.
=== Options
// component options: START
-The Stub component supports 4 options which are listed below.
+The Stub component supports 5 options which are listed below.
@@ -39,6 +39,7 @@ The Stub component supports 4 options which are listed below.
| *queueSize* (advanced) | Sets the default maximum capacity of the SEDA queue
(i.e., the number of messages it can hold). | | int
| *concurrentConsumers* (consumer) | Sets the default number of concurrent
threads processing exchanges. | 1 | int
| *defaultQueueFactory* (advanced) | Sets the default queue factory. | |
Exchange>
+| *defaultBlockWhenFull* (producer) | Whether a thread that sends messages to
a full SEDA queue will block until the queue's capacity is no longer exhausted.
By default, an exception will be thrown stating that the queue is full. By
enabling this option, the calling thread will instead block and wait until the
message can be accepted. | false | boolean
| *resolveProperty Placeholders* (advanced) | Whether the component should
resolve property placeholders on itself when starting. Only properties which
are of String type can use property placeholders. | true | boolean
|===
// component options: END
diff --git a/camel-core/src/main/docs/vm-component.adoc
b/camel-core/src/main/docs/vm-component.adoc
index d8e77a9..3a04368 100644
--- a/camel-core/src/main/docs/vm-component.adoc
+++ b/camel-core/src/main/docs/vm-component.adoc
@@ -59,7 +59,7 @@ from("vm:bar?concurrentConsumers=5").to("file://output");
=== Options
// component options: START
-The VM component supports 4 options which are listed below.
+The VM component supports 5 options which are listed below.
@@ -69,6 +69,7 @@ The VM component supports 4 options which are listed below.
| *queueSize* (advanced) | Sets the default maximum capacity of the SEDA queue
(i.e., the number of messages it can hold). | | int
| *concurrentConsumers* (consumer) | Sets the default number of concurrent
threads processing exchanges. | 1 | int
| *defaultQueueFactory* (advanced) | Sets the default queue factory. | |
Exchange>
+| *defaultBlockWhenFull* (producer) | Whether a thread that sends messages to
a full SEDA queue will block until the queue's capacity is no longer exhausted.
By default, an exception will be thrown stating that the queue is full. By
enabling this option, the calling thread will instead block and wait until the
message can be accepted. | false | boolean
| *resolveProperty Placeholders* (advanced) | Whether the component should
resolve property placeholders on itself when starting. Only properties which
are of String type can use property placeholders. | true | boolean
|===
// component options: END
diff --git
a/camel-core/src/main/java/org/apache/camel/component/seda/SedaComponent.java
b/camel-core/src/main/java/org/apache/camel/component/seda/SedaComponent.java
index 7350150..fdb52ef 100644
---
a/camel-core/src/main/java/org/apache/camel/component/seda/SedaComponent.java
+++
b/camel-core/src/main/java/org/apache/camel/component/seda/SedaComponent.java
@@ -43,6 +43,8 @@ public class SedaComponent extends UriEndpointComponent {
protected int queueSize;
@Metadata(label = "advanced")
protected BlockingQueueFactory<Exchange> defaultQueueFactory = new
LinkedBlockingQueueFactory<>();
+ @Metadata(label = "producer")
+ private boolean defaultBlockWhenFull;
private final Map<String, QueueReference> queues = new HashMap<>();
@@ -87,6 +89,19 @@ public class SedaComponent extends UriEndpointComponent {
this.defaultQueueFactory = defaultQueueFactory;
}
+ public boolean isDefaultBlockWhenFull() {
+ return defaultBlockWhenFull;
+ }
+
+ /**
+ * Whether a thread that sends messages to a full SEDA queue will block
until the queue's capacity is no longer exhausted.
+ * By default, an exception will be thrown stating that the queue is full.
+ * By enabling this option, the calling thread will instead block and wait
until the message can be accepted.
+ */
+ public void setDefaultBlockWhenFull(boolean defaultBlockWhenFull) {
+ this.defaultBlockWhenFull = defaultBlockWhenFull;
+ }
+
/**
* @deprecated use
*/
@@ -189,6 +204,11 @@ public class SedaComponent extends UriEndpointComponent {
} else {
answer = createEndpoint(uri, this, queue, consumers);
}
+
+ // if blockWhenFull is set on endpoint, defaultBlockWhenFull is
ignored.
+ boolean blockWhenFull = getAndRemoveParameter(parameters,
"blockWhenFull", boolean.class, defaultBlockWhenFull);
+
+ answer.setBlockWhenFull(blockWhenFull);
answer.configureProperties(parameters);
answer.setConcurrentConsumers(consumers);
answer.setLimitConcurrentConsumers(limitConcurrentConsumers);
diff --git
a/camel-core/src/test/java/org/apache/camel/component/seda/SedaDefaultBlockWhenFullTest.java
b/camel-core/src/test/java/org/apache/camel/component/seda/SedaDefaultBlockWhenFullTest.java
new file mode 100644
index 0000000..c23e992
--- /dev/null
+++
b/camel-core/src/test/java/org/apache/camel/component/seda/SedaDefaultBlockWhenFullTest.java
@@ -0,0 +1,117 @@
+/*
+ * 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;
+import org.apache.camel.impl.JndiRegistry;
+
+/**
+ * Tests that a Seda component properly set blockWhenFull on endpoints.
+ */
+public class SedaDefaultBlockWhenFullTest extends ContextTestSupport {
+ private static final int QUEUE_SIZE = 1;
+ private static final int DELAY = 10;
+ private static final int DELAY_LONG = 100;
+ 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) + "&timeout=0";
+ private static final String DEFAULT_URI = "seda:foo" +
String.format(SIZE_PARAM, QUEUE_SIZE) + "&blockWhenFull=false&timeout=0";
+
+
+ @Override
+ protected JndiRegistry createRegistry() throws Exception {
+ SedaComponent component = new SedaComponent();
+ component.setDefaultBlockWhenFull(true);
+
+ JndiRegistry registry = super.createRegistry();
+ registry.bind("seda", component);
+
+ return registry;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ fromF(BLOCK_WHEN_FULL_URI).delay(DELAY_LONG).to(MOCK_URI);
+
+ // use same delay as above on purpose
+ from(DEFAULT_URI).delay(DELAY).to("mock:whatever");
+ }
+ };
+ }
+
+ public void testSedaEndpoints() {
+ assertFalse(context.getEndpoint(DEFAULT_URI,
SedaEndpoint.class).isBlockWhenFull());
+ assertTrue(context.getEndpoint(BLOCK_WHEN_FULL_URI,
SedaEndpoint.class).isBlockWhenFull());
+ }
+
+ 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();
+ }
+
+ public void testAsyncSedaBlockingWhenFull() throws Exception {
+ getMockEndpoint(MOCK_URI).setExpectedMessageCount(QUEUE_SIZE + 1);
+ getMockEndpoint(MOCK_URI).setResultWaitTime(DELAY_LONG * 3);
+
+ SedaEndpoint seda = context.getEndpoint(BLOCK_WHEN_FULL_URI,
SedaEndpoint.class);
+ assertEquals(QUEUE_SIZE, seda.getQueue().remainingCapacity());
+
+ asyncSendTwoOverCapacity(BLOCK_WHEN_FULL_URI, QUEUE_SIZE + 4);
+ 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);
+ }
+ }
+
+ private void asyncSendTwoOverCapacity(String uri, int capacity) {
+ for (int i = 0; i < (capacity + 2); i++) {
+ template.asyncSendBody(uri, "Message " + i);
+ }
+ }
+
+}
diff --git
a/platforms/spring-boot/components-starter/camel-core-starter/src/main/java/org/apache/camel/component/seda/springboot/SedaComponentConfiguration.java
b/platforms/spring-boot/components-starter/camel-core-starter/src/main/java/org/apache/camel/component/seda/springboot/SedaComponentConfiguration.java
index 23057bf..286d693 100644
---
a/platforms/spring-boot/components-starter/camel-core-starter/src/main/java/org/apache/camel/component/seda/springboot/SedaComponentConfiguration.java
+++
b/platforms/spring-boot/components-starter/camel-core-starter/src/main/java/org/apache/camel/component/seda/springboot/SedaComponentConfiguration.java
@@ -50,6 +50,14 @@ public class SedaComponentConfiguration
@NestedConfigurationProperty
private BlockingQueueFactory<Exchange> defaultQueueFactory;
/**
+ * Whether a thread that sends messages to a full SEDA queue will block
+ * until the queue's capacity is no longer exhausted. By default, an
+ * exception will be thrown stating that the queue is full. By enabling
this
+ * option, the calling thread will instead block and wait until the message
+ * can be accepted.
+ */
+ private Boolean defaultBlockWhenFull = false;
+ /**
* Whether the component should resolve property placeholders on itself
when
* starting. Only properties which are of String type can use property
* placeholders.
@@ -81,6 +89,14 @@ public class SedaComponentConfiguration
this.defaultQueueFactory = defaultQueueFactory;
}
+ public Boolean getDefaultBlockWhenFull() {
+ return defaultBlockWhenFull;
+ }
+
+ public void setDefaultBlockWhenFull(Boolean defaultBlockWhenFull) {
+ this.defaultBlockWhenFull = defaultBlockWhenFull;
+ }
+
public Boolean getResolvePropertyPlaceholders() {
return resolvePropertyPlaceholders;
}
diff --git
a/platforms/spring-boot/components-starter/camel-core-starter/src/main/java/org/apache/camel/component/stub/springboot/StubComponentConfiguration.java
b/platforms/spring-boot/components-starter/camel-core-starter/src/main/java/org/apache/camel/component/stub/springboot/StubComponentConfiguration.java
index 558c877..5efee17 100644
---
a/platforms/spring-boot/components-starter/camel-core-starter/src/main/java/org/apache/camel/component/stub/springboot/StubComponentConfiguration.java
+++
b/platforms/spring-boot/components-starter/camel-core-starter/src/main/java/org/apache/camel/component/stub/springboot/StubComponentConfiguration.java
@@ -50,6 +50,14 @@ public class StubComponentConfiguration
@NestedConfigurationProperty
private BlockingQueueFactory<Exchange> defaultQueueFactory;
/**
+ * Whether a thread that sends messages to a full SEDA queue will block
+ * until the queue's capacity is no longer exhausted. By default, an
+ * exception will be thrown stating that the queue is full. By enabling
this
+ * option, the calling thread will instead block and wait until the message
+ * can be accepted.
+ */
+ private Boolean defaultBlockWhenFull = false;
+ /**
* Whether the component should resolve property placeholders on itself
when
* starting. Only properties which are of String type can use property
* placeholders.
@@ -81,6 +89,14 @@ public class StubComponentConfiguration
this.defaultQueueFactory = defaultQueueFactory;
}
+ public Boolean getDefaultBlockWhenFull() {
+ return defaultBlockWhenFull;
+ }
+
+ public void setDefaultBlockWhenFull(Boolean defaultBlockWhenFull) {
+ this.defaultBlockWhenFull = defaultBlockWhenFull;
+ }
+
public Boolean getResolvePropertyPlaceholders() {
return resolvePropertyPlaceholders;
}
diff --git
a/platforms/spring-boot/components-starter/camel-core-starter/src/main/java/org/apache/camel/component/vm/springboot/VmComponentConfiguration.java
b/platforms/spring-boot/components-starter/camel-core-starter/src/main/java/org/apache/camel/component/vm/springboot/VmComponentConfiguration.java
index 26b42cc..b497827 100644
---
a/platforms/spring-boot/components-starter/camel-core-starter/src/main/java/org/apache/camel/component/vm/springboot/VmComponentConfiguration.java
+++
b/platforms/spring-boot/components-starter/camel-core-starter/src/main/java/org/apache/camel/component/vm/springboot/VmComponentConfiguration.java
@@ -50,6 +50,14 @@ public class VmComponentConfiguration
@NestedConfigurationProperty
private BlockingQueueFactory<Exchange> defaultQueueFactory;
/**
+ * Whether a thread that sends messages to a full SEDA queue will block
+ * until the queue's capacity is no longer exhausted. By default, an
+ * exception will be thrown stating that the queue is full. By enabling
this
+ * option, the calling thread will instead block and wait until the message
+ * can be accepted.
+ */
+ private Boolean defaultBlockWhenFull = false;
+ /**
* Whether the component should resolve property placeholders on itself
when
* starting. Only properties which are of String type can use property
* placeholders.
@@ -81,6 +89,14 @@ public class VmComponentConfiguration
this.defaultQueueFactory = defaultQueueFactory;
}
+ public Boolean getDefaultBlockWhenFull() {
+ return defaultBlockWhenFull;
+ }
+
+ public void setDefaultBlockWhenFull(Boolean defaultBlockWhenFull) {
+ this.defaultBlockWhenFull = defaultBlockWhenFull;
+ }
+
public Boolean getResolvePropertyPlaceholders() {
return resolvePropertyPlaceholders;
}
--
To stop receiving notification emails like this one, please contact
lburgazz...@apache.org.
