Spring/spring-integration-samples
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add MQTT Sample

Browse Source 
* Move `ftp` `log4j` dependency to the `testCompile` configuration
* Regenerate POM for MQTT sample
This commit is contained in:
Gary Russell
2016-03-05 11:17:07 -05:00
committed by Artem Bilan
parent 55b2cfb49f
commit b52ee9ed9c
7 changed files with 346 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
README.md
View File

@@ -43,7 +43,7 @@ This directory holds demos/samples for Spring Integration 4.0 Java Configuration
This is a good place to get started. The samples here are technically motivated and demonstrate the bare minimum with regard to configuration and code to help you to get introduced to the basic concepts, API and configuration of Spring Integration. For example, if you are looking for an answer on how to wire a **Service Activator** to a **Channel** or how to apply a **Gateway** to your message exchange or how to get started with using the **MAIL** or **XML** module, this would be the right place to find a relevant sample. The bottom line is that this is a good starting point.
* **amqp** - Demonstrates the functionality of the various **AMQP Adapters**
* **barrier** - Shows how to suspend a thread until some asynchronous event occurs
* **barrier** - Shows how to suspend a thread until some asynchronous event occurs
* **control-bus** - Demonstrates the functionality of the **Control Bus**
* **enricher** - This sample demonstrates how the Enricher components can be used
* **feed** - Demonstrates the functionality of the **Feed Adapter** (RSS/ATOM)
@@ -56,6 +56,7 @@ This is a good place to get started. The samples here are technically motivated
* **jmx** - Demonstrates **JMX** support using a **JMX Attribute Polling Channel** and **JMX Operation Invoking Channel Adapter**
* **jpa** - Shows the usage of the JPA Components can be used
* **mail** - Example showing **IMAP** and **POP3** support
* **mqtt** - Demonstrates the functionality of inbound and outbound **MQTT Adapters**
* **mongodb** - Shows how to persist a Message payload to a **MongoDb** document store and how to read documents from **MongoDb**
* **oddeven** - Example combining the functionality of **Inbound Channel Adapter**, **Filter**, **Router** and **Poller**
* **jpa** - This sample illustrates how the JPA Components can be used

29
basic/mqtt/README.md Normal file
View File

