SpringCloud/stream-applications
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

add xmpp-sink application

Browse Source
This commit is contained in:
Daniel Frey
2022-11-04 14:37:48 -04:00
committed by Chris Bono
parent ef216fecbd
commit 1fe892d617
9 changed files with 379 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
README.adoc
View File

@@ -164,6 +164,9 @@ The following are the various components of this repository.
|
|
|link:applications/sink/websocket-sink/README.adoc[Websocket]
|
|
|link:applications/sink/xmpp-sink/README.adoc[XMPP]
|===
=== Build

17
applications/sink/pom.xml
View File

@@ -12,26 +12,27 @@
<modules>
<module>analytics-sink</module>
<module>cassandra-sink</module>
<module>elasticsearch-sink</module>
<module>file-sink</module>
<module>ftp-sink</module>
<module>jdbc-sink</module>
<module>log-sink</module>
<module>mongodb-sink</module>
<module>mqtt-sink</module>
<module>jdbc-sink</module>
<module>redis-sink</module>
<module>pgcopy-sink</module>
<module>rabbit-sink</module>
<module>redis-sink</module>
<module>router-sink</module>
<module>sftp-sink</module>
<module>rsocket-sink</module>
<module>s3-sink</module>
<module>sftp-sink</module>
<module>tcp-sink</module>
<module>throughput-sink</module>
<module>websocket-sink</module>
<module>twitter-update-sink</module>
<module>twitter-message-sink</module>
<module>twitter-update-sink</module>
<module>wavefront-sink</module>
<module>pgcopy-sink</module>
<module>rsocket-sink</module>
<module>elasticsearch-sink</module>
<module>websocket-sink</module>
<module>xmpp-sink</module>
<module>zeromq-sink</module>
</modules>
<build>

58
applications/sink/xmpp-sink/README.adoc Normal file
View File

@@ -0,0 +1,58 @@
//tag::ref-doc[]
= ZeroMQ Sink
The "zeromq" sink enables sending messages to a ZeroMQ socket.
== Input
* `byte[]`
== Output
=== Payload
N/A
== Options
The **$$zeromq$$** $$sink$$ has the following options:
//tag::configuration-properties[]
Properties grouped by prefix:
=== xmpp.consumer
$$chat-to$$:: $$XMPP handle to send message to.$$ *($$String$$, default: `$$<none>$$`)*
=== 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>$$`)*
//end::configuration-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 zeromq-sink.jar --zeromq.consumer.connectUrl=tcp://server:port --zeromq.consumer.topic=
```
//end::ref-doc[]

85
applications/sink/xmpp-sink/pom.xml Normal file
View 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-sink</artifactId>
<name>xmpp-sink</name>
<description>XMPP sink apps</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-consumer</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>sink</type>
<version>${project.version}</version>
<configClass>org.springframework.cloud.fn.consumer.xmpp.XmppConsumerConfiguration.class
</configClass>
<maven>
<dependencies>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>xmpp-consumer</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>

1
applications/sink/xmpp-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties Normal file
View File

@@ -0,0 +1 @@
configuration-properties.classes=org.springframework.cloud.fn.consumer.xmpp.XmppConsumerProperties,org.springframework.cloud.fn.common.xmpp.XmppConnectionFactoryProperties

1
applications/sink/xmpp-sink/src/main/resources/META-INF/dataflow-configuration-metadata.properties Normal file
View File

@@ -0,0 +1 @@
configuration-properties.classes=org.springframework.cloud.fn.consumer.xmpp.XmppConsumerProperties,org.springframework.cloud.fn.common.xmpp.XmppConnectionFactoryProperties

95
applications/sink/xmpp-sink/src/test/java/org/springframework/cloud/stream/app/sink/xmpp/XmppSinkInvalidConfigTests.java Normal file
View File

