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-7914: MQTT Endpoint disconnects on failure. Does not reconnect
Date Sun, 15 Feb 2015 10:33:05 GMT
Repository: camel
Updated Branches:
  refs/heads/camel-2.13.x dc7a1a901 -> e26a08640
  refs/heads/camel-2.14.x c6ff77dda -> 0218061f0
  refs/heads/master 594842a20 -> 15d7c596c


CAMEL-7914: MQTT Endpoint disconnects on failure. Does not reconnect


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

Branch: refs/heads/master
Commit: 15d7c596c4556ba5c85d0ebf4801228df85a2a58
Parents: 594842a
Author: Claus Ibsen <davsclaus@apache.org>
Authored: Sun Feb 15 11:15:29 2015 +0100
Committer: Claus Ibsen <davsclaus@apache.org>
Committed: Sun Feb 15 11:32:15 2015 +0100

----------------------------------------------------------------------
 .../camel/component/mqtt/MQTTConfiguration.java | 261 +++++++++++++++++++
 .../camel/component/mqtt/MQTTEndpoint.java      |  54 +++-
 .../mqtt/MQTTProducerReconnectTest.java         |  96 +++++++
 3 files changed, 407 insertions(+), 4 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/15d7c596/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTConfiguration.java
----------------------------------------------------------------------
diff --git a/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTConfiguration.java
b/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTConfiguration.java
index 428ffad..97e906c 100644
--- a/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTConfiguration.java
+++ b/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTConfiguration.java
@@ -17,11 +17,13 @@
 package org.apache.camel.component.mqtt;
 
 import java.net.URI;
+import java.net.URISyntaxException;
 import java.util.concurrent.Executor;
 import javax.net.ssl.SSLContext;
 
 import org.apache.camel.spi.UriParam;
 import org.apache.camel.spi.UriParams;
+import org.fusesource.hawtbuf.UTF8Buffer;
 import org.fusesource.hawtdispatch.DispatchQueue;
 import org.fusesource.hawtdispatch.transport.TcpTransport;
 import org.fusesource.mqtt.client.MQTT;
@@ -206,6 +208,265 @@ public class MQTTConfiguration extends MQTT {
         throw new IllegalArgumentException("There is no QoS with name " + qualityOfService);
     }
 
