diff --git a/applications/source/mqtt-source/README.adoc b/applications/source/mqtt-source/README.adoc new file mode 100644 index 00000000..2b0ee89f --- /dev/null +++ b/applications/source/mqtt-source/README.adoc @@ -0,0 +1,31 @@ +//tag::ref-doc[] += MQTT Source + +Source that enables receiving messages from MQTT. + +==== Payload: + +* `String` if binary setting is `false` (default) +* `byte[]` if binary setting is `true` + +== Options + +The **$$mqtt$$** $$source$$ has the following options: + +//tag::configuration-properties[] +$$mqtt.clean-session$$:: $$whether the client and server should remember state across restarts and reconnects.$$ *($$Boolean$$, default: `$$true$$`)* +$$mqtt.connection-timeout$$:: $$the connection timeout in seconds.$$ *($$Integer$$, default: `$$30$$`)* +$$mqtt.keep-alive-interval$$:: $$the ping interval in seconds.$$ *($$Integer$$, default: `$$60$$`)* +$$mqtt.password$$:: $$the password to use when connecting to the broker.$$ *($$String$$, default: `$$guest$$`)* +$$mqtt.persistence$$:: $$'memory' or 'file'.$$ *($$String$$, default: `$$memory$$`)* +$$mqtt.persistence-directory$$:: $$Persistence directory.$$ *($$String$$, default: `$$/tmp/paho$$`)* +$$mqtt.supplier.binary$$:: $$true to leave the payload as bytes.$$ *($$Boolean$$, default: `$$false$$`)* +$$mqtt.supplier.charset$$:: $$the charset used to convert bytes to String (when binary is false).$$ *($$String$$, default: `$$UTF-8$$`)* +$$mqtt.supplier.client-id$$:: $$identifies the client.$$ *($$String$$, default: `$$stream.client.id.source$$`)* +$$mqtt.supplier.qos$$:: $$the qos; a single value for all topics or a comma-delimited list to match the topics.$$ *($$Integer[]$$, default: `$$[0]$$`)* +$$mqtt.supplier.topics$$:: $$the topic(s) (comma-delimited) to which the source will subscribe.$$ *($$String[]$$, default: `$$[stream.mqtt]$$`)* +$$mqtt.url$$:: $$location of the mqtt broker(s) (comma-delimited list).$$ *($$String[]$$, default: `$$[tcp://localhost:1883]$$`)* +$$mqtt.username$$:: $$the username to use when connecting to the broker.$$ *($$String$$, default: `$$guest$$`)* +//end::configuration-properties[] + +//end::ref-doc[] diff --git a/applications/source/mqtt-source/pom.xml b/applications/source/mqtt-source/pom.xml new file mode 100644 index 00000000..b1ed91e4 --- /dev/null +++ b/applications/source/mqtt-source/pom.xml @@ -0,0 +1,84 @@ + + + 4.0.0 + mqtt-source + 3.0.0-SNAPSHOT + mqtt-source + mqtt source apps + jar + + + org.springframework.cloud.stream.app + stream-applications-core + 3.0.0-SNAPSHOT + + + + + + org.springframework.boot + spring-boot-starter-test + test + + + org.springframework.cloud.fn + mqtt-supplier + ${java-functions.version} + + + org.testcontainers + testcontainers + 1.9.1 + test + + + + + + + org.springframework.cloud + spring-cloud-app-starter-doc-maven-plugin + + + org.springframework.cloud.stream.app.plugin + spring-cloud-stream-app-maven-plugin + + + mqtt + source + ${project.version} + org.springframework.cloud.fn.supplier.mqtt.MqttSupplierConfiguration.class + + + + org.springframework.cloud.fn + mqtt-supplier + ${java-functions.version} + + + + + + + + + + + true + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + + + false + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + + + diff --git a/applications/source/mqtt-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties b/applications/source/mqtt-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties new file mode 100644 index 00000000..4d3fd58e --- /dev/null +++ b/applications/source/mqtt-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties @@ -0,0 +1,2 @@ +configuration-properties.classes=org.springframework.cloud.fn.supplier.mqtt.MqttSupplierProperties,\ + org.springframework.cloud.fn.common.mqtt.MqttProperties diff --git a/applications/source/mqtt-source/src/test/java/org/springframework/cloud/stream/app/source/mqtt/MqttSourceTests.java b/applications/source/mqtt-source/src/test/java/org/springframework/cloud/stream/app/source/mqtt/MqttSourceTests.java new file mode 100644 index 00000000..801c218f --- /dev/null +++ b/applications/source/mqtt-source/src/test/java/org/springframework/cloud/stream/app/source/mqtt/MqttSourceTests.java @@ -0,0 +1,99 @@ +/* + * Copyright 2020-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.stream.app.source.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.WebApplicationType; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.cloud.fn.supplier.mqtt.MqttSupplierConfiguration; +import org.springframework.cloud.stream.binder.test.OutputDestination; +import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +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; +import org.springframework.messaging.support.MessageBuilder; + +import static org.assertj.core.api.Assertions.assertThat; + +public class MqttSourceTests { + + 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(); + } + + @Test + public void testMqttSource() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration + .getCompleteConfiguration(MqttSourceTestConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--mqtt.supplier.topics=test,fake", "--mqtt.supplier.qos=0,0")) { + + final MessageHandler mqttOutbound = context.getBean("mqttOutbound", MessageHandler.class); + mqttOutbound.handleMessage(MessageBuilder.withPayload("hello").build()); + + OutputDestination target = context.getBean(OutputDestination.class); + Message sourceMessage = target.receive(10000); + + final String actual = new String(sourceMessage.getPayload()); + assertThat(actual).isEqualTo("hello"); + } + } + + @EnableAutoConfiguration + @Import(MqttSupplierConfiguration.class) + public static class MqttSourceTestConfiguration { + + @Autowired + private MqttPahoClientFactory mqttClientFactory; + + @Bean + public MessageHandler mqttOutbound(BeanFactory beanFactory) { + MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler("test", mqttClientFactory); + messageHandler.setAsync(true); + messageHandler.setDefaultTopic("test"); + messageHandler.setConverter(pahoMessageConverter1(beanFactory)); + return messageHandler; + } + + public DefaultPahoMessageConverter pahoMessageConverter1(BeanFactory beanFactory) { + final DefaultPahoMessageConverter pahoMessageConverter = new DefaultPahoMessageConverter(1, true, "UTF-8"); + pahoMessageConverter.setBeanFactory(beanFactory); + return pahoMessageConverter; + } + } +} diff --git a/applications/source/pom.xml b/applications/source/pom.xml index 1359e10b..77ae01d5 100644 --- a/applications/source/pom.xml +++ b/applications/source/pom.xml @@ -13,6 +13,7 @@ file-source jdbc-source http-source + mqtt-source time-source load-generator-source mongodb-source diff --git a/functions/common/mqtt-common/pom.xml b/functions/common/mqtt-common/pom.xml new file mode 100644 index 00000000..5d36d7e9 --- /dev/null +++ b/functions/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/functions/common/mqtt-common/src/main/java/org/springframework/cloud/fn/common/mqtt/MqttConfiguration.java b/functions/common/mqtt-common/src/main/java/org/springframework/cloud/fn/common/mqtt/MqttConfiguration.java new file mode 100644 index 00000000..0335940b --- /dev/null +++ b/functions/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/functions/common/mqtt-common/src/main/java/org/springframework/cloud/fn/common/mqtt/MqttProperties.java b/functions/common/mqtt-common/src/main/java/org/springframework/cloud/fn/common/mqtt/MqttProperties.java new file mode 100644 index 00000000..e8c414f8 --- /dev/null +++ b/functions/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/functions/pom.xml b/functions/pom.xml index ed08a63d..360d1f90 100644 --- a/functions/pom.xml +++ b/functions/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/functions/supplier/mqtt-supplier/pom.xml b/functions/supplier/mqtt-supplier/pom.xml new file mode 100644 index 00000000..1749f5d6 --- /dev/null +++ b/functions/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/functions/supplier/mqtt-supplier/src/main/java/org/springframework/cloud/fn/supplier/mqtt/MqttSupplierConfiguration.java b/functions/supplier/mqtt-supplier/src/main/java/org/springframework/cloud/fn/supplier/mqtt/MqttSupplierConfiguration.java new file mode 100644 index 00000000..c80d08cd --- /dev/null +++ b/functions/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/functions/supplier/mqtt-supplier/src/main/java/org/springframework/cloud/fn/supplier/mqtt/MqttSupplierProperties.java b/functions/supplier/mqtt-supplier/src/main/java/org/springframework/cloud/fn/supplier/mqtt/MqttSupplierProperties.java new file mode 100644 index 00000000..79f3c670 --- /dev/null +++ b/functions/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/functions/supplier/mqtt-supplier/src/test/java/org/springframework/cloud/fn/supplier/mqtt/MqttSupplierTests.java b/functions/supplier/mqtt-supplier/src/test/java/org/springframework/cloud/fn/supplier/mqtt/MqttSupplierTests.java new file mode 100644 index 00000000..3af340ee --- /dev/null +++ b/functions/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"); + } + } +}