diff --git a/function-dependencies/pom.xml b/function-dependencies/pom.xml index 6e685ac0..878f4b81 100644 --- a/function-dependencies/pom.xml +++ b/function-dependencies/pom.xml @@ -102,6 +102,11 @@ websocket-supplier ${project.version} + + org.springframework.cloud.fn + xmpp-supplier + ${project.version} + org.springframework.cloud.fn zeromq-supplier diff --git a/supplier/pom.xml b/supplier/pom.xml index 910a315b..7b9f3d34 100644 --- a/supplier/pom.xml +++ b/supplier/pom.xml @@ -28,6 +28,7 @@ twitter-supplier cdc-debezium-supplier syslog-supplier + xmpp-supplier zeromq-supplier diff --git a/supplier/xmpp-supplier/README.adoc b/supplier/xmpp-supplier/README.adoc new file mode 100644 index 00000000..ba4bf9f2 --- /dev/null +++ b/supplier/xmpp-supplier/README.adoc @@ -0,0 +1,30 @@ +# XMPP Supplier + +A supplier that allows you to receive messages through a XMPP server. + +## Beans for injection + +You can import the `XmppSupplierConfiguration` in the application and then inject the following bean. + +`Supplier xmppSupplier` + +You need to inject this as `Supplier xmppSupplier`. + +You can use `xmppSupplier` as a qualifier when injecting. + +**NOTE:** This is a functional endpoint. One will need to subscribe to this endpoint in order to start accepting data +on it. + +## Configuration Options + +All configuration properties are prefixed with `xmpp.supplier`. + +For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/consumer/xmpp/XmppSupplierProperties.java[XmppSupplierProperties]. + +## Tests + +See this link:src/test/java/org/springframework/cloud/fn/consumer/xmpp/[test suite] for the various ways, this supplier is used. + +## Other usage + +See this https://github.com/spring-cloud/stream-applications/blob/master/applications/source/xmpp-source/README.adoc[README] where this supplier is used to create a Spring Cloud Stream application where it makes a XMPP Source. diff --git a/supplier/xmpp-supplier/pom.xml b/supplier/xmpp-supplier/pom.xml new file mode 100644 index 00000000..cce9a457 --- /dev/null +++ b/supplier/xmpp-supplier/pom.xml @@ -0,0 +1,33 @@ + + + 4.0.0 + + + org.springframework.cloud.fn + spring-functions-parent + 4.0.0-SNAPSHOT + ../../spring-functions-parent/pom.xml + + + xmpp-supplier + xmpp-supplier + XMPP supplier + + + + org.springframework.cloud.fn + xmpp-common + ${project.version} + + + + + org.springframework.cloud.fn + function-test-support + ${project.version} + test + + + + diff --git a/supplier/xmpp-supplier/src/main/java/org/springframework/cloud/fn/supplier/xmpp/XmppSupplierConfiguration.java b/supplier/xmpp-supplier/src/main/java/org/springframework/cloud/fn/supplier/xmpp/XmppSupplierConfiguration.java new file mode 100644 index 00000000..2fa79356 --- /dev/null +++ b/supplier/xmpp-supplier/src/main/java/org/springframework/cloud/fn/supplier/xmpp/XmppSupplierConfiguration.java @@ -0,0 +1,67 @@ +/* + * 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.fn.supplier.xmpp; + +import java.util.function.Supplier; + +import org.jivesoftware.smack.XMPPConnection; +import reactor.core.publisher.Flux; + +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.fn.common.xmpp.XmppConnectionFactoryConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.integration.channel.FluxMessageChannel; +import org.springframework.integration.xmpp.inbound.ChatMessageListeningEndpoint; +import org.springframework.messaging.Message; + +/** + * A source module that receives data from ZeroMQ. + * + * @author Daniel Frey + * @since 4.0.0 + */ +@Configuration +@EnableConfigurationProperties(XmppSupplierProperties.class) +@Import(XmppConnectionFactoryConfiguration.class) +public class XmppSupplierConfiguration { + + private FluxMessageChannel output = new FluxMessageChannel(); + + @Bean + public ChatMessageListeningEndpoint chatMessageListeningEndpoint(XMPPConnection xmppConnection, XmppSupplierProperties properties) { + + var chatMessageListeningEndpoint = new ChatMessageListeningEndpoint(xmppConnection); + + if (properties.getPayloadExpression() != null) { + chatMessageListeningEndpoint.setPayloadExpression(properties.getPayloadExpression()); + } + + chatMessageListeningEndpoint.setStanzaFilter(properties.getStanzaFilter()); + chatMessageListeningEndpoint.setOutputChannel(output); + chatMessageListeningEndpoint.setAutoStartup(false); + + return chatMessageListeningEndpoint; + } + + @Bean + public Supplier>> xmppSupplier(ChatMessageListeningEndpoint chatMessageListeningEndpoint) { + return () -> Flux.from(output).doOnSubscribe(subscription -> chatMessageListeningEndpoint.start()); + } + +} diff --git a/supplier/xmpp-supplier/src/main/java/org/springframework/cloud/fn/supplier/xmpp/XmppSupplierProperties.java b/supplier/xmpp-supplier/src/main/java/org/springframework/cloud/fn/supplier/xmpp/XmppSupplierProperties.java new file mode 100644 index 00000000..7e8c4a26 --- /dev/null +++ b/supplier/xmpp-supplier/src/main/java/org/springframework/cloud/fn/supplier/xmpp/XmppSupplierProperties.java @@ -0,0 +1,54 @@ +/* + * 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.fn.supplier.xmpp; + +import org.jivesoftware.smack.filter.StanzaFilter; +import org.jivesoftware.smack.filter.StanzaTypeFilter; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.expression.Expression; +import org.springframework.validation.annotation.Validated; + +/** + * + * @author Daniel Frey + * @since 4.0.0 + */ +@ConfigurationProperties("xmpp.supplier") +@Validated +public class XmppSupplierProperties { + + private StanzaFilter stanzaFilter = StanzaTypeFilter.MESSAGE; + + private Expression payloadExpression; + + public void setStanzaFilter(StanzaFilter stanzaFilter) { + this.stanzaFilter = stanzaFilter; + } + + public StanzaFilter getStanzaFilter() { + return stanzaFilter; + } + + public void setPayloadExpression(Expression payloadExpression) { + this.payloadExpression = payloadExpression; + } + public Expression getPayloadExpression() { + return payloadExpression; + } + +} diff --git a/supplier/xmpp-supplier/src/test/java/org/springframework/cloud/fn/supplier/xmpp/XmppSupplierConfigurationTests.java b/supplier/xmpp-supplier/src/test/java/org/springframework/cloud/fn/supplier/xmpp/XmppSupplierConfigurationTests.java new file mode 100644 index 00000000..ce34757f --- /dev/null +++ b/supplier/xmpp-supplier/src/test/java/org/springframework/cloud/fn/supplier/xmpp/XmppSupplierConfigurationTests.java @@ -0,0 +1,128 @@ +/* + * 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.fn.supplier.xmpp; + +import java.io.IOException; +import java.time.Duration; +import java.util.function.Supplier; + +import org.assertj.core.api.InstanceOfAssertFactories; +import org.jivesoftware.smack.ConnectionConfiguration; +import org.jivesoftware.smack.SmackException; +import org.jivesoftware.smack.XMPPException; +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 reactor.core.publisher.Flux; +import reactor.test.StepVerifier; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.SpringBootConfiguration; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.cloud.fn.test.support.xmpp.XmppTestContainerSupport; +import org.springframework.context.annotation.Import; +import org.springframework.integration.xmpp.XmppHeaders; +import org.springframework.messaging.Message; +import org.springframework.test.context.DynamicPropertyRegistry; +import org.springframework.test.context.DynamicPropertySource; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Daniel Frey + */ +@SpringBootTest +public class XmppSupplierConfigurationTests implements XmppTestContainerSupport { + + @DynamicPropertySource + static void registerConfigurationProperties(DynamicPropertyRegistry registry) { + registry.add("xmpp.factory.host", () -> XmppTestContainerSupport.getXmppHost()); + registry.add("xmpp.factory.port", () -> XmppTestContainerSupport.getXmppMappedPort()); + registry.add("xmpp.factory.user", () -> JANE_USER); // Connect as user intended to listen for messages on behalf of + registry.add("xmpp.factory.password", () -> USER_PW); + registry.add("xmpp.factory.service-name", () -> SERVICE_NAME); + registry.add("xmpp.factory.security-mode", () -> "disabled"); + } + + @Autowired + Supplier>> subject; + + // A client source connection is needed to send the message to the xmpp server + private XMPPTCPConnection sourceConnection; + + @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(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 + void testSubscriptionConfiguration() throws XmppStringprepException, SmackException.NotConnectedException, InterruptedException { + + var payload = "test"; + + var stepVerifier = + StepVerifier.create(subject.get()) + .assertNext((message) -> { + + assertThat(message.getPayload()) + .asInstanceOf(InstanceOfAssertFactories.type(String.class)) + .isEqualTo(payload); + + assertThat(message.getHeaders().containsKey(XmppHeaders.TO)).isTrue(); + assertThat(message.getHeaders().get(XmppHeaders.TO, String.class)).isEqualTo(JANE_USER + "@" + SERVICE_NAME); + + }) + .thenCancel() + .verifyLater(); + + var chatManager = ChatManager.getInstanceFor(this.sourceConnection); + var jid = JidCreate.entityBareFrom(JANE_USER + "@" + SERVICE_NAME); + var chat = chatManager.chatWith(jid); + chat.send(payload); + + stepVerifier.verify(Duration.ofSeconds(10)); + + } + + @SpringBootConfiguration + @EnableAutoConfiguration + @Import(XmppSupplierConfiguration.class) + static class XmppSupplierTestApplication { } + +}