MQTT source as supplier/source
Resolves https://github.com/spring-cloud/stream-applications/issues/38
This commit is contained in:
Soby Chacko
31
applications/source/mqtt-source/README.adoc
Normal file
31
applications/source/mqtt-source/README.adoc
Normal file
@@ -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[]
|
||||
84
applications/source/mqtt-source/pom.xml
Normal file
84
applications/source/mqtt-source/pom.xml
Normal file
@@ -0,0 +1,84 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
|
||||
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-source</artifactId>
|
||||
<version>3.0.0-SNAPSHOT</version>
|
||||
<name>mqtt-source</name>
|
||||
<description>mqtt source apps</description>
|
||||
<packaging>jar</packaging>
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud.stream.app</groupId>
|
||||
<artifactId>stream-applications-core</artifactId>
|
||||
<version>3.0.0-SNAPSHOT</version>
|
||||
<relativePath/>
|
||||
</parent>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud.fn</groupId>
|
||||
<artifactId>mqtt-supplier</artifactId>
|
||||
<version>${java-functions.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.testcontainers</groupId>
|
||||
<artifactId>testcontainers</artifactId>
|
||||
<version>1.9.1</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-app-starter-doc-maven-plugin</artifactId>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.springframework.cloud.stream.app.plugin</groupId>
|
||||
<artifactId>spring-cloud-stream-app-maven-plugin</artifactId>
|
||||
<configuration>
|
||||
<generatedApp>
|
||||
<name>mqtt</name>
|
||||
<type>source</type>
|
||||
<version>${project.version}</version>
|
||||
<configClass>org.springframework.cloud.fn.supplier.mqtt.MqttSupplierConfiguration.class</configClass>
|
||||
</generatedApp>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud.fn</groupId>
|
||||
<artifactId>mqtt-supplier</artifactId>
|
||||
<version>${java-functions.version}</version>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</configuration>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
|
||||
<repositories>
|
||||
<repository>
|
||||
<snapshots>
|
||||
<enabled>true</enabled>
|
||||
</snapshots>
|
||||
<id>spring-snapshots</id>
|
||||
<name>Spring Snapshots</name>
|
||||
<url>https://repo.spring.io/libs-snapshot-local</url>
|
||||
</repository>
|
||||
<repository>
|
||||
<snapshots>
|
||||
<enabled>false</enabled>
|
||||
</snapshots>
|
||||
<id>spring-milestones</id>
|
||||
<name>Spring Milestones</name>
|
||||
<url>https://repo.spring.io/libs-milestone-local</url>
|
||||
</repository>
|
||||
</repositories>
|
||||
|
||||
</project>
|
||||
@@ -0,0 +1,2 @@
|
||||
configuration-properties.classes=org.springframework.cloud.fn.supplier.mqtt.MqttSupplierProperties,\
|
||||
org.springframework.cloud.fn.common.mqtt.MqttProperties
|
||||
@@ -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<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();
|
||||
}
|
||||
|
||||
@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<byte[]> 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;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -13,6 +13,7 @@
|
||||
<module>file-source</module>
|
||||
<module>jdbc-source</module>
|
||||
<module>http-source</module>
|
||||
<module>mqtt-source</module>
|
||||
<module>time-source</module>
|
||||
<module>load-generator-source</module>
|
||||
<module>mongodb-source</module>
|
||||
|
||||
36
functions/common/mqtt-common/pom.xml
Normal file
36
functions/common/mqtt-common/pom.xml
Normal file
@@ -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>
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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>
|
||||
|
||||
|
||||
53
functions/supplier/mqtt-supplier/pom.xml
Normal file
53
functions/supplier/mqtt-supplier/pom.xml
Normal file
@@ -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>
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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");
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user