MQTT source as supplier/source

Resolves https://github.com/spring-cloud/stream-applications/issues/38

common/mqtt-common/pom.xml

@@ -0,0 +1,36 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>mqtt-common</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <name>mqtt-common</name>
+    <description>file consumer</description>
+
+    <parent>
+        <groupId>org.springframework.cloud.fn</groupId>
+        <artifactId>spring-functions-parent</artifactId>
+        <version>1.0.0-SNAPSHOT</version>
+        <relativePath>../../spring-functions-parent</relativePath>
+    </parent>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-mqtt</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-integration</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-validation</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+            <scope>provided</scope>
+        </dependency>
+    </dependencies>
+
+</project>

common/mqtt-common/src/main/java/org/springframework/cloud/fn/common/mqtt/MqttConfiguration.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2017-2020 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.cloud.fn.common.mqtt;
+
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
+import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
+import org.springframework.util.ObjectUtils;
+
+/**
+ * Generic mqtt configuration.
+ *
+ * @author Janne Valkealahti
+ *
+ */
+@Configuration
+public class MqttConfiguration {
+
+	@Autowired
+	private MqttProperties mqttProperties;
+
+	@Bean
+	public MqttPahoClientFactory mqttClientFactory() {
+
+		MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
+		mqttConnectOptions.setServerURIs(mqttProperties.getUrl());
+		mqttConnectOptions.setUserName(mqttProperties.getUsername());
+		mqttConnectOptions.setPassword(mqttProperties.getPassword().toCharArray());
+		mqttConnectOptions.setCleanSession(mqttProperties.isCleanSession());
+		mqttConnectOptions.setConnectionTimeout(mqttProperties.getConnectionTimeout());
+		mqttConnectOptions.setKeepAliveInterval(mqttProperties.getKeepAliveInterval());
+
+		DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
+		factory.setConnectionOptions(mqttConnectOptions);
+
+		if (ObjectUtils.nullSafeEquals(mqttProperties.getPersistence(), "file")) {
+			factory.setPersistence(new MqttDefaultFilePersistence(mqttProperties.getPersistenceDirectory()));
+		}
+		else if (ObjectUtils.nullSafeEquals(mqttProperties.getPersistence(), "memory")) {
+			factory.setPersistence(new MemoryPersistence());
+		}
+		return factory;
+	}
+}

common/mqtt-common/src/main/java/org/springframework/cloud/fn/common/mqtt/MqttProperties.java

@@ -0,0 +1,138 @@
+/*
+ * Copyright 2017-2020 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.cloud.fn.common.mqtt;
+
+import javax.validation.constraints.Size;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.validation.annotation.Validated;
+
+/**
+ * Generic mqtt connection properties.
+ *
+ * @author Janne Valkealahti
+ *
+ */
+@Validated
+@ConfigurationProperties("mqtt")
+public class MqttProperties {
+
+	/**
+	 * location of the mqtt broker(s) (comma-delimited list).
+	 */
+	private String[] url = new String[] { "tcp://localhost:1883" };
+
+	/**
+	 * the username to use when connecting to the broker.
+	 */
+	private String username = "guest";
+
+	/**
+	 * the password to use when connecting to the broker.
+	 */
+	private String password = "guest";
+
+	/**
+	 * whether the client and server should remember state across restarts and reconnects.
+	 */
+	private boolean cleanSession = true;
+
+	/**
+	 * the connection timeout in seconds.
+	 */
+	private int connectionTimeout = 30;
+
+	/**
+	 * the ping interval in seconds.
+	 */
+	private int keepAliveInterval = 60;
+
+	/**
+	 * 'memory' or 'file'.
+	 */
+	private String persistence = "memory";
+
+	/**
+	 * Persistence directory.
+	 */
+	private String persistenceDirectory = "/tmp/paho";
+
+	@Size(min = 1)
+	public String[] getUrl() {
+		return url;
+	}
+
+	public void setUrl(String[] url) {
+		this.url = url;
+	}
+
+	public String getUsername() {
+		return username;
+	}
+
+	public void setUsername(String username) {
+		this.username = username;
+	}
+
+	public String getPassword() {
+		return password;
+	}
+
+	public void setPassword(String password) {
+		this.password = password;
+	}
+
+	public boolean isCleanSession() {
+		return cleanSession;
+	}
+
+	public void setCleanSession(boolean cleanSession) {
+		this.cleanSession = cleanSession;
+	}
+
+	public int getKeepAliveInterval() {
+		return keepAliveInterval;
+	}
+
+	public void setKeepAliveInterval(int keepAliveInterval) {
+		this.keepAliveInterval = keepAliveInterval;
+	}
+
+	public int getConnectionTimeout() {
+		return connectionTimeout;
+	}
+
+	public void setConnectionTimeout(int connectionTimeout) {
+		this.connectionTimeout = connectionTimeout;
+	}
+
+	public String getPersistence() {
+		return persistence;
+	}
+
+	public void setPersistence(String persistence) {
+		this.persistence = persistence;
+	}
+
+	public String getPersistenceDirectory() {
+		return persistenceDirectory;
+	}
+
+	public void setPersistenceDirectory(String persistenceDirectory) {
+		this.persistenceDirectory = persistenceDirectory;
+	}
+}

