Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add Reactive Stream chapter into docs

Browse Source 
* Improve `channel.adoc`: polishing, new channels, java config
* Make `LoadBalancingStrategy` in the `DirectChannel` ctor as
`@Nullable` to reflect the real logic behind

* More Dos and polishing according PR comments
* Mention WebFlux and RSocket endpoints in the `endpoint-summary.adoc`
table

* More reactive streams docs
* `SourcePollingChannelAdapter` polishing
* Wrap multi-value publisher into `Mono.just()` in the `AbstractMessageProducingHandler`
instead of emitting just only a first item for the standard reply

Doc polishing
This commit is contained in:
Artem Bilan
2019-09-27 11:45:34 -04:00
committed by Gary Russell
parent eeb951b0da
commit ad96ca49f1
11 changed files with 369 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
spring-integration-core/src/main/java/org/springframework/integration/channel/DirectChannel.java
View File

@@ -20,6 +20,7 @@ import org.springframework.integration.context.IntegrationProperties;
import org.springframework.integration.dispatcher.LoadBalancingStrategy;
import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
import org.springframework.integration.dispatcher.UnicastingDispatcher;
import org.springframework.lang.Nullable;
/**
* A channel that invokes a single subscriber for each sent Message.
@@ -30,6 +31,7 @@ import org.springframework.integration.dispatcher.UnicastingDispatcher;
* @author Iwein Fuld
* @author Oleg Zhurakousky
* @author Gary Russell
* @author Artem Bilan
*/
public class DirectChannel extends AbstractSubscribableChannel {
@@ -45,12 +47,13 @@ public class DirectChannel extends AbstractSubscribableChannel {
}
/**
* Create a DirectChannel with a {@link LoadBalancingStrategy}. The
* strategy <em>must not</em> be null.
*
* Create a DirectChannel with a {@link LoadBalancingStrategy}.
* Can be {@code null} meaning that no balancing is applied;
* every message is always going to be handled by the first subscriber.
* @param loadBalancingStrategy The load balancing strategy implementation.
* @see #setFailover(boolean)
*/
public DirectChannel(LoadBalancingStrategy loadBalancingStrategy) {
public DirectChannel(@Nullable LoadBalancingStrategy loadBalancingStrategy) {
this.dispatcher.setLoadBalancingStrategy(loadBalancingStrategy);
}
@@ -58,7 +61,6 @@ public class DirectChannel extends AbstractSubscribableChannel {
/**
* Specify whether the channel's dispatcher should have failover enabled.
* By default, it will. Set this value to 'false' to disable it.
*
* @param failover The failover boolean.
*/
public void setFailover(boolean failover) {
@@ -68,7 +70,6 @@ public class DirectChannel extends AbstractSubscribableChannel {
/**
* Specify the maximum number of subscribers supported by the
* channel's dispatcher.
*
* @param maxSubscribers The maximum number of subscribers allowed.
*/
public void setMaxSubscribers(int maxSubscribers) {
@@ -85,9 +86,8 @@ public class DirectChannel extends AbstractSubscribableChannel {
protected void onInit() {
super.onInit();
if (this.maxSubscribers == null) {
Integer max = this.getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_UNICAST_SUBSCRIBERS,
Integer.class);
this.setMaxSubscribers(max);
Integer max = getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_UNICAST_SUBSCRIBERS, Integer.class);
setMaxSubscribers(max);
}
}

4
spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java
View File

@@ -63,7 +63,7 @@ public class PollingConsumer extends AbstractPollingEndpoint implements Integrat
public PollingConsumer(PollableChannel inputChannel, MessageHandler handler) {
Assert.notNull(inputChannel, "inputChannel must not be null");
Assert.notNull(handler, "handler must not be null");
if (inputChannel instanceof NullChannel && logger.isWarnEnabled()) {
if (inputChannel instanceof NullChannel) {
logger.warn("The polling from the NullChannel does not have any effects: " +
"it doesn't forward messages sent to it. A NullChannel is the end of the flow.");
}
@@ -134,7 +134,7 @@ public class PollingConsumer extends AbstractPollingEndpoint implements Integrat
try {
if (this.channelInterceptors != null
&& ((ExecutorChannelInterceptorAware) this.inputChannel).hasExecutorInterceptors()) {
interceptorStack = new ArrayDeque<ExecutorChannelInterceptor>();
interceptorStack = new ArrayDeque<>();
theMessage = applyBeforeHandle(theMessage, interceptorStack);
if (theMessage == null) {
return;

4
spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java
View File

@@ -147,11 +147,11 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint
if (AopUtils.isAopProxy(this.source)) {
Advised advised = (Advised) this.source;
this.appliedAdvices.forEach(advised::removeAdvice);
chain.stream().forEach(advice -> advised.addAdvisor(adviceToReceiveAdvisor(advice)));
chain.forEach(advice -> advised.addAdvisor(adviceToReceiveAdvisor(advice)));
}
else {
ProxyFactory proxyFactory = new ProxyFactory(this.source);
chain.stream().forEach(advice -> proxyFactory.addAdvisor(adviceToReceiveAdvisor(advice)));
chain.forEach(advice -> proxyFactory.addAdvisor(adviceToReceiveAdvisor(advice)));
this.source = (MessageSource<?>) proxyFactory.getProxy(getBeanClassLoader());
}
this.appliedAdvices.clear();

13
spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java
View File

@@ -27,6 +27,8 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.reactivestreams.Publisher;
import org.springframework.core.ReactiveAdapter;
import org.springframework.core.ReactiveAdapterRegistry;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.channel.ReactiveStreamsSubscribableChannel;
import org.springframework.integration.context.IntegrationContextUtils;
@@ -346,8 +348,15 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
}
else {
SettableListenableFuture<Object> settableListenableFuture = new SettableListenableFuture<>();
Mono.from((Publisher<?>) reply)
.subscribe(settableListenableFuture::set, settableListenableFuture::setException);
Mono<?> reactiveReply;
ReactiveAdapter adapter = ReactiveAdapterRegistry.getSharedInstance().getAdapter(null, reply);
if (adapter != null && adapter.isMultiValue()) {
reactiveReply = Mono.just(reply);
}
else {
reactiveReply = Mono.from((Publisher<?>) reply);
}
reactiveReply.subscribe(settableListenableFuture::set, settableListenableFuture::setException);
future = settableListenableFuture;
}
future.addCallback(new ReplyFutureCallback(requestMessage, replyChannel));