+    @Override
+    public void setTracer(Tracer tracer) {
+        super.setTracer(tracer);
+    }
+
+    @Override
+    public void setClientId(String clientId) {
+        super.setClientId(clientId);
+    }
+
+    @Override
+    public void setClientId(UTF8Buffer clientId) {
+        super.setClientId(clientId);
+    }
+
+    @Override
+    public void setKeepAlive(short keepAlive) {
+        super.setKeepAlive(keepAlive);
+    }
+
+    @Override
+    public void setPassword(String password) {
+        super.setPassword(password);
+    }
+
+    @Override
+    public void setPassword(UTF8Buffer password) {
+        super.setPassword(password);
+    }
+
+    @Override
+    public void setUserName(String userName) {
+        super.setUserName(userName);
+    }
+
+    @Override
+    public void setUserName(UTF8Buffer userName) {
+        super.setUserName(userName);
+    }
+
+    @Override
+    public void setWillMessage(String willMessage) {
+        super.setWillMessage(willMessage);
+    }
+
+    @Override
+    public void setWillMessage(UTF8Buffer willMessage) {
+        super.setWillMessage(willMessage);
+    }
+
+    @Override
+    public void setWillQos(QoS willQos) {
+        super.setWillQos(willQos);
+    }
+
+    @Override
+    public void setVersion(String version) {
+        super.setVersion(version);
+    }
+
+    @Override
+    public String getVersion() {
+        return super.getVersion();
+    }
+
+    @Override
+    public void setWillRetain(boolean willRetain) {
+        super.setWillRetain(willRetain);
+    }
+
+    @Override
+    public void setWillTopic(String willTopic) {
+        super.setWillTopic(willTopic);
+    }
+
+    @Override
+    public void setWillTopic(UTF8Buffer willTopic) {
+        super.setWillTopic(willTopic);
+    }
+
+    @Override
+    public Executor getBlockingExecutor() {
+        return super.getBlockingExecutor();
+    }
+
+    @Override
+    public void setBlockingExecutor(Executor blockingExecutor) {
+        super.setBlockingExecutor(blockingExecutor);
+    }
+
+    @Override
+    public DispatchQueue getDispatchQueue() {
+        return super.getDispatchQueue();
+    }
+
+    @Override
+    public void setDispatchQueue(DispatchQueue dispatchQueue) {
+        super.setDispatchQueue(dispatchQueue);
+    }
+
+    @Override
+    public URI getLocalAddress() {
+        return super.getLocalAddress();
+    }
+
+    @Override
+    public void setLocalAddress(String localAddress) throws URISyntaxException {
+        super.setLocalAddress(localAddress);
+    }
+
+    @Override
+    public void setLocalAddress(URI localAddress) {
+        super.setLocalAddress(localAddress);
+    }
+
+    @Override
+    public int getMaxReadRate() {
+        return super.getMaxReadRate();
+    }
+
+    @Override
+    public void setMaxReadRate(int maxReadRate) {
+        super.setMaxReadRate(maxReadRate);
+    }
+
+    @Override
+    public int getMaxWriteRate() {
+        return super.getMaxWriteRate();
+    }
+
+    @Override
+    public void setMaxWriteRate(int maxWriteRate) {
+        super.setMaxWriteRate(maxWriteRate);
+    }
+
+    @Override
+    public int getReceiveBufferSize() {
+        return super.getReceiveBufferSize();
+    }
+
+    @Override
+    public void setReceiveBufferSize(int receiveBufferSize) {
+        super.setReceiveBufferSize(receiveBufferSize);
+    }
+
+    @Override
+    public URI getHost() {
+        return super.getHost();
+    }
+
+    @Override
+    public void setHost(String host, int port) throws URISyntaxException {
+        super.setHost(host, port);
+    }
+
+    @Override
+    public void setHost(String host) throws URISyntaxException {
+        super.setHost(host);
+    }
+
+    @Override
+    public void setHost(URI host) {
+        super.setHost(host);
+    }
+
+    @Override
+    public int getSendBufferSize() {
+        return super.getSendBufferSize();
+    }
+
+    @Override
+    public void setSendBufferSize(int sendBufferSize) {
+        super.setSendBufferSize(sendBufferSize);
+    }
+
+    @Override
+    public SSLContext getSslContext() {
+        return super.getSslContext();
+    }
+
+    @Override
+    public void setSslContext(SSLContext sslContext) {
+        super.setSslContext(sslContext);
+    }
+
+    @Override
+    public int getTrafficClass() {
+        return super.getTrafficClass();
+    }
+
+    @Override
+    public void setTrafficClass(int trafficClass) {
+        super.setTrafficClass(trafficClass);
+    }
+
+    @Override
+    public boolean isUseLocalHost() {
+        return super.isUseLocalHost();
+    }
+
+    @Override
+    public void setUseLocalHost(boolean useLocalHost) {
+        super.setUseLocalHost(useLocalHost);
+    }
+
+    @Override
+    public long getConnectAttemptsMax() {
+        return super.getConnectAttemptsMax();
+    }
+
+    @Override
+    public void setConnectAttemptsMax(long connectAttemptsMax) {
+        super.setConnectAttemptsMax(connectAttemptsMax);
+    }
+
+    @Override
+    public long getReconnectAttemptsMax() {
+        return super.getReconnectAttemptsMax();
+    }
+
+    @Override
+    public void setReconnectAttemptsMax(long reconnectAttemptsMax) {
+        super.setReconnectAttemptsMax(reconnectAttemptsMax);
+    }
+
+    @Override
+    public double getReconnectBackOffMultiplier() {
+        return super.getReconnectBackOffMultiplier();
+    }
+
+    @Override
+    public void setReconnectBackOffMultiplier(double reconnectBackOffMultiplier) {
+        super.setReconnectBackOffMultiplier(reconnectBackOffMultiplier);
+    }
+
+    @Override
+    public long getReconnectDelay() {
+        return super.getReconnectDelay();
+    }
+
+    @Override
+    public void setReconnectDelay(long reconnectDelay) {
+        super.setReconnectDelay(reconnectDelay);
+    }
+
+    @Override
+    public long getReconnectDelayMax() {
+        return super.getReconnectDelayMax();
+    }
+
+    @Override
+    public void setReconnectDelayMax(long reconnectDelayMax) {
+        super.setReconnectDelayMax(reconnectDelayMax);
+    }
+
+    @Override
+    public Tracer getTracer() {
+        return super.getTracer();
+    }
 }
 
 

