diff --git a/README.adoc b/README.adoc index 5c34864d..77bec007 100644 --- a/README.adoc +++ b/README.adoc @@ -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 diff --git a/applications/sink/pom.xml b/applications/sink/pom.xml index df36a0c7..fda6db90 100644 --- a/applications/sink/pom.xml +++ b/applications/sink/pom.xml @@ -12,26 +12,27 @@ analytics-sink cassandra-sink + elasticsearch-sink file-sink ftp-sink + jdbc-sink log-sink mongodb-sink mqtt-sink - jdbc-sink - redis-sink + pgcopy-sink rabbit-sink + redis-sink router-sink - sftp-sink + rsocket-sink s3-sink + sftp-sink tcp-sink throughput-sink - websocket-sink - twitter-update-sink twitter-message-sink + twitter-update-sink wavefront-sink - pgcopy-sink - rsocket-sink - elasticsearch-sink + websocket-sink + xmpp-sink zeromq-sink diff --git a/applications/sink/xmpp-sink/README.adoc b/applications/sink/xmpp-sink/README.adoc new file mode 100644 index 00000000..af29bd60 --- /dev/null +++ b/applications/sink/xmpp-sink/README.adoc @@ -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: `$$$$`)* + +=== xmpp.factory + +$$host$$:: $$XMPP Host server to connect to.$$ *($$String$$, default: `$$$$`)* +$$password$$:: $$The Password for the connected user.$$ *($$String$$, default: `$$$$`)* +$$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: `$$$$`)* +$$security-mode$$:: $$$$ *($$SecurityMode$$, default: `$$$$`, possible values: `required`,`ifpossible`,`disabled`)* +$$service-name$$:: $$The Service Name to set for the XMPP Domain.$$ *($$String$$, default: `$$$$`)* +$$subscription-mode$$:: $$$$ *($$SubscriptionMode$$, default: `$$$$`, possible values: `accept_all`,`reject_all`,`manual`)* +$$user$$:: $$The User the connection should connect as.$$ *($$String$$, default: `$$$$`)* +//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[] diff --git a/applications/sink/xmpp-sink/pom.xml b/applications/sink/xmpp-sink/pom.xml new file mode 100644 index 00000000..6ffb2ace --- /dev/null +++ b/applications/sink/xmpp-sink/pom.xml @@ -0,0 +1,85 @@ + + + 4.0.0 + + xmpp-sink + xmpp-sink + XMPP sink apps + jar + + + org.springframework.cloud.stream.app + stream-applications-core + 4.0.0-SNAPSHOT + ../../stream-applications-core/pom.xml + + + + + + + org.springframework.cloud.fn + xmpp-consumer + + + + + io.projectreactor + reactor-test + test + + + org.awaitility + awaitility + test + + + + org.springframework.cloud.fn + function-test-support + ${project.version} + test + + + + + + + + org.springframework.cloud + spring-cloud-dataflow-apps-docs-plugin + + + org.springframework.cloud + spring-cloud-dataflow-apps-generator-plugin + + + xmpp + sink + ${project.version} + org.springframework.cloud.fn.consumer.xmpp.XmppConsumerConfiguration.class + + + + + + org.springframework.cloud.fn + xmpp-consumer + + + org.springframework.cloud.stream.app + stream-applications-composite-function-support + ${stream-apps-core.version} + + + + + + + + + + + + diff --git a/applications/sink/xmpp-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties b/applications/sink/xmpp-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties new file mode 100644 index 00000000..880770fa --- /dev/null +++ b/applications/sink/xmpp-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties @@ -0,0 +1 @@ +configuration-properties.classes=org.springframework.cloud.fn.consumer.xmpp.XmppConsumerProperties,org.springframework.cloud.fn.common.xmpp.XmppConnectionFactoryProperties diff --git a/applications/sink/xmpp-sink/src/main/resources/META-INF/dataflow-configuration-metadata.properties b/applications/sink/xmpp-sink/src/main/resources/META-INF/dataflow-configuration-metadata.properties new file mode 100644 index 00000000..880770fa --- /dev/null +++ b/applications/sink/xmpp-sink/src/main/resources/META-INF/dataflow-configuration-metadata.properties @@ -0,0 +1 @@ +configuration-properties.classes=org.springframework.cloud.fn.consumer.xmpp.XmppConsumerProperties,org.springframework.cloud.fn.common.xmpp.XmppConnectionFactoryProperties diff --git a/applications/sink/xmpp-sink/src/test/java/org/springframework/cloud/stream/app/sink/xmpp/XmppSinkInvalidConfigTests.java b/applications/sink/xmpp-sink/src/test/java/org/springframework/cloud/stream/app/sink/xmpp/XmppSinkInvalidConfigTests.java new file mode 100644 index 00000000..52f5ba3c --- /dev/null +++ b/applications/sink/xmpp-sink/src/test/java/org/springframework/cloud/stream/app/sink/xmpp/XmppSinkInvalidConfigTests.java @@ -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 { + } + +} diff --git a/applications/sink/xmpp-sink/src/test/java/org/springframework/cloud/stream/app/sink/xmpp/XmppSinkTests.java b/applications/sink/xmpp-sink/src/test/java/org/springframework/cloud/stream/app/sink/xmpp/XmppSinkTests.java new file mode 100644 index 00000000..f70970f3 --- /dev/null +++ b/applications/sink/xmpp-sink/src/test/java/org/springframework/cloud/stream/app/sink/xmpp/XmppSinkTests.java @@ -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 { + + } + +} diff --git a/functions/common/xmpp-common/src/main/java/org/springframework/cloud/fn/common/xmpp/XmppConnectionFactoryProperties.java b/functions/common/xmpp-common/src/main/java/org/springframework/cloud/fn/common/xmpp/XmppConnectionFactoryProperties.java index aaf0e958..99cd071e 100644 --- a/functions/common/xmpp-common/src/main/java/org/springframework/cloud/fn/common/xmpp/XmppConnectionFactoryProperties.java +++ b/functions/common/xmpp-common/src/main/java/org/springframework/cloud/fn/common/xmpp/XmppConnectionFactoryProperties.java @@ -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; }