camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From acosent...@apache.org
Subject camel git commit: CAMEL-10489: Camel-Nats: Add Flush option with timeout
Date Thu, 17 Nov 2016 09:59:47 GMT
Repository: camel
Updated Branches:
  refs/heads/master 06048c832 -> cb5617e0d


CAMEL-10489: Camel-Nats: Add Flush option with timeout


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

Branch: refs/heads/master
Commit: cb5617e0d94288b8597ff23cbe69ddd72d86c999
Parents: 06048c8
Author: Andrea Cosentino <ancosen@gmail.com>
Authored: Thu Nov 17 10:59:11 2016 +0100
Committer: Andrea Cosentino <ancosen@gmail.com>
Committed: Thu Nov 17 10:59:11 2016 +0100

----------------------------------------------------------------------
 .../src/main/docs/nats-component.adoc           |  4 ++-
 .../camel/component/nats/NatsConfiguration.java | 26 ++++++++++++++++++++
 .../camel/component/nats/NatsConsumer.java      |  6 +++--
 .../camel/component/nats/NatsProducer.java      |  4 +++
 .../camel/component/nats/NatsConsumerTest.java  |  4 +--
 5 files changed, 39 insertions(+), 5 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/cb5617e0/components/camel-nats/src/main/docs/nats-component.adoc
----------------------------------------------------------------------
diff --git a/components/camel-nats/src/main/docs/nats-component.adoc b/components/camel-nats/src/main/docs/nats-component.adoc
index 2c0839c..3965ef7 100644
--- a/components/camel-nats/src/main/docs/nats-component.adoc
+++ b/components/camel-nats/src/main/docs/nats-component.adoc
@@ -44,13 +44,15 @@ The Nats component has no options.
 
 
 // endpoint options: START
-The Nats component supports 21 endpoint options which are listed below:
+The Nats component supports 23 endpoint options which are listed below:
 
 {% raw %}
 [width="100%",cols="2,1,1m,1m,5",options="header"]
 |=======================================================================
 | Name | Group | Default | Java Type | Description
 | servers | common |  | String | *Required* URLs to one or more NAT servers. Use comma to
separate URLs when specifying multiple servers.
+| flushConnection | common | false | boolean | Define if we want to flush connection or not
+| flushTimeout | common | 1000 | int | Set the flush timeout
 | maxReconnectAttempts | common | 3 | int | Max reconnection attempts
 | noRandomizeServers | common | false | boolean | Whether or not randomizing the order of
servers for the connection attempts
 | pedantic | common | false | boolean | Whether or not running in pedantic mode (this affects
performace)

http://git-wip-us.apache.org/repos/asf/camel/blob/cb5617e0/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConfiguration.java
----------------------------------------------------------------------
diff --git a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConfiguration.java
b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConfiguration.java
index 641b24b..dab069e 100644
--- a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConfiguration.java
+++ b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConfiguration.java
@@ -57,6 +57,10 @@ public class NatsConfiguration {
     private String maxMessages;
     @UriParam(label = "consumer", defaultValue = "10")
     private int poolSize = 10;
+    @UriParam(label = "common", defaultValue = "false")
+    private boolean flushConnection;
+    @UriParam(label = "common", defaultValue = "1000")
+    private int flushTimeout = 1000;
     @UriParam(label = "security")
     private boolean secure;
     @UriParam(label = "security")
@@ -219,6 +223,28 @@ public class NatsConfiguration {
         this.poolSize = poolSize;
     }
 
+    public boolean isFlushConnection() {
+        return flushConnection;
+    }
+
+    /**
+     * Define if we want to flush connection or not
+     */
+    public void setFlushConnection(boolean flushConnection) {
+        this.flushConnection = flushConnection;
+    }
+
+    public int getFlushTimeout() {
+        return flushTimeout;
+    }
+
+    /**
+     * Set the flush timeout
+     */
+    public void setFlushTimeout(int flushTimeout) {
+        this.flushTimeout = flushTimeout;
+    }
+
     /**
      * Set secure option indicating TLS is required
      */

http://git-wip-us.apache.org/repos/asf/camel/blob/cb5617e0/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java
b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java
index 37e5171..95bc0e3 100644
--- a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java
+++ b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java
@@ -73,8 +73,10 @@ public class NatsConsumer extends DefaultConsumer {
     protected void doStop() throws Exception {
         super.doStop();
 
-        LOG.debug("Flushing Messages before stopping");
-        connection.flush();
+        if (getEndpoint().getNatsConfiguration().isFlushConnection()) {
+            LOG.debug("Flushing Messages before stopping");
+            connection.flush(getEndpoint().getNatsConfiguration().getFlushTimeout());
+        }
         
         try {
             sid.unsubscribe();

http://git-wip-us.apache.org/repos/asf/camel/blob/cb5617e0/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsProducer.java
b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsProducer.java
index 0efafe2..1be13e3 100644
--- a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsProducer.java
+++ b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsProducer.java
@@ -78,6 +78,10 @@ public class NatsProducer extends DefaultProducer {
         
         LOG.debug("Closing Nats Connection");
         if (connection != null && !connection.isClosed()) {
+            if (getEndpoint().getNatsConfiguration().isFlushConnection()) {
+                LOG.debug("Flushing Nats Connection");
+                connection.flush(getEndpoint().getNatsConfiguration().getFlushTimeout());
+            }
             connection.close();
         }
     }

http://git-wip-us.apache.org/repos/asf/camel/blob/cb5617e0/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerTest.java
b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerTest.java
index ca63048..24d4877 100644
--- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerTest.java
+++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerTest.java
@@ -45,8 +45,8 @@ public class NatsConsumerTest extends CamelTestSupport {
         return new RouteBuilder() {
             @Override
             public void configure() throws Exception {
-                from("direct:send").to("nats://localhost:4222?topic=test");
-                from("nats://localhost:4222?topic=test").to(mockResultEndpoint);
+                from("direct:send").to("nats://localhost:4222?topic=test&flushConnection=true");
+                from("nats://localhost:4222?topic=test&flushConnection=true").to(mockResultEndpoint);
             }
         };
     }


Mime
View raw message