camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject camel git commit: Polished
Date Tue, 19 Jan 2016 09:11:45 GMT
Repository: camel
Updated Branches:
  refs/heads/master 4e0437c40 -> e50136b7d


Polished


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

Branch: refs/heads/master
Commit: e50136b7d9ed6f217346ec4a6b4f13e2fa17e52b
Parents: 4e0437c
Author: Claus Ibsen <davsclaus@apache.org>
Authored: Tue Jan 19 10:07:17 2016 +0100
Committer: Claus Ibsen <davsclaus@apache.org>
Committed: Tue Jan 19 10:08:35 2016 +0100

----------------------------------------------------------------------
 .../camel/component/nats/NatsConfiguration.java | 71 ++++++++++++--------
 .../camel/component/nats/NatsConsumer.java      | 19 +++---
 .../camel/component/nats/NatsEndpoint.java      | 11 +--
 .../camel/component/nats/NatsProducer.java      | 12 ++--
 4 files changed, 59 insertions(+), 54 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/e50136b7/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 43182b9..260d1a7 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
@@ -26,9 +26,11 @@ import org.apache.camel.spi.UriPath;
 @UriParams
 public class NatsConfiguration {
 
-	@UriPath @Metadata(required = "true")
+    @UriPath
+    @Metadata(required = "true")
     private String servers;
-    @UriParam @Metadata(required = "true")
+    @UriParam
+    @Metadata(required = "true")
     private String topic;
     @UriParam(defaultValue = "true")
     private boolean reconnect = true;
@@ -52,143 +54,156 @@ public class NatsConfiguration {
     private String maxMessages;
     @UriParam(label = "consumer", defaultValue = "10")
     private int poolSize = 10;
-    
+
     /**
-     * The Nats servers
+     * URLs to one or more NAT servers. Use comma to separate URLs when specifying multiple
servers.
      */
     public String getServers() {
         return servers;
     }
+
     public void setServers(String servers) {
         this.servers = servers;
     }
-    
+
     /**
      * The name of topic we want to use
-     */   
+     */
     public String getTopic() {
         return topic;
     }
+
     public void setTopic(String topic) {
         this.topic = topic;
     }
-        
+
     /**
      * Whether or not using reconnection feature
-     */   
+     */
     public boolean getReconnect() {
         return reconnect;
     }
+
     public void setReconnect(boolean reconnect) {
         this.reconnect = reconnect;
     }
-     
+
     /**
      * Whether or not running in pedantic mode (this affects performace)
-     */  
+     */
     public boolean getPedantic() {
         return pedantic;
     }
+
     public void setPedantic(boolean pedantic) {
         this.pedantic = pedantic;
     }
-    
+
     /**
      * Whether or not running in verbose mode
-     */  
+     */
     public boolean getVerbose() {
         return verbose;
     }
+
     public void setVerbose(boolean verbose) {
         this.verbose = verbose;
     }
-    
+
     /**
      * Whether or not using SSL
-     */ 
+     */
     public boolean getSsl() {
         return ssl;
     }
+
     public void setSsl(boolean ssl) {
         this.ssl = ssl;
     }
-    
+
     /**
      * Waiting time before attempts reconnection (in milliseconds)
-     */ 
+     */
     public int getReconnectTimeWait() {
         return reconnectTimeWait;
     }
+
     public void setReconnectTimeWait(int reconnectTimeWait) {
         this.reconnectTimeWait = reconnectTimeWait;
     }
-    
+
     /**
      * Max reconnection attempts
-     */ 
+     */
     public int getMaxReconnectAttempts() {
         return maxReconnectAttempts;
     }
+
     public void setMaxReconnectAttempts(int maxReconnectAttempts) {
         this.maxReconnectAttempts = maxReconnectAttempts;
     }
-    
+
     /**
      * Ping interval to be aware if connection is still alive (in milliseconds)
      */
     public int getPingInterval() {
         return pingInterval;
     }
+
     public void setPingInterval(int pingInterval) {
         this.pingInterval = pingInterval;
     }
-    
+
     /**
      * Whether or not randomizing the order of servers for the connection attempts
      */
     public boolean getNoRandomizeServers() {
         return noRandomizeServers;
     }
+
     public void setNoRandomizeServers(boolean noRandomizeServers) {
         this.noRandomizeServers = noRandomizeServers;
     }
-    
+
     /**
      * The Queue name if we are using nats for a queue configuration
      */
     public String getQueueName() {
         return queueName;
     }
+
     public void setQueueName(String queueName) {
         this.queueName = queueName;
     }
-    
+
     /**
      * Stop receiving messages from a topic we are subscribing to after maxMessages 
      */
     public String getMaxMessages() {
         return maxMessages;
     }
+
     public void setMaxMessages(String maxMessages) {
         this.maxMessages = maxMessages;
     }
-    
+
     /**
      * Consumer pool size
      */
     public int getPoolSize() {
         return poolSize;
     }
+
     public void setPoolSize(int poolSize) {
         this.poolSize = poolSize;
     }
-    
+
     private static <T> void addPropertyIfNotNull(Properties props, String key, T value)
{
         if (value != null) {
             props.put(key, value);
         }
     }
-    
+
     public Properties createProperties() {
         Properties props = new Properties();
         addPropertyIfNotNull(props, NatsPropertiesConstants.NATS_PROPERTY_URI, splitServers());
@@ -202,18 +217,18 @@ public class NatsConfiguration {
         addPropertyIfNotNull(props, NatsPropertiesConstants.NATS_PROPERTY_DONT_RANDOMIZE_SERVERS,
getNoRandomizeServers());
         return props;
     }
-    
+
     public Properties createSubProperties() {
         Properties props = new Properties();
         addPropertyIfNotNull(props, NatsPropertiesConstants.NATS_PROPERTY_QUEUE, getQueueName());
         addPropertyIfNotNull(props, NatsPropertiesConstants.NATS_PROPERTY_MAX_MESSAGES, getMaxMessages());
         return props;
     }
-    
+
     private String splitServers() {
         StringBuilder servers = new StringBuilder();
         String prefix = "nats://";
-        
+
         String[] pieces = getServers().split(",");
         for (int i = 0; i < pieces.length; i++) {
             if (i < pieces.length - 1) {

http://git-wip-us.apache.org/repos/asf/camel/blob/e50136b7/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 e63f485..029ab13 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
@@ -32,15 +32,13 @@ public class NatsConsumer extends DefaultConsumer {
 
     private static final Logger LOG = LoggerFactory.getLogger(NatsConsumer.class);
 
-    protected ExecutorService executor;
-    private final NatsEndpoint endpoint;
     private final Processor processor;
+    private ExecutorService executor;
     private Connection connection;
     private int sid;
 
     public NatsConsumer(NatsEndpoint endpoint, Processor processor) {
         super(endpoint, processor);
-        this.endpoint = endpoint;
         this.processor = processor;
     }
 
@@ -53,18 +51,20 @@ public class NatsConsumer extends DefaultConsumer {
     protected void doStart() throws Exception {
         super.doStart();
         LOG.debug("Starting Nats Consumer");
-        executor = endpoint.createExecutor();
+        executor = getEndpoint().createExecutor();
 
         LOG.debug("Getting Nats Connection");
         connection = getConnection();
 
         executor.submit(new NatsConsumingTask(connection, getEndpoint().getNatsConfiguration()));
-
     }
 
     @Override
     protected void doStop() throws Exception {
         super.doStop();
+
+        // TODO: Should we not unsubscribe first?
+
         LOG.debug("Stopping Nats Consumer");
         if (executor != null) {
             if (getEndpoint() != null && getEndpoint().getCamelContext() != null)
{
@@ -82,17 +82,14 @@ public class NatsConsumer extends DefaultConsumer {
     }
 
     private Connection getConnection() throws IOException, InterruptedException {
-
         Properties prop = getEndpoint().getNatsConfiguration().createProperties();
         connection = Connection.connect(prop);
-
         return connection;
     }
 
     class NatsConsumingTask implements Runnable {
 
         private final Connection connection;
-        
         private final NatsConfiguration configuration;
 
         public NatsConsumingTask(Connection connection, NatsConfiguration configuration)
{
@@ -105,7 +102,7 @@ public class NatsConsumer extends DefaultConsumer {
             try {
                 sid = connection.subscribe(getEndpoint().getNatsConfiguration().getTopic(),
configuration.createSubProperties(), new MsgHandler() {
                     public void execute(String msg) {
-                        LOG.debug("Received Message: " + msg);
+                        LOG.debug("Received Message: {}", msg);
                         Exchange exchange = getEndpoint().createExchange();
                         exchange.getIn().setBody(msg);
                         exchange.getIn().setHeader(NatsConstants.NATS_MESSAGE_TIMESTAMP,
System.currentTimeMillis());
@@ -117,8 +114,8 @@ public class NatsConsumer extends DefaultConsumer {
                         }
                     }
                 });
-            } catch (IOException e1) {
-                getExceptionHandler().handleException("Error during processing", e1);
+            } catch (Throwable e) {
+                getExceptionHandler().handleException("Error during processing", e);
             }
         }
     }

http://git-wip-us.apache.org/repos/asf/camel/blob/e50136b7/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsEndpoint.java
b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsEndpoint.java
index 6a076d4..160b8ee 100644
--- a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsEndpoint.java
+++ b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsEndpoint.java
@@ -24,14 +24,10 @@ import org.apache.camel.Producer;
 import org.apache.camel.impl.DefaultEndpoint;
 import org.apache.camel.spi.UriEndpoint;
 import org.apache.camel.spi.UriParam;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
-@UriEndpoint(scheme = "nats", title = "Nats", syntax = "nats:host", label = "messaging")
+@UriEndpoint(scheme = "nats", title = "Nats", syntax = "nats:servers", label = "messaging",
consumerClass = NatsConsumer.class)
 public class NatsEndpoint extends DefaultEndpoint {
 
-    private static final Logger LOG = LoggerFactory.getLogger(NatsEndpoint.class);
-
     @UriParam
     private NatsConfiguration configuration;
     
@@ -56,12 +52,9 @@ public class NatsEndpoint extends DefaultEndpoint {
 
     @Override
     public boolean isSingleton() {
-        return false;
+        return true;
     }
     
-    /**
-     * The nats Configuration
-     */
     public NatsConfiguration getNatsConfiguration() {
         return configuration;
     }

http://git-wip-us.apache.org/repos/asf/camel/blob/e50136b7/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 92e2424..89b2b23 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
@@ -43,10 +43,12 @@ public class NatsProducer extends DefaultProducer {
     @Override
     public void process(Exchange exchange) throws Exception {
         NatsConfiguration config = getEndpoint().getNatsConfiguration();
-        connection.publish(config.getTopic(), exchange.getIn().getBody(String.class).getBytes());
+        String body = exchange.getIn().getMandatoryBody(String.class);
+
+        LOG.debug("Publishing to topic: {}", config.getTopic());
+        connection.publish(config.getTopic(), body.getBytes());
     }
     
-
     @Override
     protected void doStart() throws Exception {
         super.doStart();
@@ -62,17 +64,15 @@ public class NatsProducer extends DefaultProducer {
         LOG.debug("Stopping Nats Producer");
         
         LOG.debug("Closing Nats Connection");
-        if (connection.isConnected()) {
+        if (connection != null && connection.isConnected()) {
             connection.close();
         }
     }
 
-    
     private Connection getConnection() throws IOException, InterruptedException {
-
         Properties prop = getEndpoint().getNatsConfiguration().createProperties();
         connection = Connection.connect(prop);
-
         return connection;
     }
+
 }


Mime
View raw message