MQTT client skeleton example

application-skeleton-mqtt/pom.xml

@@ -0,0 +1,54 @@
+<project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd" xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+	<modelVersion>4.0.0</modelVersion>
+
+	<parent>
+    	<groupId>eu.arrowhead</groupId>	
+		<artifactId>application-skeleton-java-spring</artifactId>
+		<version>4.4.0.0</version>
+  	</parent>
+
+	<artifactId>arrowhead-application-skeleton-mqtt</artifactId>
+	<name>Arrowhead MQTT PubSub Skeleton</name>
+
+	<build>
+		<plugins>
+			<plugin>
+    			<groupId>org.springframework.boot</groupId>
+    			<artifactId>spring-boot-maven-plugin</artifactId>
+    			<executions>
+    				<execution>
+    					<goals>
+    						<goal>repackage</goal>
+    					</goals>
+    				</execution>
+    			</executions>
+  			</plugin>
+
+  			<plugin>
+	 			<artifactId>maven-resources-plugin</artifactId>
+	        	<version>3.1.0</version>
+	        	<executions>
+	          		<execution>
+	            		<id>copy-resources</id>
+	            		<phase>validate</phase>
+		            	<goals>
+		              		<goal>copy-resources</goal>
+		            	</goals>
+		            	<configuration>
+		              		<outputDirectory>${basedir}/target</outputDirectory>
+		              		<resources>
+		                		<resource>
+		                  			<directory>src/main/resources</directory>
+		                  			<includes>
+		                  				<include>application.properties</include>
+		                  			</includes>
+		                		</resource>
+		              		</resources>
+		            	</configuration>
+	          		</execution>
+	        	</executions>
+	      	</plugin>
+		</plugins>
+	</build>
+
+</project>

application-skeleton-mqtt/src/main/java/eu/arrowhead/application/skeleton/pubsub/PubSubApplicationInitListener.java

@@ -0,0 +1,119 @@
+package eu.arrowhead.application.skeleton.pubsub;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.event.ContextRefreshedEvent;
+import org.springframework.stereotype.Component;
+
+import ai.aitia.arrowhead.application.library.ArrowheadService;
+import ai.aitia.arrowhead.application.library.config.ApplicationInitListener;
+import ai.aitia.arrowhead.application.library.util.ApplicationCommonConstants;
+import eu.arrowhead.common.CommonConstants;
+import eu.arrowhead.common.Utilities;
+import eu.arrowhead.common.core.CoreSystem;
+import eu.arrowhead.common.exception.ArrowheadException;
+
+import org.eclipse.paho.client.mqttv3.MqttCallback;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.eclipse.paho.client.mqttv3.internal.security.SSLSocketFactoryFactory;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
+import org.eclipse.paho.client.mqttv3.MqttException;
+
+@Component
+public class PubSubApplicationInitListener extends ApplicationInitListener implements MqttCallback {
+
+	//=================================================================================================
+	// members
+
+	@Autowired
+	private ArrowheadService arrowheadService;
+
+	//@Value(CommonConstants.$APPLICATION_SYSTEM_NAME)
+	@Value("${application_system_name}")
+	private String systemName;
+
+	//@Value(CommonConstants.$MQTT_BROKER_ADDRESS)
+	@Value("${mqtt.broker.address}")
+	private String brokerAddress;
+
+	//@Value(CommonConstants.$MQTT_BROKER_PORT)
+	@Value("${mqtt.broker.port}")
+	private int brokerPort;
+
+	//@Value(CommonConstants.$MQTT_BROKER_USERNAME)
+	@Value("${mqtt.broker.username}")
+	private String brokerUsername;
+
+	//@Value(CommonConstants.$MQTT_BROKER_PASSWORD)
+	@Value("${mqtt.broker.password}")
+	private String brokerPassword;
+
+	@Value(CommonConstants.$SERVER_SSL_ENABLED_WD)
+	private boolean sslEnabled;
+
+	MqttClient client = null;
+	private final Logger logger = LogManager.getLogger(PubSubApplicationInitListener.class);
+
+	//=================================================================================================
+	// methods
+
+	//-------------------------------------------------------------------------------------------------
+	@Override
+	protected void customInit(final ContextRefreshedEvent event) {
+
+        //TODO: implement here any custom behavior on application start up
+        try {
+            client = arrowheadService.connectMQTTBroker(this, brokerAddress, brokerPort, brokerUsername, brokerPassword, systemName);
+
+            //Checking the availability of necessary core systems  add MQTT support here as well?
+            //checkCoreSystemReachability(CoreSystem.SERVICEREGISTRY); add MQTT support here as well?
+            //checkCoreSystemReachability(CoreSystem.AUTHORIZATION); add MQTT support here as well?
+            client.subscribe("ah/pubsub/messages");
+        } catch (Exception e) {
+            client = null;
+            throw new ArrowheadException("Could not connect to MQTT broker!\n" + e.toString());
+        }
+	}
+
+	//-------------------------------------------------------------------------------------------------
+	@Override
+	public void customDestroy() {
+        //TODO: implement here any custom behavior on application shut down
+        if (client != null) {
+            try {
+                arrowheadService.disconnectMQTTBroker(client);
+                arrowheadService.closeMQTTBroker(client);
+            } catch(Exception e) {
+            }
+            client = null;
+        }
+    }
+
+	//=================================================================================================
+	// MQTT callback methods
+
+    @Override
+    public void connectionLost(Throwable t) {
+
+        //TODO: implement here any custom behavior on broker disconnect, typically a reconnect after a short timeout
+    }
+
+    @Override
+    public void messageArrived(String topic, MqttMessage message) throws Exception {
+
+        //TODO: implement here any custom behavior on incoming messages, such as filtering in topics etc
+    }
+
+    @Override
+    public void deliveryComplete(IMqttDeliveryToken token) {
+    }
+
+	//=================================================================================================
+	// assistant methods
+
+}

