SpringCloud/spring-functions-catalog
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Xmpp supplier (#403)

Browse Source 
* WIP: initial xmpp supplier support

* implement xmpp supplier and test

* remove test logging config

* update for consistency

* add documentation references for xmpp consumer and supplier, fix formatting for functions where geode was commented out
This commit is contained in:
Daniel Frey
2022-11-07 10:02:01 -05:00
committed by GitHub
parent caa0ad20e5
commit 29c51684bb
7 changed files with 318 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
function-dependencies/pom.xml
View File

@@ -102,6 +102,11 @@
<artifactId>websocket-supplier</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>xmpp-supplier</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>zeromq-supplier</artifactId>

1
supplier/pom.xml
View File

@@ -28,6 +28,7 @@
<module>twitter-supplier</module>
<module>cdc-debezium-supplier</module>
<module>syslog-supplier</module>
<module>xmpp-supplier</module>
<module>zeromq-supplier</module>
</modules>
<build>

30
supplier/xmpp-supplier/README.adoc Normal file
View File

@@ -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<Message<?> xmppSupplier`
You need to inject this as `Supplier<Message<?> 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.

33
supplier/xmpp-supplier/pom.xml Normal file
View File

@@ -0,0 +1,33 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
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>
<parent>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>spring-functions-parent</artifactId>
<version>4.0.0-SNAPSHOT</version>
<relativePath>../../spring-functions-parent/pom.xml</relativePath>
</parent>
<artifactId>xmpp-supplier</artifactId>
<name>xmpp-supplier</name>
<description>XMPP supplier</description>
<dependencies>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>xmpp-common</artifactId>
<version>${project.version}</version>
</dependency>
<!-- Test dependencies -->
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>function-test-support</artifactId>
<version>${project.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
</project>

67
supplier/xmpp-supplier/src/main/java/org/springframework/cloud/fn/supplier/xmpp/XmppSupplierConfiguration.java Normal file
View File

@@ -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<Flux<Message<?>>> xmppSupplier(ChatMessageListeningEndpoint chatMessageListeningEndpoint) {
return () -> Flux.from(output).doOnSubscribe(subscription -> chatMessageListeningEndpoint.start());
}
}

54
supplier/xmpp-supplier/src/main/java/org/springframework/cloud/fn/supplier/xmpp/XmppSupplierProperties.java Normal file
View File

@@ -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;
}
}

128
supplier/xmpp-supplier/src/test/java/org/springframework/cloud/fn/supplier/xmpp/XmppSupplierConfigurationTests.java Normal file
View File

@@ -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<Flux<Message<?>>> 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 { }
}