diff --git a/consumer/mqtt-consumer/pom.xml b/consumer/mqtt-consumer/pom.xml new file mode 100644 index 00000000..e1a540ab --- /dev/null +++ b/consumer/mqtt-consumer/pom.xml @@ -0,0 +1,52 @@ + + + 4.0.0 + mqtt-consumer + 1.0.0-SNAPSHOT + mqtt-consumer + mqtt consumer + + + org.springframework.cloud.fn + spring-functions-parent + 1.0.0-SNAPSHOT + ../../spring-functions-parent + + + + + org.springframework.cloud.fn + mqtt-common + ${project.version} + + + org.springframework.boot + spring-boot-starter-integration + + + org.springframework.boot + spring-boot-starter-validation + + + org.springframework.boot + spring-boot-configuration-processor + provided + + + org.springframework.boot + spring-boot-starter-test + test + + + io.projectreactor + reactor-test + test + + + org.testcontainers + testcontainers + 1.9.1 + test + + + diff --git a/consumer/mqtt-consumer/src/main/java/org/springframework/cloud/fn/consumer/mqtt/MqttConsumerConfiguration.java b/consumer/mqtt-consumer/src/main/java/org/springframework/cloud/fn/consumer/mqtt/MqttConsumerConfiguration.java new file mode 100644 index 00000000..85a69726 --- /dev/null +++ b/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> 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; + } +} diff --git a/consumer/mqtt-consumer/src/main/java/org/springframework/cloud/fn/consumer/mqtt/MqttConsumerProperties.java b/consumer/mqtt-consumer/src/main/java/org/springframework/cloud/fn/consumer/mqtt/MqttConsumerProperties.java new file mode 100644 index 00000000..7b60ad66 --- /dev/null +++ b/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; + } +} diff --git a/consumer/mqtt-consumer/src/test/java/org/springframework/cloud/fn/consumer/mqtt/MqttConsumerTests.java b/consumer/mqtt-consumer/src/test/java/org/springframework/cloud/fn/consumer/mqtt/MqttConsumerTests.java new file mode 100644 index 00000000..a5cd3608 --- /dev/null +++ b/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 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> 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(); + } + } +} diff --git a/pom.xml b/pom.xml index 44d8cb63..1f921301 100644 --- a/pom.xml +++ b/pom.xml @@ -42,7 +42,10 @@ common/ftp-common common/function-test-support +<<<<<<< HEAD common/tcp-common +======= +>>>>>>> MQTT source as supplier/source common/mqtt-common consumer/cassandra-consumer @@ -52,6 +55,7 @@ consumer/jdbc-consumer consumer/log-consumer consumer/mongodb-consumer + consumer/mqtt-consumer consumer/rabbit-consumer consumer/redis-consumer consumer/sftp-consumer