camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject [1/3] camel git commit: CAMEL-5911: seda/vm add discardIfNoConsumers option
Date Fri, 10 Jul 2015 14:15:28 GMT
Repository: camel
Updated Branches:
  refs/heads/master 821bddf58 -> 7747c4aec


CAMEL-5911: seda/vm add discardIfNoConsumers option


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/7bfdef25
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/7bfdef25
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/7bfdef25

Branch: refs/heads/master
Commit: 7bfdef2538ccecb46c1769c31afe88dd17249b2b
Parents: 821bddf
Author: Claus Ibsen <davsclaus@apache.org>
Authored: Fri Jul 10 15:45:33 2015 +0200
Committer: Claus Ibsen <davsclaus@apache.org>
Committed: Fri Jul 10 16:21:54 2015 +0200

----------------------------------------------------------------------
 .../camel/component/seda/SedaComponent.java     |  3 ++
 .../camel/component/seda/SedaEndpoint.java      | 35 +++++++++++++-
 .../camel/component/seda/SedaProducer.java      | 11 ++++-
 .../seda/SedaDiscardIfNoConsumerTest.java       | 49 ++++++++++++++++++++
 .../component/vm/VmDiscardIfNoConsumerTest.java | 49 ++++++++++++++++++++
 5 files changed, 144 insertions(+), 3 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/7bfdef25/camel-core/src/main/java/org/apache/camel/component/seda/SedaComponent.java
----------------------------------------------------------------------
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 c8084ce..87a0843 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
@@ -171,6 +171,7 @@ public class SedaComponent extends UriEndpointComponent {
             throw new IllegalArgumentException("The limitConcurrentConsumers flag in set
to true. ConcurrentConsumers cannot be set at a value greater than "
                     + maxConcurrentConsumers + " was " + consumers);
         }
+
         // Resolve queue reference
         BlockingQueue<Exchange> queue = resolveAndRemoveReferenceParameter(parameters,
"queue", BlockingQueue.class);
         SedaEndpoint answer;
@@ -183,6 +184,8 @@ public class SedaComponent extends UriEndpointComponent {
             answer = createEndpoint(uri, this, queue, consumers);
         }
         answer.configureProperties(parameters);
+        answer.setConcurrentConsumers(consumers);
+        answer.setLimitConcurrentConsumers(limitConcurrentConsumers);
         return answer;
     }
 

http://git-wip-us.apache.org/repos/asf/camel/blob/7bfdef25/camel-core/src/main/java/org/apache/camel/component/seda/SedaEndpoint.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/component/seda/SedaEndpoint.java b/camel-core/src/main/java/org/apache/camel/component/seda/SedaEndpoint.java
index e8090a4..1b82935 100644
--- a/camel-core/src/main/java/org/apache/camel/component/seda/SedaEndpoint.java
+++ b/camel-core/src/main/java/org/apache/camel/component/seda/SedaEndpoint.java
@@ -73,6 +73,8 @@ public class SedaEndpoint extends DefaultEndpoint implements BrowsableEndpoint,
 
     @UriParam(label = "consumer", defaultValue = "1")
     private int concurrentConsumers = 1;
+    @UriParam(label = "consumer", defaultValue = "true")
+    private boolean limitConcurrentConsumers = true;
     @UriParam(label = "consumer")
     private boolean multipleConsumers;
     @UriParam(label = "consumer")
@@ -88,6 +90,8 @@ public class SedaEndpoint extends DefaultEndpoint implements BrowsableEndpoint,
     private boolean blockWhenFull;
     @UriParam(label = "producer")
     private boolean failIfNoConsumers;
+    @UriParam(label = "producer")
+    private boolean discardIfNoConsumers;
 
     private BlockingQueueFactory<Exchange> queueFactory;
 
@@ -292,6 +296,19 @@ public class SedaEndpoint extends DefaultEndpoint implements BrowsableEndpoint,
         return concurrentConsumers;
     }
 
+    @ManagedAttribute
+    public boolean isLimitConcurrentConsumers() {
+        return limitConcurrentConsumers;
+    }
+
+    /**
+     * Whether to limit the number of concurrentConsumers to the maximum of 500.
+     * By default, an exception will be thrown if an endpoint is configured with a greater
number. You can disable that check by turning this option off.
+     */
+    public void setLimitConcurrentConsumers(boolean limitConcurrentConsumers) {
+        this.limitConcurrentConsumers = limitConcurrentConsumers;
+    }
+
     public WaitForTaskToComplete getWaitForTaskToComplete() {
         return waitForTaskToComplete;
     }
@@ -326,13 +343,29 @@ public class SedaEndpoint extends DefaultEndpoint implements BrowsableEndpoint,
     }
 
     /**
-     * Whether the producer should fail by throwing an exception, when sending to a SEDA
queue with no active consumers.
+     * Whether the producer should fail by throwing an exception, when sending to a queue
with no active consumers.
+     * <p/>
+     * Only one of the options <tt>discardIfNoConsumers</tt> and <tt>failIfNoConsumers</tt>
can be enabled at the same time.
      */
     public void setFailIfNoConsumers(boolean failIfNoConsumers) {
         this.failIfNoConsumers = failIfNoConsumers;
     }
 
     @ManagedAttribute
