diff --git a/common/mqtt-common/pom.xml b/common/mqtt-common/pom.xml new file mode 100644 index 00000000..5d36d7e9 --- /dev/null +++ b/common/mqtt-common/pom.xml @@ -0,0 +1,36 @@ + + + 4.0.0 + mqtt-common + 1.0.0-SNAPSHOT + mqtt-common + file consumer + + + org.springframework.cloud.fn + spring-functions-parent + 1.0.0-SNAPSHOT + ../../spring-functions-parent + + + + + org.springframework.integration + spring-integration-mqtt + + + org.springframework.boot + spring-boot-starter-integration + + + org.springframework.boot + spring-boot-starter-validation + + + org.springframework.boot + spring-boot-configuration-processor + provided + + + + diff --git a/common/mqtt-common/src/main/java/org/springframework/cloud/fn/common/mqtt/MqttConfiguration.java b/common/mqtt-common/src/main/java/org/springframework/cloud/fn/common/mqtt/MqttConfiguration.java new file mode 100644 index 00000000..0335940b --- /dev/null +++ b/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; + } +} diff --git a/common/mqtt-common/src/main/java/org/springframework/cloud/fn/common/mqtt/MqttProperties.java b/common/mqtt-common/src/main/java/org/springframework/cloud/fn/common/mqtt/MqttProperties.java new file mode 100644 index 00000000..e8c414f8 --- /dev/null +++ b/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; + } +} diff --git a/pom.xml b/pom.xml index ed08a63d..360d1f90 100644 --- a/pom.xml +++ b/pom.xml @@ -43,6 +43,7 @@ common/ftp-common common/function-test-support common/tcp-common + common/mqtt-common consumer/cassandra-consumer consumer/counter-consumer @@ -65,6 +66,7 @@ supplier/http-supplier supplier/jdbc-supplier supplier/mongodb-supplier + supplier/mqtt-supplier supplier/tcp-supplier supplier/time-supplier diff --git a/supplier/mqtt-supplier/pom.xml b/supplier/mqtt-supplier/pom.xml new file mode 100644 index 00000000..1749f5d6 --- /dev/null +++ b/supplier/mqtt-supplier/pom.xml @@ -0,0 +1,53 @@ + + + 4.0.0 + mqtt-supplier + 1.0.0-SNAPSHOT + mqtt-supplier + mqtt supplier + + + 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/supplier/mqtt-supplier/src/main/java/org/springframework/cloud/fn/supplier/mqtt/MqttSupplierConfiguration.java b/supplier/mqtt-supplier/src/main/java/org/springframework/cloud/fn/supplier/mqtt/MqttSupplierConfiguration.java new file mode 100644 index 00000000..c80d08cd --- /dev/null +++ b/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>> 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; + } +} diff --git a/supplier/mqtt-supplier/src/main/java/org/springframework/cloud/fn/supplier/mqtt/MqttSupplierProperties.java b/supplier/mqtt-supplier/src/main/java/org/springframework/cloud/fn/supplier/mqtt/MqttSupplierProperties.java new file mode 100644 index 00000000..79f3c670 --- /dev/null +++ b/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; + } + +} diff --git a/supplier/mqtt-supplier/src/test/java/org/springframework/cloud/fn/supplier/mqtt/MqttSupplierTests.java b/supplier/mqtt-supplier/src/test/java/org/springframework/cloud/fn/supplier/mqtt/MqttSupplierTests.java new file mode 100644 index 00000000..3af340ee --- /dev/null +++ b/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 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>> mqttSupplier; + + @Autowired + private MessageHandler mqttOutbound; + + @Test + public void testBasicFlow() { + + mqttOutbound.handleMessage(MessageBuilder.withPayload("hello").build()); + + final Flux> 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"); + } + } +}