diff --git a/applications/stream-applications-core/pom.xml b/applications/stream-applications-core/pom.xml index 01213f4e..fa943851 100644 --- a/applications/stream-applications-core/pom.xml +++ b/applications/stream-applications-core/pom.xml @@ -21,8 +21,8 @@ 1.0.3-SNAPSHOT springcloud/baseimage:1.0.0 5.10 - Horsham.SR11 - 3.0.11.RELEASE + Horsham.BUILD-SNAPSHOT + 3.0.12.BUILD-SNAPSHOT 1.0.2 1.0.2 1.0.2 @@ -30,7 +30,7 @@ 1.2.1 2.0.2 - Hoxton.SR10 + Hoxton.BUILD-SNAPSHOT diff --git a/functions/supplier/tcp-supplier/README.adoc b/functions/supplier/tcp-supplier/README.adoc index 0d36a235..a957d350 100644 --- a/functions/supplier/tcp-supplier/README.adoc +++ b/functions/supplier/tcp-supplier/README.adoc @@ -2,15 +2,14 @@ This module provides a TCP supplier that can be reused and composed in other applications. The `Supplier` uses the `TcpReceivingChannelAdapter` from Spring Integration. -`tcpSupplier` is implemented as a `java.util.function.Supplier`. -This supplier gives you a reactive stream from TCP sources. The supplier has a signature of `Supplier>>`. +A `tcpSupplier` bean is implemented as a `java.util.function.Supplier`. +This supplier gives you a reactive stream from TCP sources. +The supplier has a signature of `Supplier>>`. Users have to subscribe to this `Flux` and then receive the data. ## Beans for injection -You can import the `TcpSupplierConfiguration` in the application and then inject the following bean. - -`tcpSupplier` +You can import the `TcpSupplierConfiguration` in the application and then inject the following bean: `tcpSupplier`. You need to inject this as `Supplier>>`. diff --git a/functions/supplier/tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/TcpSupplierConfiguration.java b/functions/supplier/tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/TcpSupplierConfiguration.java index 4730d862..8ad657e5 100644 --- a/functions/supplier/tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/TcpSupplierConfiguration.java +++ b/functions/supplier/tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/TcpSupplierConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2020 the original author or authors. + * Copyright 2015-2021 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. @@ -18,16 +18,17 @@ package org.springframework.cloud.fn.supplier.tcp; import java.util.function.Supplier; +import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.fn.common.tcp.EncoderDecoderFactoryBean; import org.springframework.cloud.fn.common.tcp.TcpConnectionFactoryProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import org.springframework.integration.channel.FluxMessageChannel; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.ip.IpHeaders; import org.springframework.integration.ip.config.TcpConnectionFactoryFactoryBean; import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter; import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; @@ -40,59 +41,59 @@ import org.springframework.messaging.Message; * @author Gary Russell * @author Christian Tzolov * @author Soby Chacko + * @author Artem Bilan */ -@Configuration -@EnableConfigurationProperties({TcpSupplierProperties.class, TcpConnectionFactoryProperties.class}) +@Configuration(proxyBeanMethods = false) +@EnableConfigurationProperties({ TcpSupplierProperties.class, TcpConnectionFactoryProperties.class }) public class TcpSupplierConfiguration { - @Autowired - private TcpSupplierProperties properties; - - @Autowired - private TcpConnectionFactoryProperties tcpConnectionProperties; - - @Qualifier("tcpSourceConnectionFactory") - @Autowired - private AbstractConnectionFactory connectionFactory; - @Bean - public Supplier>> tcpSupplier() { - return () -> Flux.from(output()) - .doOnSubscribe(subscription -> adapter().start()); + public EncoderDecoderFactoryBean tcpSourceDecoder(TcpSupplierProperties properties) { + EncoderDecoderFactoryBean factoryBean = new EncoderDecoderFactoryBean(properties.getDecoder()); + factoryBean.setMaxMessageSize(properties.getBufferSize()); + return factoryBean; } @Bean - public TcpReceivingChannelAdapter adapter() { + public TcpConnectionFactoryFactoryBean tcpSourceConnectionFactory( + TcpConnectionFactoryProperties tcpConnectionProperties, + @Qualifier("tcpSourceDecoder") AbstractByteArraySerializer decoder) { + + TcpConnectionFactoryFactoryBean factoryBean = new TcpConnectionFactoryFactoryBean(); + factoryBean.setType("server"); + factoryBean.setPort(tcpConnectionProperties.getPort()); + factoryBean.setUsingNio(tcpConnectionProperties.isNio()); + factoryBean.setUsingDirectBuffers(tcpConnectionProperties.isUseDirectBuffers()); + factoryBean.setLookupHost(tcpConnectionProperties.isReverseLookup()); + factoryBean.setDeserializer(decoder); + factoryBean.setSoTimeout(tcpConnectionProperties.getSocketTimeout()); + return factoryBean; + } + + @Bean + public TcpReceivingChannelAdapter adapter( + @Qualifier("tcpSourceConnectionFactory") AbstractConnectionFactory connectionFactory) { + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); adapter.setConnectionFactory(connectionFactory); - adapter.setOutputChannel(output()); adapter.setAutoStartup(false); return adapter; } @Bean - public FluxMessageChannel output() { - return new FluxMessageChannel(); + public Publisher> tcpSupplierFlow(TcpReceivingChannelAdapter adapter) { + return IntegrationFlows.from(adapter) + .headerFilter(IpHeaders.LOCAL_ADDRESS) + .toReactivePublisher(); } @Bean - public TcpConnectionFactoryFactoryBean tcpSourceConnectionFactory( - @Qualifier("tcpSourceDecoder") AbstractByteArraySerializer decoder) { - TcpConnectionFactoryFactoryBean factoryBean = new TcpConnectionFactoryFactoryBean(); - factoryBean.setType("server"); - factoryBean.setPort(this.tcpConnectionProperties.getPort()); - factoryBean.setUsingNio(this.tcpConnectionProperties.isNio()); - factoryBean.setUsingDirectBuffers(this.tcpConnectionProperties.isUseDirectBuffers()); - factoryBean.setLookupHost(this.tcpConnectionProperties.isReverseLookup()); - factoryBean.setDeserializer(decoder); - factoryBean.setSoTimeout(this.tcpConnectionProperties.getSocketTimeout()); - return factoryBean; + public Supplier>> tcpSupplier( + Publisher> tcpSupplierFlow, + TcpReceivingChannelAdapter tcpReceivingChannelAdapter) { + + return () -> Flux.from(tcpSupplierFlow) + .doOnSubscribe(subscription -> tcpReceivingChannelAdapter.start()); } - @Bean - public EncoderDecoderFactoryBean tcpSourceDecoder() { - EncoderDecoderFactoryBean factoryBean = new EncoderDecoderFactoryBean(this.properties.getDecoder()); - factoryBean.setMaxMessageSize(this.properties.getBufferSize()); - return factoryBean; - } } diff --git a/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/AbstractTcpSupplierTests.java b/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/AbstractTcpSupplierTests.java index 8f08d2ed..d3ab0995 100644 --- a/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/AbstractTcpSupplierTests.java +++ b/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/AbstractTcpSupplierTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2020 the original author or authors. + * Copyright 2015-2021 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. @@ -38,9 +38,9 @@ import static org.assertj.core.api.Assertions.assertThat; * * @author Gary Russell * @author Soby Chacko + * @author Artem Bilan */ -@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, - properties = { "tcp.host = localhost", "tcp.port = 0" }) +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = "tcp.port = 0") @DirtiesContext public class AbstractTcpSupplierTests {