+    public boolean isDiscardIfNoConsumers() {
+        return discardIfNoConsumers;
+    }
+
+    /**
+     * Whether the producer should discard the message (do not add the message to the queue),
when sending to a queue with no active consumers.
+     * <p/>
+     * Only one of the options <tt>discardIfNoConsumers</tt> and <tt>failIfNoConsumers</tt>
can be enabled at the same time.
+     */
+    public void setDiscardIfNoConsumers(boolean discardIfNoConsumers) {
+        this.discardIfNoConsumers = discardIfNoConsumers;
+    }
+
+    @ManagedAttribute
     public boolean isMultipleConsumers() {
         return multipleConsumers;
     }

http://git-wip-us.apache.org/repos/asf/camel/blob/7bfdef25/camel-core/src/main/java/org/apache/camel/component/seda/SedaProducer.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/component/seda/SedaProducer.java b/camel-core/src/main/java/org/apache/camel/component/seda/SedaProducer.java
index 56b83f7..a87ddf3 100644
--- a/camel-core/src/main/java/org/apache/camel/component/seda/SedaProducer.java
+++ b/camel-core/src/main/java/org/apache/camel/component/seda/SedaProducer.java
@@ -216,9 +216,16 @@ public class SedaProducer extends DefaultAsyncProducer {
             throw new SedaConsumerNotAvailableException("No queue available on endpoint:
" + endpoint, exchange);
         }
 
-        if (endpoint.isFailIfNoConsumers() && !queueReference.hasConsumers()) {
-            throw new SedaConsumerNotAvailableException("No consumers available on endpoint:
" + endpoint, exchange);
+        boolean empty = !queueReference.hasConsumers();
+        if (empty) {
+            if (endpoint.isFailIfNoConsumers()) {
+                throw new SedaConsumerNotAvailableException("No consumers available on endpoint:
" + endpoint, exchange);
+            } else if (endpoint.isDiscardIfNoConsumers()) {
+                log.debug("Discard message as no active consumers on endpoint: " + endpoint);
+                return;
+            }
         }
+
         if (blockWhenFull) {
             try {
                 queue.put(exchange);

http://git-wip-us.apache.org/repos/asf/camel/blob/7bfdef25/camel-core/src/test/java/org/apache/camel/component/seda/SedaDiscardIfNoConsumerTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/component/seda/SedaDiscardIfNoConsumerTest.java
b/camel-core/src/test/java/org/apache/camel/component/seda/SedaDiscardIfNoConsumerTest.java
new file mode 100644
index 0000000..630abd4
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/component/seda/SedaDiscardIfNoConsumerTest.java
@@ -0,0 +1,49 @@
+/**
+ * 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;
+
+/**
+ * @version 
+ */
+public class SedaDiscardIfNoConsumerTest extends ContextTestSupport {
+
+    public void testDiscard() throws Exception {
+        SedaEndpoint bar = getMandatoryEndpoint("seda:bar", SedaEndpoint.class);
+        assertEquals(0, bar.getCurrentQueueSize());
+
+        getMockEndpoint("mock:result").expectedBodiesReceived("Hello World");
+
+        template.sendBody("direct:start", "Hello World");
+
+        assertMockEndpointsSatisfied();
+
+        assertEquals(0, bar.getCurrentQueueSize());
+    }
+
+    @Override
+    protected RouteBuilder createRouteBuilder() throws Exception {
+        return new RouteBuilder() {
+            @Override
+            public void configure() throws Exception {
+                from("direct:start").to("seda:bar?discardIfNoConsumers=true").to("mock:result");
+            }
+        };
+    }
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/7bfdef25/camel-core/src/test/java/org/apache/camel/component/vm/VmDiscardIfNoConsumerTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/component/vm/VmDiscardIfNoConsumerTest.java
b/camel-core/src/test/java/org/apache/camel/component/vm/VmDiscardIfNoConsumerTest.java
new file mode 100644
index 0000000..6581d73
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/component/vm/VmDiscardIfNoConsumerTest.java
@@ -0,0 +1,49 @@
+/**
+ * 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.vm;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.builder.RouteBuilder;
+
+/**
+ * @version 
+ */
+public class VmDiscardIfNoConsumerTest extends ContextTestSupport {
+
+    public void testDiscard() throws Exception {
+        VmEndpoint bar = getMandatoryEndpoint("vm:bar", VmEndpoint.class);
+        assertEquals(0, bar.getCurrentQueueSize());
+
+        getMockEndpoint("mock:result").expectedBodiesReceived("Hello World");
+
+        template.sendBody("direct:start", "Hello World");
+
+        assertMockEndpointsSatisfied();
+
+        assertEquals(0, bar.getCurrentQueueSize());
+    }
+
+    @Override
+    protected RouteBuilder createRouteBuilder() throws Exception {
+        return new RouteBuilder() {
+            @Override
+            public void configure() throws Exception {
+                from("direct:start").to("vm:bar?discardIfNoConsumers=true").to("mock:result");
+            }
+        };
+    }
+}


Mime
View raw message