pom.xml

@@ -43,6 +43,7 @@
 		<module>common/ftp-common</module>
 		<module>common/function-test-support</module>
 		<module>common/tcp-common</module>
+		<module>common/mqtt-common</module>
 
 		<module>consumer/cassandra-consumer</module>
 		<module>consumer/counter-consumer</module>
@@ -65,6 +66,7 @@
 		<module>supplier/http-supplier</module>
 		<module>supplier/jdbc-supplier</module>
 		<module>supplier/mongodb-supplier</module>
+		<module>supplier/mqtt-supplier</module>
 		<module>supplier/tcp-supplier</module>
 		<module>supplier/time-supplier</module>
 

supplier/mqtt-supplier/pom.xml

@@ -0,0 +1,53 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>mqtt-supplier</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <name>mqtt-supplier</name>
+    <description>mqtt supplier</description>
+
+    <parent>
+        <groupId>org.springframework.cloud.fn</groupId>
+        <artifactId>spring-functions-parent</artifactId>
+        <version>1.0.0-SNAPSHOT</version>
+        <relativePath>../../spring-functions-parent</relativePath>
+    </parent>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.cloud.fn</groupId>
+            <artifactId>mqtt-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-integration</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-validation</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>testcontainers</artifactId>
+            <version>1.9.1</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+</project>

supplier/mqtt-supplier/src/main/java/org/springframework/cloud/fn/supplier/mqtt/MqttSupplierConfiguration.java

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2017-2020 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.cloud.fn.supplier.mqtt;
+
+import java.util.function.Supplier;
+
+import reactor.core.publisher.Flux;
+
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.cloud.fn.common.mqtt.MqttConfiguration;
+import org.springframework.cloud.fn.common.mqtt.MqttProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.integration.channel.FluxMessageChannel;
+import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
+import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
+import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
+import org.springframework.messaging.Message;
+
+/**
+ * A source module that receives data from Mqtt.
+ *
+ * @author Janne Valkealahti
+ * @author Soby Chacko
+ */
+@Configuration
+@EnableConfigurationProperties({MqttProperties.class, MqttSupplierProperties.class})
+@Import(MqttConfiguration.class)
+public class MqttSupplierConfiguration {
+
+	@Autowired
+	private MqttSupplierProperties properties;
+
+	@Autowired
+	private MqttPahoClientFactory mqttClientFactory;
+
+	@Autowired
+	private BeanFactory beanFactory;
+
+	@Bean
+	public Supplier<Flux<Message<?>>> mqttSupplier() {
+		return () -> Flux.from(output())
+				.doOnSubscribe(subscription -> mqttInbound().start());
+	}
+
+	@Bean
+	public MqttPahoMessageDrivenChannelAdapter mqttInbound() {
+		MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(properties.getClientId(),
+				mqttClientFactory, properties.getTopics());
+		adapter.setQos(properties.getQos());
+		adapter.setConverter(pahoMessageConverter(beanFactory));
+		adapter.setOutputChannel(output());
+		adapter.setAutoStartup(false);
+		return adapter;
+	}
+
+	@Bean
+	public FluxMessageChannel output() {
+		return new FluxMessageChannel();
+	}
+
+	public DefaultPahoMessageConverter pahoMessageConverter(BeanFactory beanFactory) {
+		DefaultPahoMessageConverter converter = new DefaultPahoMessageConverter(properties.getCharset());
+		converter.setPayloadAsBytes(properties.isBinary());
+		converter.setBeanFactory(beanFactory);
+		return converter;
+	}
+}

supplier/mqtt-supplier/src/main/java/org/springframework/cloud/fn/supplier/mqtt/MqttSupplierProperties.java

