Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Auto-reconnect to MQTT-Broker #28

Open
wants to merge 4 commits into
base: master
Choose a base branch
from
Open

Auto-reconnect to MQTT-Broker #28

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
1904d31
add possibility to to auto reconnect in config
chris060986 Jan 28, 2017
9d2c738
refactore start method to make code reusable
chris060986 Jan 28, 2017
f5e3234
implement periodic reconnect (1second) in own thread
chris060986 Jan 28, 2017
c9d4edc
also subscribe to mqtt topic after reconnect
chris060986 Jan 30, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,8 @@ dependencies {
compile 'commons-io:commons-io:2.4'
compile 'org.slf4j:slf4j-api:1.7.14'
testCompile 'org.slf4j:slf4j-simple:1.7.14'
testCompile "org.mockito:mockito-core:2.6.9"
testCompile "org.assertj:assertj-core:3.6.2"
}

checkstyle {
Expand Down
2 changes: 2 additions & 0 deletions config/mqtt.properties
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,8 @@ tasks.max=1
#mqtt.connection_timeout=30
# What mqtt connection keep alive to use - defaults to `60` seconds
#mqtt.keep_alive_interval=60
# try to reconnect if connection lost, default false
#mqtt.auto_reconnect=false
# Mqtt broker address to use - defaults to `tcp://localhost:1883`
# if using TLS then certs can be used
#mqtt.server_uris=tcp://localhost:1883
Expand Down
2 changes: 2 additions & 0 deletions src/main/java/com/evokly/kafka/connect/mqtt/MqttSourceConnectorConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,8 @@ public static ConfigDef baseConfigDef() {
"connection timeout to use")
.define(MqttSourceConstant.MQTT_KEEP_ALIVE_INTERVAL, Type.INT, 60, Importance.LOW,
"keepalive interval to use")
.define(MqttSourceConstant.MQTT_AUTO_RECONNECT, Type.BOOLEAN, false, Importance.LOW,
"flag if client should reconnect when connection is lost")
.define(MqttSourceConstant.MQTT_SERVER_URIS, Type.STRING,
"tcp://localhost:1883", Importance.HIGH,
"mqtt server to connect to")
Expand Down
3 changes: 2 additions & 1 deletion src/main/java/com/evokly/kafka/connect/mqtt/MqttSourceConstant.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,14 +13,15 @@ public class MqttSourceConstant {
public static final String MQTT_CLEAN_SESSION = "mqtt.clean_session";
public static final String MQTT_CONNECTION_TIMEOUT = "mqtt.connection_timeout";
public static final String MQTT_KEEP_ALIVE_INTERVAL = "mqtt.keep_alive_interval";
public static final String MQTT_AUTO_RECONNECT = "mqtt.auto_reconnect";
public static final String MQTT_SERVER_URIS = "mqtt.server_uris";
public static final String MQTT_TOPIC = "mqtt.topic";
public static final String MQTT_QUALITY_OF_SERVICE = "mqtt.qos";
public static final String MQTT_SSL_CA_CERT = "mqtt.ssl.ca_cert";
public static final String MQTT_SSL_CERT = "mqtt.ssl.cert";
public static final String MQTT_SSL_PRIV_KEY = "mqtt.ssl.key";
public static final String MQTT_USERNAME = "mqtt.user";
public static final String MQTT_PASSWORD = "mqtt.password";

public static final String MQTT_PASSWORD = "mqtt.password";
public static final String MESSAGE_PROCESSOR = "message_processor_class";
}
Loading