diff --git a/README.adoc b/README.adoc index 77bec007..6ef2677a 100644 --- a/README.adoc +++ b/README.adoc @@ -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] | diff --git a/applications/source/pom.xml b/applications/source/pom.xml index 2753466e..4d319a5e 100644 --- a/applications/source/pom.xml +++ b/applications/source/pom.xml @@ -10,26 +10,27 @@ pom + cdc-debezium-source file-source ftp-source jdbc-source jms-source http-source - mqtt-source - time-source load-generator-source + mail-source mongodb-source - tcp-source + mqtt-source rabbit-source s3-source sftp-source - twitter-stream-source - twitter-search-source - twitter-message-source - websocket-source - cdc-debezium-source syslog-source - mail-source + tcp-source + time-source + twitter-message-source + twitter-search-source + twitter-stream-source + websocket-source + xmpp-source zeromq-source diff --git a/applications/source/xmpp-source/README.adoc b/applications/source/xmpp-source/README.adoc new file mode 100644 index 00000000..6048f9bd --- /dev/null +++ b/applications/source/xmpp-source/README.adoc @@ -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: `$$$$`)* +$$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: `$$$$`)* + +=== xmpp.supplier + +$$payload-expression$$:: $$$$ *($$Expression$$, default: `$$$$`)* +$$stanza-filter$$:: $$$$ *($$StanzaFilter$$, default: `$$$$`)* +//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[] diff --git a/applications/source/xmpp-source/pom.xml b/applications/source/xmpp-source/pom.xml new file mode 100644 index 00000000..9f873231 --- /dev/null +++ b/applications/source/xmpp-source/pom.xml @@ -0,0 +1,85 @@ + + + 4.0.0 + + xmpp-source + xmpp-source + XMPP source app + jar + + + org.springframework.cloud.stream.app + stream-applications-core + 4.0.0-SNAPSHOT + ../../stream-applications-core/pom.xml + + + + + + + org.springframework.cloud.fn + xmpp-supplier + + + + + 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 + source + ${project.version} + org.springframework.cloud.fn.supplier.xmpp.XmppSupplierConfiguration.class + + + + + + org.springframework.cloud.fn + xmpp-supplier + + + org.springframework.cloud.stream.app + stream-applications-composite-function-support + ${stream-apps-core.version} + + + + + + + + + + + + diff --git a/applications/source/xmpp-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties b/applications/source/xmpp-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties new file mode 100644 index 00000000..7fe4f978 --- /dev/null +++ b/applications/source/xmpp-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties @@ -0,0 +1 @@ +configuration-properties.classes=org.springframework.cloud.fn.supplier.xmpp.XmppSupplierProperties,org.springframework.cloud.fn.common.xmpp.XmppConnectionFactoryProperties diff --git a/applications/source/xmpp-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties b/applications/source/xmpp-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties new file mode 100644 index 00000000..7fe4f978 --- /dev/null +++ b/applications/source/xmpp-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties @@ -0,0 +1 @@ +configuration-properties.classes=org.springframework.cloud.fn.supplier.xmpp.XmppSupplierProperties,org.springframework.cloud.fn.common.xmpp.XmppConnectionFactoryProperties diff --git a/applications/source/xmpp-source/src/test/java/org/springframework/cloud/stream/app/source/xmpp/XmppSourceInvalidConfigTests.java b/applications/source/xmpp-source/src/test/java/org/springframework/cloud/stream/app/source/xmpp/XmppSourceInvalidConfigTests.java new file mode 100644 index 00000000..4dd3ced2 --- /dev/null +++ b/applications/source/xmpp-source/src/test/java/org/springframework/cloud/stream/app/source/xmpp/XmppSourceInvalidConfigTests.java @@ -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 { + } + +} diff --git a/applications/source/xmpp-source/src/test/java/org/springframework/cloud/stream/app/source/xmpp/XmppSourceTests.java b/applications/source/xmpp-source/src/test/java/org/springframework/cloud/stream/app/source/xmpp/XmppSourceTests.java new file mode 100644 index 00000000..f9ebf3a5 --- /dev/null +++ b/applications/source/xmpp-source/src/test/java/org/springframework/cloud/stream/app/source/xmpp/XmppSourceTests.java @@ -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 { } + +}