http://git-wip-us.apache.org/repos/asf/camel/blob/15d7c596/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTEndpoint.java
b/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTEndpoint.java
index 7e1737b..3563f6e 100644
--- a/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTEndpoint.java
+++ b/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTEndpoint.java
@@ -19,6 +19,7 @@ package org.apache.camel.component.mqtt;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 import org.apache.camel.Consumer;
 import org.apache.camel.Exchange;
@@ -47,6 +48,8 @@ import org.slf4j.LoggerFactory;
 public class MQTTEndpoint extends DefaultEndpoint {
     private static final Logger LOG = LoggerFactory.getLogger(MQTTEndpoint.class);
 
+    private static final int PUBLISH_MAX_RECONNECT_ATTEMPTS = 3;
+
     private CallbackConnection connection;
     @UriPath
     private String name;
@@ -87,14 +90,21 @@ public class MQTTEndpoint extends DefaultEndpoint {
     @Override
     protected void doStart() throws Exception {
         super.doStart();
+
+        createConnection();
+    }
+
+    protected void createConnection() {
         connection = configuration.callbackConnection();
 
         connection.listener(new Listener() {
             public void onConnected() {
+                connected = true;
                 LOG.info("MQTT Connection connected to {}", configuration.getHost());
             }
 
             public void onDisconnected() {
+                connected = false;
                 LOG.debug("MQTT Connection disconnected from {}", configuration.getHost());
             }
 
@@ -113,6 +123,9 @@ public class MQTTEndpoint extends DefaultEndpoint {
             }
 
             public void onFailure(Throwable value) {
+                // mark this connection as disconnected so we force re-connect
+                connected = false;
+                LOG.warn("Connection to " + configuration.getHost() + " failure due " + value.getMessage()
+ ". Forcing a disconnect to re-connect on next attempt.");
                 connection.disconnect(new Callback<Void>() {
                     public void onSuccess(Void value) {
                     }
@@ -123,11 +136,11 @@ public class MQTTEndpoint extends DefaultEndpoint {
                 });
             }
         });
-
-        
     }
 
     protected void doStop() throws Exception {
+        super.doStop();
+
         if (connection != null) {
             final Promise<Void> promise = new Promise<Void>();
             connection.getDispatchQueue().execute(new Task() {
@@ -146,13 +159,14 @@ public class MQTTEndpoint extends DefaultEndpoint {
             });
             promise.await(configuration.getDisconnectWaitInSeconds(), TimeUnit.SECONDS);
         }
-        super.doStop();
     }
-    
+
     void connect() throws Exception {
         final Promise<Object> promise = new Promise<Object>();
         connection.connect(new Callback<Void>() {
             public void onSuccess(Void value) {
+                LOG.debug("Connected to {}", configuration.getHost());
+
                 String subscribeTopicName = configuration.getSubscribeTopicName();
                 subscribeTopicName = subscribeTopicName != null ? subscribeTopicName.trim()
: null;
 
@@ -178,11 +192,13 @@ public class MQTTEndpoint extends DefaultEndpoint {
             }
 
             public void onFailure(Throwable value) {
+                LOG.warn("Failed to connect to " + configuration.getHost() + " due " + value.getMessage());
                 promise.onFailure(value);
                 connection.disconnect(null);
                 connected = false;
             }
         });
+        LOG.info("Connecting to {} using {} seconds timeout", configuration.getHost(), configuration.getConnectWaitInSeconds());
         promise.await(configuration.getConnectWaitInSeconds(), TimeUnit.SECONDS);
     }
     
@@ -191,9 +207,39 @@ public class MQTTEndpoint extends DefaultEndpoint {
     }
  
     void publish(final String topic, final byte[] payload, final QoS qoS, final boolean retain,
final Callback<Void> callback) throws Exception {
+        // if not connected then create a new connection to re-connect
+        boolean done = isConnected();
+        int attempt = 0;
+        TimeoutException timeout = null;
+        while (!done && attempt <= PUBLISH_MAX_RECONNECT_ATTEMPTS) {
+            attempt++;
+            try {
+                LOG.warn("#{} attempt to re-create connection to {} before publishing", attempt,
configuration.getHost());
+                createConnection();
+                connect();
+            } catch (TimeoutException e) {
+                timeout = e;
+                LOG.debug("Timed out after {} seconds after {} attempt to re-create connection
to {}",
+                        new Object[]{configuration.getConnectWaitInSeconds(), attempt, configuration.getHost()});
+            } catch (Throwable e) {
+                // other kind of error then exit asap
+                callback.onFailure(e);
+                return;
+            }
+
+            done = isConnected();
+        }
+
+        if (attempt > 3 && !isConnected()) {
+            LOG.warn("Cannot re-connect to {} after {} attempts", configuration.getHost(),
attempt);
+            callback.onFailure(timeout);
+            return;
+        }
+
         connection.getDispatchQueue().execute(new Task() {
             @Override
             public void run() {
+                LOG.debug("Publishing to {}", configuration.getHost());
                 connection.publish(topic, payload, qoS, retain, callback);
             }
         });

http://git-wip-us.apache.org/repos/asf/camel/blob/15d7c596/components/camel-mqtt/src/test/java/org/apache/camel/component/mqtt/MQTTProducerReconnectTest.java
----------------------------------------------------------------------
diff --git a/components/camel-mqtt/src/test/java/org/apache/camel/component/mqtt/MQTTProducerReconnectTest.java
b/components/camel-mqtt/src/test/java/org/apache/camel/component/mqtt/MQTTProducerReconnectTest.java
new file mode 100644
index 0000000..15281a8
--- /dev/null
+++ b/components/camel-mqtt/src/test/java/org/apache/camel/component/mqtt/MQTTProducerReconnectTest.java
@@ -0,0 +1,96 @@
+/**
+ * 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.mqtt;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.fusesource.mqtt.client.BlockingConnection;
+import org.fusesource.mqtt.client.MQTT;
+import org.fusesource.mqtt.client.Message;
+import org.fusesource.mqtt.client.QoS;
+import org.fusesource.mqtt.client.Topic;
+import org.junit.Test;
+
+public class MQTTProducerReconnectTest extends MQTTBaseTest {
+
+    @Test
+    public void testProduce() throws Exception {
+        MQTT mqtt = new MQTT();
+        final BlockingConnection subscribeConnection = mqtt.blockingConnection();
+        subscribeConnection.connect();
+        Topic topic = new Topic(TEST_TOPIC, QoS.AT_MOST_ONCE);
+        Topic[] topics = {topic};
+        subscribeConnection.subscribe(topics);
+        final CountDownLatch latch = new CountDownLatch(numberOfMessages);
+
+        Thread thread = new Thread(new Runnable() {
+            public void run() {
+                for (int i = 0; i < numberOfMessages; i++) {
+                    try {
+                        Message message = subscribeConnection.receive();
+                        message.ack();
+                        latch.countDown();
+                    } catch (Exception e) {
+                        e.printStackTrace();
+                        break;
+                    }
+                }
+            }
+        });
+        thread.start();
+
+        for (int i = 0; i < numberOfMessages; i++) {
+
+            if (i == 5) {
+                LOG.info("Stopping MQTT transport");
+                brokerService.getTransportConnectorByScheme("mqtt").stop();
+
+                Thread starter = new Thread(new Runnable() {
+                    public void run() {
+                        try {
+                            Thread.sleep(3000);
+                            LOG.info("Starting MQTT transport again");
+                            brokerService.getTransportConnectorByScheme("mqtt").start();
+                        } catch (Exception e) {
+                            // ignore
+                        }
+                    }
+                });
+                starter.start();
+            }
+
+            try {
+                template.sendBody("direct:foo", "test message " + i);
+            } catch (Exception e) {
+                // ignore
+            }
+        }
+
+        latch.await(20, TimeUnit.SECONDS);
+        assertTrue("Messages not consumed = " + latch.getCount(), latch.getCount() == 0);
+    }
+
+    protected RouteBuilder createRouteBuilder() {
+        return new RouteBuilder() {
+            public void configure() {
+                from("direct:foo").to("mqtt:boo?reconnectDelay=1000&publishTopicName="
+ TEST_TOPIC);
+            }
+        };
+    }
+}


Mime
View raw message