@@ -0,0 +1,103 @@
+/*
+ * Copyright 2017-2020 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.cloud.fn.supplier.mqtt;
+
+import javax.validation.constraints.NotBlank;
+import javax.validation.constraints.Size;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.validation.annotation.Validated;
+
+/**
+ * Properties for the Mqtt Source.
+ *
+ * @author Janne Valkealahti
+ * @author Soby Chacko
+ *
+ */
+@Validated
+@ConfigurationProperties("mqtt.supplier")
+public class MqttSupplierProperties {
+
+	/**
+	 * identifies the client.
+	 */
+	private String clientId = "stream.client.id.source";
+
+	/**
+	 * the topic(s) (comma-delimited) to which the source will subscribe.
+	 */
+	private String[] topics = new String[] { "stream.mqtt" };
+
+	/**
+	 * the qos; a single value for all topics or a comma-delimited list to match the topics.
+	 */
+	private int[] qos = new int[] { 0 };
+
+	/**
+	 * true to leave the payload as bytes.
+	 */
+	private boolean binary = false;
+
+	/**
+	 * the charset used to convert bytes to String (when binary is false).
+	 */
+	private String charset = "UTF-8";
+
+	@NotBlank
+	@Size(min = 1, max = 23)
+	public String getClientId() {
+		return this.clientId;
+	}
+
+	public void setClientId(String clientId) {
+		this.clientId = clientId;
+	}
+
+	public String[] getTopics() {
+		return this.topics;
+	}
+
+	public void setTopics(String[] topics) {
+		this.topics = topics;
+	}
+
+	public int[] getQos() {
+		return this.qos;
+	}
+
+	public void setQos(int[] qos) {
+		this.qos = qos;
+	}
+
+	public String getCharset() {
+		return this.charset;
+	}
+
+	public void setCharset(String charset) {
+		this.charset = charset;
+	}
+
+	public boolean isBinary() {
+		return this.binary;
+	}
+
+	public void setBinary(boolean binary) {
+		this.binary = binary;
+	}
+
+}

supplier/mqtt-supplier/src/test/java/org/springframework/cloud/fn/supplier/mqtt/MqttSupplierTests.java

@@ -0,0 +1,108 @@
+/*
+ * Copyright 2017-2020 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.cloud.fn.supplier.mqtt;
+
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+import com.github.dockerjava.api.command.CreateContainerCmd;
+import com.github.dockerjava.api.model.ExposedPort;
+import com.github.dockerjava.api.model.PortBinding;
+import com.github.dockerjava.api.model.Ports;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.GenericContainer;
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Bean;
+import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
+import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
+import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.test.annotation.DirtiesContext;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for Mqtt Supplier.
+ *
+ * @author Janne Valkealahti
+ * @author Gary Russell
+ * @author Soby Chacko
+ *
+ */
+@SpringBootTest(properties = {"mqtt.supplier.topics=test,fake", "mqtt.supplier.qos=0,0"})
+@DirtiesContext
+public class MqttSupplierTests {
+
+	static {
+		Consumer<CreateContainerCmd> cmd = e -> e.withPortBindings(new PortBinding(Ports.Binding.bindPort(1883), new ExposedPort(1883)));
+		GenericContainer mosquitto = new GenericContainer("eclipse-mosquitto")
+				.withExposedPorts(1883)
+				.withCreateContainerCmdModifier(cmd);
+		mosquitto.start();
+	}
+
+	@Autowired
+	private Supplier<Flux<Message<?>>> mqttSupplier;
+
+	@Autowired
+	private MessageHandler mqttOutbound;
+
+	@Test
+	public void testBasicFlow() {
+
+		mqttOutbound.handleMessage(MessageBuilder.withPayload("hello").build());
+
+		final Flux<Message<?>> messageFlux = mqttSupplier.get();
+
+		StepVerifier.create(messageFlux)
+				.assertNext((message) -> {
+							assertThat(message.getPayload())
+									.isEqualTo("hello");
+						}
+				)
+				.thenCancel()
+				.verify();
+	}
+
+	@SpringBootApplication
+	static class TestApplication {
+
+		@Autowired
+		private MqttPahoClientFactory mqttClientFactory;
+
+		@Bean
+		public MessageHandler mqttOutbound() {
+			MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler("test", mqttClientFactory);
+			messageHandler.setAsync(true);
+			messageHandler.setDefaultTopic("test");
+			messageHandler.setConverter(producerConverter());
+			return messageHandler;
+		}
+
+		@Bean
+		public DefaultPahoMessageConverter producerConverter() {
+			return new DefaultPahoMessageConverter(1, true, "UTF-8");
+		}
+	}
+}