application-skeleton-mqtt/src/main/java/eu/arrowhead/application/skeleton/pubsub/PubSubMain.java

@@ -0,0 +1,20 @@
+package eu.arrowhead.application.skeleton.pubsub;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.ComponentScan;
+
+import eu.arrowhead.common.CommonConstants;
+
+@SpringBootApplication
+@ComponentScan(basePackages = {CommonConstants.BASE_PACKAGE, "ai.aitia"}) //TODO: add custom packages if any
+public class PubSubMain {
+
+	//=================================================================================================
+	// methods
+
+	//-------------------------------------------------------------------------------------------------
+	public static void main(final String[] args) {
+		SpringApplication.run(PubSubMain.class, args);
+	}	
+}

application-skeleton-mqtt/src/main/resources/META-INF/additional-spring-configuration-metadata.json

@@ -0,0 +1,32 @@
+{"properties": [
+  {
+    "name": "application_system_name",
+    "type": "java.lang.String",
+    "description": "A description for 'application_system_name'"
+  },
+  {
+    "name": "broker.address",
+    "type": "java.lang.String",
+    "description": "A description for 'broker.address'"
+  },
+  {
+    "name": "broker.port",
+    "type": "java.lang.Integer",
+    "description": "A description for 'broker.port'"
+  },
+  {
+    "name": "broker.username",
+    "type": "java.lang.String",
+    "description": "A description for 'broker.username'"
+  },
+  {
+    "name": "broker.password",
+    "type": "java.lang.String",
+    "description": "A description for 'broker.password'"
+  },
+  {
+    "name": "container.max_keep_alive_requests",
+    "type": "java.lang.Integer",
+    "description": "A description for 'container.max_keep_alive_requests'"
+  }
+]}

application-skeleton-mqtt/src/main/resources/application.properties

@@ -0,0 +1,36 @@
+############################################
+###       CUSTOM PARAMETERS              ###
+############################################
+
+# Name of the client system
+application_system_name=PUBSUBSKELETON
+
+# Disable Tomcat for MQTT only applications
+spring.main.web-application-type=none
+
+# MQTT Broker parameters
+mqtt.broker.address=127.0.0.1
+mqtt.broker.port=8883
+mqtt.broker.username=pubsub
+mqtt.broker.password=badpassword
+
+
+############################################
+###           SECURE MODE                ###
+############################################
+
+# configure secure mode
+
+# Set this to false to disable mqtt(s) mode
+server.ssl.enabled=true
+
+
+server.ssl.key-store-type=PKCS12
+server.ssl.key-store=classpath:certificates/pubsub.p12
+server.ssl.key-store-password=123456
+server.ssl.key-alias=providerskeleton
+server.ssl.key-password=123456
+server.ssl.client-auth=need
+server.ssl.trust-store-type=PKCS12
+server.ssl.trust-store=classpath:certificates/truststore.p12
+server.ssl.trust-store-password=123456

pom.xml

@@ -28,6 +28,7 @@
 	    <module>application-skeleton-provider</module>
 	    <module>application-skeleton-subscriber</module>
 	    <module>application-skeleton-publisher</module>
+	    <module>application-skeleton-mqtt</module>
   	</modules>
 
 	<dependencies>