Mqtt sink

* MQTT Consumer/Sink

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

consumer/mqtt-consumer/pom.xml

@@ -0,0 +1,52 @@
+<?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-consumer</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <name>mqtt-consumer</name>
+    <description>mqtt 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.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>

consumer/mqtt-consumer/src/main/java/org/springframework/cloud/fn/consumer/mqtt/MqttConsumerConfiguration.java

@@ -0,0 +1,75 @@
+/*
+ * 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.consumer.mqtt;
+
+import java.util.function.Consumer;
+
+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.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;
+
+/**
+ * A consumer that sends data to Mqtt.
+ *
+ * @author Janne Valkealahti
+ *
+ */
+@Configuration
+@EnableConfigurationProperties({ MqttProperties.class, MqttConsumerProperties.class })
+@Import(MqttConfiguration.class)
+public class MqttConsumerConfiguration {
+
+	@Autowired
+	private MqttConsumerProperties properties;
+
+	@Autowired
+	private MqttPahoClientFactory mqttClientFactory;
+
+	@Autowired
+	private BeanFactory beanFactory;
+
+	@Bean
+	public Consumer<Message<?>> mqttConsumer() {
+		return mqttOutbound()::handleMessage;
+	}
+
+	@Bean
+	public MessageHandler mqttOutbound() {
+		MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(properties.getClientId(), mqttClientFactory);
+		messageHandler.setAsync(properties.isAsync());
+		messageHandler.setDefaultTopic(properties.getTopic());
+		messageHandler.setConverter(pahoMessageConverter());
+		return messageHandler;
+	}
+
+	public DefaultPahoMessageConverter pahoMessageConverter() {
+		DefaultPahoMessageConverter converter = new DefaultPahoMessageConverter(properties.getQos(),
+				properties.isRetained(), properties.getCharset());
+		converter.setBeanFactory(beanFactory);
+		return converter;
+	}
+}

consumer/mqtt-consumer/src/main/java/org/springframework/cloud/fn/consumer/mqtt/MqttConsumerProperties.java

@@ -0,0 +1,118 @@
+/*
+ * 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.consumer.mqtt;
+
+import javax.validation.constraints.NotBlank;
+import javax.validation.constraints.Size;
+
+import org.hibernate.validator.constraints.Range;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.validation.annotation.Validated;
+
+/**
+ * Properties for the Mqtt Consumer.
+ *
+ * @author Janne Valkealahti
+ *
+ */
+@Validated
+@ConfigurationProperties("mqtt.consumer")
+public class MqttConsumerProperties {
+
+	/**
+	 * identifies the client.
+	 */
+	private String clientId = "stream.client.id.sink";
+
+	/**
+	 * the topic to which the sink will publish.
+	 */
+	private String topic = "stream.mqtt";
+
+	/**
+	 * the quality of service to use.
+	 */
+	private int qos = 1;
+
+	/**
+	 * whether to set the 'retained' flag.
+	 */
+	private boolean retained = false;
+
+	/**
+	 * the charset used to convert a String payload to byte[].
+	 */
+	private String charset = "UTF-8";
+
+	/**
+	 * whether or not to use async sends.
+	 */
+	private boolean async = false;
+
+	@Range(min = 0, max = 2)
+	public int getQos() {
+		return this.qos;
+	}
+
+	public boolean isRetained() {
+		return this.retained;
+	}
+
+	public void setQos(int qos) {
+		this.qos = qos;
+	}
+
+	public void setRetained(boolean retained) {
+		this.retained = retained;
+	}
+
+	@NotBlank
+	@Size(min = 1, max = 23)
+	public String getClientId() {
+		return this.clientId;
+	}
+
+	public void setClientId(String clientId) {
+		this.clientId = clientId;
+	}
+
+	@NotBlank
+	public String getTopic() {
+		return this.topic;
+	}
+
+	public void setTopic(String topic) {
+		this.topic = topic;
+	}
+
+	public String getCharset() {
+		return this.charset;
+	}
+
+	public void setCharset(String charset) {
+		this.charset = charset;
+	}
+
+	public boolean isAsync() {
+		return this.async;
+	}
+
+	public void setAsync(boolean async) {
+		this.async = async;
+	}
+}

consumer/mqtt-consumer/src/test/java/org/springframework/cloud/fn/consumer/mqtt/MqttConsumerTests.java

@@ -0,0 +1,97 @@
+/*
+ * 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.consumer.mqtt;
+
+import java.util.function.Consumer;
+
+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 org.springframework.beans.factory.BeanFactory;
+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.channel.QueueChannel;
+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;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.test.annotation.DirtiesContext;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@SpringBootTest(properties = "mqtt.consumer.topic=test")
+@DirtiesContext
+public class MqttConsumerTests {
+
+	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 Consumer<Message<?>> mqttConsumer;
+
+	@Autowired
+	protected QueueChannel queue;
+
+	@Test
+	public void testMqttConsumer() {
+		this.mqttConsumer.accept(MessageBuilder.withPayload("hello").build());
+		Message<?> in = this.queue.receive(10000);
+		assertThat(in).isNotNull();
+		assertThat(in.getPayload()).isEqualTo("hello");
+	}
+
+	@SpringBootApplication
+	static class TestApplication {
+
+		@Autowired
+		private MqttPahoClientFactory mqttClientFactory;
+
+		@Bean
+		public MqttPahoMessageDrivenChannelAdapter mqttInbound(BeanFactory beanFactory) {
+			MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("test",
+					mqttClientFactory, "test");
+			adapter.setQos(0);
+			adapter.setConverter(pahoMessageConverter(beanFactory));
+			adapter.setOutputChannelName("queue");
+			return adapter;
+		}
+
+		public DefaultPahoMessageConverter pahoMessageConverter(BeanFactory beanFactory) {
+			DefaultPahoMessageConverter converter = new DefaultPahoMessageConverter(1, true, "UTF-8");
+			converter.setPayloadAsBytes(false);
+			converter.setBeanFactory(beanFactory);
+			return converter;
+		}
+
+		@Bean
+		public QueueChannel queue() {
+			return new QueueChannel();
+		}
+	}
+}

pom.xml

@@ -42,7 +42,10 @@
   	<modules>
 		<module>common/ftp-common</module>
 		<module>common/function-test-support</module>
+<<<<<<< HEAD
 		<module>common/tcp-common</module>
+=======
+>>>>>>> MQTT source as supplier/source
 		<module>common/mqtt-common</module>
 
 		<module>consumer/cassandra-consumer</module>
@@ -52,6 +55,7 @@
 		<module>consumer/jdbc-consumer</module>
 		<module>consumer/log-consumer</module>
 		<module>consumer/mongodb-consumer</module>
+		<module>consumer/mqtt-consumer</module>
 		<module>consumer/rabbit-consumer</module>
 		<module>consumer/redis-consumer</module>
 		<module>consumer/sftp-consumer</module>