@@ -0,0 +1,29 @@
Spring Integration - MQTT Sample
================================
# Overview
This sample demonstrates basic functionality of the **Spring Integration MQTT Adapters**.
It assumes a broker is running on localhost on port 1883.
Once the application is started, you enter some text on the command prompt and a message containing that entered text is
dispatched to the MQTT topic. In return that message is retrieved by Spring Integration and then logged.
# How to Run the Sample
If you imported the example into your IDE, you can just run class **org.springframework.integration.samples.mqtt.Application**.
For example in [SpringSource Tool Suite](http://www.springsource.com/developer/sts) (STS) do:
* Right-click on SampleSimple class --> Run As --> Spring Boot App
(or run from the boot console).
Alternatively, you can start the sample from the command line:
* ./gradlew :mqtt:run
Enter some data (e.g. `foo`) on the console; you will see `foo sent to MQTT, received from MQTT`
Ctrl-C to terminate.

123
basic/mqtt/pom.xml Normal file
View File

@@ -0,0 +1,123 @@
<?xml version="1.0" encoding="UTF-8"?>
<project 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" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>1.4.0.M1</version>
</parent>
<groupId>org.springframework.integration.samples</groupId>
<artifactId>mqtt</artifactId>
<version>4.3.0.BUILD-SNAPSHOT</version>
<name>MQTT Basic Sample</name>
<description>MQTT Basic Sample</description>
<url>http://projects.spring.io/spring-integration</url>
<organization>
<name>SpringIO</name>
<url>https://spring.io</url>
</organization>
<licenses>
<license>
<name>The Apache Software License, Version 2.0</name>
<url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
<distribution>repo</distribution>
</license>
</licenses>
<developers>
<developer>
<id>garyrussell</id>
<name>Gary Russell</name>
<email>grussell@pivotal.io</email>
<roles>
<role>project lead</role>
</roles>
</developer>
<developer>
<id>markfisher</id>
<name>Mark Fisher</name>
<email>mfisher@pivotal.io</email>
<roles>
<role>project founder and lead emeritus</role>
</roles>
</developer>
<developer>
<id>ghillert</id>
<name>Gunnar Hillert</name>
<email>ghillert@pivotal.io</email>
</developer>
<developer>
<id>abilan</id>
<name>Artem Bilan</name>
<email>abilan@pivotal.io</email>
</developer>
</developers>
<scm>
<connection>scm:git:scm:git:git://github.com/spring-projects/spring-integration-samples.git</connection>
<developerConnection>scm:git:scm:git:ssh://git@github.com:spring-projects/spring-integration-samples.git</developerConnection>
<url>https://github.com/spring-projects/spring-integration-samples</url>
</scm>
<dependencies>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.11</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.hamcrest</groupId>
<artifactId>hamcrest-all</artifactId>
<version>1.3</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-integration</artifactId>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-java-dsl</artifactId>
<version>1.1.2.RELEASE</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-stream</artifactId>
<version>4.3.0.M1</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
<version>4.2.3.RELEASE</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-mqtt</artifactId>
<version>4.3.0.M1</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<version>1.9.5</version>
<scope>test</scope>
</dependency>
</dependencies>
<repositories>
<repository>
<id>repo.spring.io.milestone</id>
<name>Spring Framework Maven Milestone Repository</name>
<url>https://repo.spring.io/libs-milestone</url>
</repository>
</repositories>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>

133
basic/mqtt/src/main/java/org/springframework/integration/samples/mqtt/Application.java Normal file
View File

@@ -0,0 +1,133 @@
/*
* Copyright 2016 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
*
* http://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.integration.samples.mqtt;
import org.apache.log4j.Logger;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.core.Pollers;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.handler.LoggingHandler;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.integration.stream.CharacterStreamReadingMessageSource;
import org.springframework.messaging.MessageHandler;
/**
* Starts the Spring Context and will initialize the Spring Integration message flow.
*
* @author Gary Russell
*
*/
@SpringBootApplication
@IntegrationComponentScan
public class Application {
private static final Logger LOGGER = Logger.getLogger(Application.class);
/**
* Load the Spring Integration Application Context
*
* @param args - command line arguments
*/
public static void main(final String... args) {
LOGGER.info("\n========================================================="
+ "\n "
+ "\n Welcome to Spring Integration! "
+ "\n "
+ "\n For more information please visit: "
+ "\n https://spring.io/projects/spring-integration "
+ "\n "
+ "\n=========================================================" );
LOGGER.info("\n========================================================="
+ "\n "
+ "\n This is the MQTT Sample - "
+ "\n "
+ "\n Please enter some text and press return. The entered "
+ "\n Message will be sent to the configured MQTT topic, "
+ "\n then again immediately retrieved from the Message "
+ "\n Broker and ultimately printed to the command line. "
+ "\n "
+ "\n=========================================================" );
SpringApplication.run(Application.class, args);
}
@Bean
public MqttPahoClientFactory mqttClientFactory() {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
factory.setServerURIs("tcp://localhost:1883");
factory.setUserName("guest");
factory.setPassword("guest");
return factory;
}
// publisher
@Bean
public IntegrationFlow mqttOutFlow() {
return IntegrationFlows.from(CharacterStreamReadingMessageSource.stdin(),
e -> e.poller(Pollers.fixedDelay(1000)))
.transform(p -> p + " sent to MQTT")
.handle(mqttOutbound())
.get();
}
@Bean
public MessageHandler mqttOutbound() {
MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler("siSamplePublisher", mqttClientFactory());
messageHandler.setAsync(true);
messageHandler.setDefaultTopic("siSampleTopic");
return messageHandler;
}
// consumer
@Bean
public IntegrationFlow mqttInFlow() {
return IntegrationFlows.from(mqttInbound())
.transform(p -> p + ", received from MQTT")
.handle(logger())
.get();
}
private LoggingHandler logger() {
LoggingHandler loggingHandler = new LoggingHandler("INFO");
loggingHandler.setLoggerName("siSample");
return loggingHandler;
}
@Bean
public MessageProducerSupport mqttInbound() {
MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("siSampleConsumer",
mqttClientFactory(), "siSampleTopic");
adapter.setCompletionTimeout(5000);
adapter.setConverter(new DefaultPahoMessageConverter());
adapter.setQos(1);
return adapter;
}
}

24
basic/mqtt/src/main/resources/logback.xml Normal file
View File

@@ -0,0 +1,24 @@
<configuration>
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<!-- encoders are assigned the type ch.qos.logback.classic.encoder.PatternLayoutEncoder
by default -->
<encoder>
<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n
</pattern>
</encoder>
</appender>
<root level="debug">
<appender-ref ref="STDOUT" />
</root>
<logger name="org.springframework" level="WARN" />
<logger name="org.springframework.integration" level="ERROR" />
<logger name="org.springframework.integration.samples" level="INFO" />
<logger name="siSample" level="INFO" />
</configuration>

36
build.gradle
View File

@@ -192,8 +192,8 @@ subprojects { subproject ->
postgresVersion = '9.1-901-1.jdbc4'
subethasmtpVersion = '1.2'
slf4jVersion = '1.7.11'
springIntegrationVersion = '4.2.2.RELEASE'
springIntegrationDslVersion = '1.1.1.RELEASE'
springIntegrationVersion = '4.3.0.M1'
springIntegrationDslVersion = '1.1.2.RELEASE'
springIntegrationKafkaVersion = '1.3.0.RELEASE'
springIntegrationSplunkVersion = '1.1.0.RELEASE'
springVersion = '4.2.3.RELEASE'
@@ -517,9 +517,9 @@ project('ftp') {
compile "org.apache.ftpserver:ftpserver-core:$ftpServerVersion"
compile "org.slf4j:slf4j-api:$slf4jVersion"
runtime "log4j:log4j:$log4jVersion"
runtime "org.slf4j:slf4j-log4j12:$slf4jVersion"
testCompile "log4j:log4j:$log4jVersion"
testCompile "org.springframework.integration:spring-integration-test:$springIntegrationVersion"
}
@@ -615,6 +615,36 @@ project('kafka') {
classpath = sourceSets.main.runtimeClasspath
}
}
project('mqtt') {
description = 'MQTT Basic Sample'
apply plugin: 'spring-boot'
sourceCompatibility = 1.8
dependencies {
compile 'org.springframework.boot:spring-boot-starter-integration'
compile "org.springframework.integration:spring-integration-stream:$springIntegrationVersion"
compile "org.springframework.integration:spring-integration-mqtt:$springIntegrationVersion"
compile "org.springframework.integration:spring-integration-java-dsl:$springIntegrationDslVersion"
}
springBoot {
mainClass = 'org.springframework.integration.samples.mqtt.Application'
}
task run(type: JavaExec) {
main 'org.springframework.integration.samples.mqtt.Application'
classpath = sourceSets.main.runtimeClasspath
}
tasks.withType(JavaExec) {
standardInput = System.in
}
}
project('si4demo') {
description = 'Java Configuration/DSL Sample'

4
gradle.properties
View File

@@ -1,2 +1,2 @@
version=4.1.0.BUILD-SNAPSHOT
springBootVersion=1.3.0.RELEASE
version=4.3.0.BUILD-SNAPSHOT
springBootVersion=1.4.0.M1