@@ -0,0 +1,95 @@
/*
* Copyright 2014-2022 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.sink.xmpp;
import org.junit.jupiter.api.Test;
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.consumer.xmpp.XmppConsumerProperties;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Configuration;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
/**
* Tests for XMPP Sink with invalid config.
*
* @author Daniel Frey
* @since 4.0.0
*/
public class XmppSinkInvalidConfigTests {
@Test
public void testEmptyFactoryHost() {
assertThatExceptionOfType(BeanCreationException.class)
.isThrownBy(() -> {
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
TestPropertyValues.of("xmpp.factory.host: ").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 'host': rejected value []");
}
@Test
public void testEmptyFactoryUser() {
assertThatExceptionOfType(BeanCreationException.class)
.isThrownBy(() -> {
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
TestPropertyValues.of("xmpp.factory.user: ").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 'user': rejected value []");
}
@Test
public void testEmptyFactoryPassword() {
assertThatExceptionOfType(BeanCreationException.class)
.isThrownBy(() -> {
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
TestPropertyValues.of("xmpp.factory.password: ").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 'password': rejected value []");
}
@Configuration
@EnableConfigurationProperties({XmppConnectionFactoryProperties.class, XmppConsumerProperties.class})
static class Config {
}
}

126
applications/sink/xmpp-sink/src/test/java/org/springframework/cloud/stream/app/sink/xmpp/XmppSinkTests.java Normal file
View File

@@ -0,0 +1,126 @@
/*
* Copyright 2014-2022 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.sink.xmpp;
import java.io.IOException;
import java.time.Duration;
import org.jivesoftware.smack.ConnectionConfiguration;
import org.jivesoftware.smack.SmackException;
import org.jivesoftware.smack.XMPPException;
import org.jivesoftware.smack.filter.StanzaTypeFilter;
import org.jivesoftware.smack.packet.Stanza;
import org.jivesoftware.smack.tcp.XMPPTCPConnection;
import org.jivesoftware.smack.tcp.XMPPTCPConnectionConfiguration;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.cloud.fn.consumer.xmpp.XmppConsumerConfiguration;
import org.springframework.cloud.fn.test.support.xmpp.XmppTestContainerSupport;
import org.springframework.cloud.stream.binder.test.InputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Import;
import org.springframework.integration.xmpp.XmppHeaders;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
/**
* Tests for XMPP Sink.
*
* @author Daniel Frey
*
* @since 4.0.0
*/
public class XmppSinkTests implements XmppTestContainerSupport {
private XMPPTCPConnection clientConnection;
@BeforeEach
void setup() throws IOException, SmackException, XMPPException, InterruptedException {
var builder = XMPPTCPConnectionConfiguration.builder();
builder.setSecurityMode(ConnectionConfiguration.SecurityMode.disabled);
builder.setHost(XmppTestContainerSupport.getXmppHost());
builder.setPort(XmppTestContainerSupport.getXmppMappedPort());
builder.setResource(SERVICE_NAME);
builder.setUsernameAndPassword(JANE_USER, USER_PW)
.setXmppDomain(SERVICE_NAME);
this.clientConnection = new XMPPTCPConnection(builder.build());
this.clientConnection.connect();
this.clientConnection.login();
}
@Test
public void testSinkFromFunction() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(XmppSinkTestApplication.class)).run(
"--spring.cloud.function.definition=xmppConsumer",
"--xmpp.factory.host=" + XmppTestContainerSupport.getXmppHost(),
"--xmpp.factory.port=" + XmppTestContainerSupport.getXmppMappedPort(),
"--xmpp.factory.user=" + JOHN_USER,
"--xmpp.factory.password=" + USER_PW,
"--xmpp.factory.service-name=" + SERVICE_NAME,
"--xmpp.factory.security-mode=disabled"
)) {
var inputDestination = context.getBean(InputDestination.class);
var collector
= this.clientConnection.createStanzaCollector(StanzaTypeFilter.MESSAGE);
var testMessage = MessageBuilder.withPayload("test")
.setHeader(XmppHeaders.TO, JANE_USER + "@" + SERVICE_NAME)
.build();
await().atMost(Duration.ofSeconds(20)).pollDelay(Duration.ofMillis(100))
.untilAsserted(() -> {
inputDestination.send(testMessage);
Stanza stanza = collector.nextResult();
assertStanza(stanza);
});
}
}
private void assertStanza(Stanza stanza) {
assertTo(stanza);
assertFrom(stanza);
}
private void assertTo(Stanza stanza) {
assertThat(stanza.getTo().asBareJid().asUnescapedString()).isEqualTo(JANE_USER + "@" + SERVICE_NAME);
}
private void assertFrom(Stanza stanza) {
assertThat(stanza.getFrom().asBareJid().asUnescapedString()).isEqualTo(JOHN_USER + "@" + SERVICE_NAME);
}
@SpringBootConfiguration
@EnableAutoConfiguration
@Import(XmppConsumerConfiguration.class)
static class XmppSinkTestApplication {
}
}

2
functions/common/xmpp-common/src/main/java/org/springframework/cloud/fn/common/xmpp/XmppConnectionFactoryProperties.java
View File

@@ -89,7 +89,7 @@ public class XmppConnectionFactoryProperties {
this.password = password;
}
@NotEmpty(message = "port is required")
@NotEmpty(message = "password is required")
public String getPassword() {
return password;
}