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

Some fixes for the XmppSupplierConfiguration

Browse Source
This commit is contained in:
Artem Bilan
2024-12-19 16:50:11 -05:00
parent 24950e12e2
commit 34892ae762
1 changed files with 11 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
supplier/spring-xmpp-supplier/src/main/java/org/springframework/cloud/fn/supplier/xmpp/XmppSupplierConfiguration.java
View File

@@ -19,28 +19,27 @@ package org.springframework.cloud.fn.supplier.xmpp;
import java.util.function.Supplier;
import org.jivesoftware.smack.XMPPConnection;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.fn.common.xmpp.XmppConnectionFactoryConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.channel.FluxMessageChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.xmpp.inbound.ChatMessageListeningEndpoint;
import org.springframework.messaging.Message;
/**
* A source module that receives data from ZeroMQ.
* A auto-configuration for XMPP supplier.
*
* @author Daniel Frey
* @since 4.0.0
* @author Artem Bilan
*/
@AutoConfiguration(after = XmppConnectionFactoryConfiguration.class)
@EnableConfigurationProperties(XmppSupplierProperties.class)
public class XmppSupplierConfiguration {
private final FluxMessageChannel output = new FluxMessageChannel();
@Bean
public ChatMessageListeningEndpoint chatMessageListeningEndpoint(XMPPConnection xmppConnection,
XmppSupplierProperties properties) {
@@ -52,15 +51,17 @@ public class XmppSupplierConfiguration {
}
chatMessageListeningEndpoint.setStanzaFilter(properties.getStanzaFilter());
chatMessageListeningEndpoint.setOutputChannel(this.output);
chatMessageListeningEndpoint.setAutoStartup(false);
return chatMessageListeningEndpoint;
}
@Bean
public Supplier<Flux<Message<?>>> xmppSupplier(ChatMessageListeningEndpoint chatMessageListeningEndpoint) {
return () -> Flux.from(this.output).doOnSubscribe((subscription) -> chatMessageListeningEndpoint.start());
Publisher<Message<Object>> xmppSupplierFlow(ChatMessageListeningEndpoint chatMessageListeningEndpoint) {
return IntegrationFlow.from(chatMessageListeningEndpoint).toReactivePublisher(true);
}
@Bean
public Supplier<Flux<Message<?>>> xmppSupplier(Publisher<Message<Object>> xmppSupplierFlow) {
return () -> Flux.from(xmppSupplierFlow);
}
}