Add Xmpp source app (#413)
This commit is contained in:
Daniel Frey
@@ -158,7 +158,7 @@ The following are the various components of this repository.
|
||||
|link:applications/source/websocket-source/README.adoc[Websocket]
|
||||
|
|
||||
|link:applications/sink/twitter-update-sink/README.adoc[Twitter Update]
|
||||
|
|
||||
|link:applications/source/xmpp-source/README.adoc[XMPP]
|
||||
|
|
||||
|link:applications/sink/wavefront-sink/README.adoc[Wavefront]
|
||||
|
|
||||
|
||||
@@ -10,26 +10,27 @@
|
||||
<packaging>pom</packaging>
|
||||
|
||||
<modules>
|
||||
<module>cdc-debezium-source</module>
|
||||
<module>file-source</module>
|
||||
<module>ftp-source</module>
|
||||
<module>jdbc-source</module>
|
||||
<module>jms-source</module>
|
||||
<module>http-source</module>
|
||||
<module>mqtt-source</module>
|
||||
<module>time-source</module>
|
||||
<module>load-generator-source</module>
|
||||
<module>mail-source</module>
|
||||
<module>mongodb-source</module>
|
||||
<module>tcp-source</module>
|
||||
<module>mqtt-source</module>
|
||||
<module>rabbit-source</module>
|
||||
<module>s3-source</module>
|
||||
<module>sftp-source</module>
|
||||
<module>twitter-stream-source</module>
|
||||
<module>twitter-search-source</module>
|
||||
<module>twitter-message-source</module>
|
||||
<module>websocket-source</module>
|
||||
<module>cdc-debezium-source</module>
|
||||
<module>syslog-source</module>
|
||||
<module>mail-source</module>
|
||||
<module>tcp-source</module>
|
||||
<module>time-source</module>
|
||||
<module>twitter-message-source</module>
|
||||
<module>twitter-search-source</module>
|
||||
<module>twitter-stream-source</module>
|
||||
<module>websocket-source</module>
|
||||
<module>xmpp-source</module>
|
||||
<module>zeromq-source</module>
|
||||
</modules>
|
||||
<build>
|
||||
|
||||
62
applications/source/xmpp-source/README.adoc
Normal file
62
applications/source/xmpp-source/README.adoc
Normal file
@@ -0,0 +1,62 @@
|
||||
//tag::ref-doc[]
|
||||
= XMPP Source
|
||||
|
||||
The "xmpp" source enables receiving messages from an XMPP Server.
|
||||
|
||||
== Input
|
||||
|
||||
N/A
|
||||
|
||||
== Output
|
||||
|
||||
=== Payload
|
||||
|
||||
* `byte[]`
|
||||
|
||||
== Options
|
||||
|
||||
The **$$xmpp$$** $$source$$ has the following options:
|
||||
|
||||
//tag::configuration-properties[]
|
||||
Properties grouped by prefix:
|
||||
|
||||
|
||||
=== xmpp.factory
|
||||
|
||||
$$host$$:: $$XMPP Host server to connect to.$$ *($$String$$, default: `$$<none>$$`)*
|
||||
$$password$$:: $$The Password for the connected user.$$ *($$String$$, default: `$$<none>$$`)*
|
||||
$$port$$:: $$Port for connecting to the host. - Default Client Port: 5222$$ *($$Integer$$, default: `$$5222$$`)*
|
||||
$$resource$$:: $$The Resource to bind to on the XMPP Host. - Can be empty, server will generate one if not set$$ *($$String$$, default: `$$<none>$$`)*
|
||||
$$security-mode$$:: $$<documentation missing>$$ *($$SecurityMode$$, default: `$$<none>$$`, possible values: `required`,`ifpossible`,`disabled`)*
|
||||
$$service-name$$:: $$The Service Name to set for the XMPP Domain.$$ *($$String$$, default: `$$<none>$$`)*
|
||||
$$subscription-mode$$:: $$<documentation missing>$$ *($$SubscriptionMode$$, default: `$$<none>$$`, possible values: `accept_all`,`reject_all`,`manual`)*
|
||||
$$user$$:: $$The User the connection should connect as.$$ *($$String$$, default: `$$<none>$$`)*
|
||||
|
||||
=== xmpp.supplier
|
||||
|
||||
$$payload-expression$$:: $$<documentation missing>$$ *($$Expression$$, default: `$$<none>$$`)*
|
||||
$$stanza-filter$$:: $$<documentation missing>$$ *($$StanzaFilter$$, default: `$$<none>$$`)*
|
||||
//end::configuration-properties[]
|
||||
|
||||
Also see the https://docs.spring.io/spring-boot/docs/current/reference/html/common-application-properties.html[Spring Boot Documentation]
|
||||
for addition properties for the broker connections and listener properties.
|
||||
|
||||
== Build
|
||||
|
||||
```
|
||||
$ ./mvnw clean install -PgenerateApps
|
||||
$ cd apps
|
||||
```
|
||||
You can find the corresponding binder based projects here.
|
||||
You can then cd into one of the folders and build it:
|
||||
```
|
||||
$ ./mvnw clean package
|
||||
```
|
||||
|
||||
== Examples
|
||||
|
||||
```
|
||||
java -jar xmpp-source.jar --xmpp.factory.host=localhost --xmpp.factory.port=5222 --xmpp.factory.user=jane --xmpp.factory.password=secret --xmpp.factory.service-name=localhost
|
||||
```
|
||||
|
||||
//end::ref-doc[]
|
||||
85
applications/source/xmpp-source/pom.xml
Normal file
85
applications/source/xmpp-source/pom.xml
Normal file
@@ -0,0 +1,85 @@
|
||||
<?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>xmpp-source</artifactId>
|
||||
<name>xmpp-source</name>
|
||||
<description>XMPP source app</description>
|
||||
<packaging>jar</packaging>
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud.stream.app</groupId>
|
||||
<artifactId>stream-applications-core</artifactId>
|
||||
<version>4.0.0-SNAPSHOT</version>
|
||||
<relativePath>../../stream-applications-core/pom.xml</relativePath>
|
||||
</parent>
|
||||
|
||||
<dependencies>
|
||||
|
||||
<!-- Spring Cloud Function dependencies -->
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud.fn</groupId>
|
||||
<artifactId>xmpp-supplier</artifactId>
|
||||
</dependency>
|
||||
|
||||
<!-- Test dependencies -->
|
||||
<dependency>
|
||||
<groupId>io.projectreactor</groupId>
|
||||
<artifactId>reactor-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.awaitility</groupId>
|
||||
<artifactId>awaitility</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud.fn</groupId>
|
||||
<artifactId>function-test-support</artifactId>
|
||||
<version>${project.version}</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-dataflow-apps-docs-plugin</artifactId>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-dataflow-apps-generator-plugin</artifactId>
|
||||
<configuration>
|
||||
<application>
|
||||
<name>xmpp</name>
|
||||
<type>source</type>
|
||||
<version>${project.version}</version>
|
||||
<configClass>org.springframework.cloud.fn.supplier.xmpp.XmppSupplierConfiguration.class
|
||||
</configClass>
|
||||
|
||||
<maven>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud.fn</groupId>
|
||||
<artifactId>xmpp-supplier</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud.stream.app</groupId>
|
||||
<artifactId>stream-applications-composite-function-support</artifactId>
|
||||
<version>${stream-apps-core.version}</version>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</maven>
|
||||
</application>
|
||||
|
||||
</configuration>
|
||||
</plugin>
|
||||
|
||||
</plugins>
|
||||
</build>
|
||||
|
||||
</project>
|
||||
@@ -0,0 +1 @@
|
||||
configuration-properties.classes=org.springframework.cloud.fn.supplier.xmpp.XmppSupplierProperties,org.springframework.cloud.fn.common.xmpp.XmppConnectionFactoryProperties
|
||||
@@ -0,0 +1 @@
|
||||
configuration-properties.classes=org.springframework.cloud.fn.supplier.xmpp.XmppSupplierProperties,org.springframework.cloud.fn.common.xmpp.XmppConnectionFactoryProperties
|
||||
@@ -0,0 +1,61 @@
|
||||
/*
|
||||
* Copyright 2016-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.xmpp;
|
||||
|
||||
import org.junit.jupiter.params.ParameterizedTest;
|
||||
import org.junit.jupiter.params.provider.ValueSource;
|
||||
|
||||
import org.springframework.beans.factory.BeanCreationException;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.boot.test.util.TestPropertyValues;
|
||||
import org.springframework.cloud.fn.common.xmpp.XmppConnectionFactoryProperties;
|
||||
import org.springframework.cloud.fn.supplier.xmpp.XmppSupplierProperties;
|
||||
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
|
||||
|
||||
/**
|
||||
* Tests for XmppSource with invalid config.
|
||||
*
|
||||
* @author Daniel Frey
|
||||
* @since 4.0.0
|
||||
*/
|
||||
public class XmppSourceInvalidConfigTests {
|
||||
|
||||
@ParameterizedTest
|
||||
@ValueSource(strings = { "host", "user", "password" })
|
||||
void requiredXmppConnectionFactoryPropertyIsSetEmpty(String propertyName) {
|
||||
assertThatExceptionOfType(BeanCreationException.class)
|
||||
.isThrownBy(() -> {
|
||||
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
|
||||
TestPropertyValues.of(String.format("xmpp.factory.%s: ", propertyName)).applyTo(context);
|
||||
context.register(Config.class);
|
||||
context.refresh();
|
||||
})
|
||||
.withMessageContaining("Error creating bean with name 'xmpp.factory-org.springframework.cloud.fn.common.xmpp.XmppConnectionFactoryProperties': Could not bind properties to 'XmppConnectionFactoryProperties'")
|
||||
.havingRootCause()
|
||||
.withMessageContaining("Binding validation errors on xmpp.factory")
|
||||
.withMessageContaining("Field error in object 'xmpp.factory' on field '%s': rejected value []", propertyName);
|
||||
}
|
||||
|
||||
@Configuration
|
||||
@EnableConfigurationProperties({XmppConnectionFactoryProperties.class, XmppSupplierProperties.class})
|
||||
static class Config {
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,122 @@
|
||||
/*
|
||||
* 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.xmpp;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
|
||||
import org.assertj.core.api.InstanceOfAssertFactories;
|
||||
import org.jivesoftware.smack.ConnectionConfiguration;
|
||||
import org.jivesoftware.smack.SmackException;
|
||||
import org.jivesoftware.smack.chat2.ChatManager;
|
||||
import org.jivesoftware.smack.tcp.XMPPTCPConnection;
|
||||
import org.jivesoftware.smack.tcp.XMPPTCPConnectionConfiguration;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.jxmpp.jid.impl.JidCreate;
|
||||
import org.jxmpp.stringprep.XmppStringprepException;
|
||||
|
||||
import org.springframework.boot.SpringBootConfiguration;
|
||||
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
|
||||
import org.springframework.boot.builder.SpringApplicationBuilder;
|
||||
import org.springframework.cloud.fn.supplier.xmpp.XmppSupplierConfiguration;
|
||||
import org.springframework.cloud.fn.test.support.xmpp.XmppTestContainerSupport;
|
||||
import org.springframework.cloud.stream.binder.test.OutputDestination;
|
||||
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
|
||||
import org.springframework.cloud.stream.config.BindingServiceConfiguration;
|
||||
import org.springframework.context.ConfigurableApplicationContext;
|
||||
import org.springframework.context.annotation.Import;
|
||||
import org.springframework.integration.xmpp.XmppHeaders;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/**
|
||||
* Tests for XmppSource.
|
||||
*
|
||||
* @author Daniel Frey
|
||||
* @since 4.0.0
|
||||
*/
|
||||
public class XmppSourceTests implements XmppTestContainerSupport {
|
||||
|
||||
private XMPPTCPConnection sourceConnection;
|
||||
|
||||
@BeforeEach
|
||||
void prepareForTest() throws Exception {
|
||||
|
||||
var builder = XMPPTCPConnectionConfiguration.builder();
|
||||
builder.setSecurityMode(ConnectionConfiguration.SecurityMode.disabled);
|
||||
builder.setHost(XmppTestContainerSupport.getXmppHost());
|
||||
builder.setPort(XmppTestContainerSupport.getXmppMappedPort());
|
||||
builder.setResource(SERVICE_NAME);
|
||||
builder.setUsernameAndPassword(JOHN_USER, USER_PW) // Connect as user intended to send messages from
|
||||
.setXmppDomain(SERVICE_NAME);
|
||||
this.sourceConnection = new XMPPTCPConnection(builder.build());
|
||||
this.sourceConnection.connect();
|
||||
this.sourceConnection.login();
|
||||
|
||||
}
|
||||
|
||||
@AfterEach
|
||||
void teardown() {
|
||||
this.sourceConnection.instantShutdown();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void sourceFromSupplier() throws InterruptedException {
|
||||
|
||||
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
|
||||
TestChannelBinderConfiguration.getCompleteConfiguration(XmppSourceTestApplication.class)).run(
|
||||
"--spring.cloud.function.definition=xmppSupplier",
|
||||
"--xmpp.factory.host=" + XmppTestContainerSupport.getXmppHost(),
|
||||
"--xmpp.factory.port=" + XmppTestContainerSupport.getXmppMappedPort(),
|
||||
"--xmpp.factory.user=" + JANE_USER,
|
||||
"--xmpp.factory.password=" + USER_PW,
|
||||
"--xmpp.factory.service-name=" + SERVICE_NAME,
|
||||
"--xmpp.factory.security-mode=disabled"
|
||||
)) {
|
||||
|
||||
var outputDestination = context.getBean(OutputDestination.class);
|
||||
|
||||
var payload = "test";
|
||||
|
||||
var chatManager = ChatManager.getInstanceFor(this.sourceConnection);
|
||||
var jid = JidCreate.entityBareFrom(JANE_USER + "@" + SERVICE_NAME);
|
||||
var chat = chatManager.chatWith(jid);
|
||||
chat.send(payload);
|
||||
|
||||
var message = outputDestination.receive(10000, "xmppSupplier-out-0");
|
||||
|
||||
assertThat(message.getPayload())
|
||||
.asInstanceOf(InstanceOfAssertFactories.type(byte[].class))
|
||||
.isEqualTo(payload.getBytes(StandardCharsets.UTF_8));
|
||||
|
||||
assertThat(message.getHeaders().containsKey(XmppHeaders.TO)).isTrue();
|
||||
assertThat(message.getHeaders().get(XmppHeaders.TO, String.class)).isEqualTo(JANE_USER + "@" + SERVICE_NAME);
|
||||
|
||||
}
|
||||
catch (SmackException.NotConnectedException | XmppStringprepException e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@SpringBootConfiguration
|
||||
@EnableAutoConfiguration
|
||||
@Import({XmppSupplierConfiguration.class, TestChannelBinderConfiguration.class, BindingServiceConfiguration.class})
|
||||
public static class XmppSourceTestApplication { }
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user