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

Option to preserve publish order

Browse Source 
Issue: SPR-13989
This commit is contained in:
Rossen Stoyanchev
2018-07-23 23:22:16 -04:00
parent 430250c80f
commit 7500b144ae
17 changed files with 496 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
spring-messaging/src/main/java/org/springframework/messaging/simp/broker/AbstractBrokerMessageHandler.java
View File

@@ -58,6 +58,8 @@ public abstract class AbstractBrokerMessageHandler
private final Collection<String> destinationPrefixes;
private boolean preservePublishOrder = false;
@Nullable
private ApplicationEventPublisher eventPublisher;
@@ -132,6 +134,31 @@ public abstract class AbstractBrokerMessageHandler
this.eventPublisher = publisher;
}
/**
* Whether the client must receive messages in the order of publication.
* <p>By default messages sent to the {@code "clientOutboundChannel"} may
* not be processed in the same order because the channel is backed by a
* ThreadPoolExecutor that in turn does not guarantee processing in order.
* <p>When this flag is set to {@code true} messages within the same session
* will be sent to the {@code "clientOutboundChannel"} one at a time in
* order to preserve the order of publication. Enable this only if needed
* since there is some performance overhead to keep messages in order.
* @param preservePublishOrder whether to publish in order
* @since 5.1
*/
public void setPreservePublishOrder(boolean preservePublishOrder) {
OrderedMessageSender.configureOutboundChannel(this.clientOutboundChannel, preservePublishOrder);
this.preservePublishOrder = preservePublishOrder;
}
/**
* Whether to ensure messages are received in the order of publication.
* @since 5.1
*/
public boolean isPreservePublishOrder() {
return this.preservePublishOrder;
}
@Nullable
public ApplicationEventPublisher getApplicationEventPublisher() {
return this.eventPublisher;
@@ -269,6 +296,16 @@ public abstract class AbstractBrokerMessageHandler
}
}
/**
* Get the MessageChannel to use for sending messages to clients, possibly
* a per-session wrapper when {@code preservePublishOrder=true}.
* @since 5.1
*/
protected MessageChannel getClientOutboundChannelForSession(String sessionId) {
return this.preservePublishOrder ?
new OrderedMessageSender(getClientOutboundChannel(), logger) : getClientOutboundChannel();
}
/**
* Detect unsent DISCONNECT messages and process them anyway.

152
spring-messaging/src/main/java/org/springframework/messaging/simp/broker/OrderedMessageSender.java Normal file
View File

@@ -0,0 +1,152 @@
/*
* Copyright 2002-2018 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
*
* http://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.messaging.simp.broker;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.logging.Log;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
import org.springframework.messaging.support.ExecutorChannelInterceptor;
import org.springframework.messaging.support.ExecutorSubscribableChannel;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.util.Assert;
/**
* Submit messages to an ExecutorSubscribableChannel, one at a time. The channel
* must have been configured with {@link #configureOutboundChannel}.
*
* @author Rossen Stoyanchev
* @since 5.1
*/
class OrderedMessageSender implements MessageChannel {
static final String COMPLETION_TASK_HEADER = "simpSendCompletionTask";
private final MessageChannel channel;
private final Log logger;
private final Queue<Message<?>> messages = new ConcurrentLinkedQueue<>();
private final AtomicBoolean sendInProgress = new AtomicBoolean(false);
public OrderedMessageSender(MessageChannel channel, Log logger) {
this.channel = channel;
this.logger = logger;
}
public boolean send(Message<?> message) {
return send(message, -1);
}
@Override
public boolean send(Message<?> message, long timeout) {
this.messages.add(message);
trySend();
return true;
}
private void trySend() {
// Take sendInProgress flag only if queue is not empty
if (this.messages.isEmpty()) {
return;
}
if (this.sendInProgress.compareAndSet(false, true)) {
sendNextMessage();
}
}
private void sendNextMessage() {
for (;;) {
Message<?> message = this.messages.poll();
if (message != null) {
try {
addCompletionCallback(message);
if (this.channel.send(message)) {
return;
}
}
catch (Throwable ex) {
if (logger.isErrorEnabled()) {
logger.error("Failed to send " + message, ex);
}
}
}
else {
// We ran out of messages..
this.sendInProgress.set(false);
trySend();
break;
}
}
}
private void addCompletionCallback(Message<?> msg) {
SimpMessageHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(msg, SimpMessageHeaderAccessor.class);
Assert.isTrue(accessor != null && accessor.isMutable(), "Expected mutable SimpMessageHeaderAccessor");
accessor.setHeader(COMPLETION_TASK_HEADER, (Runnable) this::sendNextMessage);
}
/**
* Install or remove an {@link ExecutorChannelInterceptor} that invokes a
* completion task once the message is handled.
* @param channel the channel to configure
* @param preservePublishOrder whether preserve order is on or off based on
* which an interceptor is either added or removed.
*/
static void configureOutboundChannel(MessageChannel channel, boolean preservePublishOrder) {
if (preservePublishOrder) {
Assert.isInstanceOf(ExecutorSubscribableChannel.class, channel,
"An ExecutorSubscribableChannel is required for `preservePublishOrder`");
ExecutorSubscribableChannel execChannel = (ExecutorSubscribableChannel) channel;
if (execChannel.getInterceptors().stream().noneMatch(i -> i instanceof CallbackInterceptor)) {
execChannel.addInterceptor(0, new CallbackInterceptor());
}
}
else if (channel instanceof ExecutorSubscribableChannel) {
ExecutorSubscribableChannel execChannel = (ExecutorSubscribableChannel) channel;
execChannel.getInterceptors().stream().filter(i -> i instanceof CallbackInterceptor)
.findFirst()
.map(execChannel::removeInterceptor);
}
}
private static class CallbackInterceptor implements ExecutorChannelInterceptor {
@Override
public void afterMessageHandled(Message<?> msg, MessageChannel ch, MessageHandler handler, Exception ex) {
Runnable task = (Runnable) msg.getHeaders().get(OrderedMessageSender.COMPLETION_TASK_HEADER);
if (task != null) {
task.run();
}
}
}
}

43
spring-messaging/src/main/java/org/springframework/messaging/simp/broker/SimpleBrokerMessageHandler.java
View File

@@ -306,10 +306,11 @@ public class SimpleBrokerMessageHandler extends AbstractBrokerMessageHandler {
else if (SimpMessageType.CONNECT.equals(messageType)) {
logMessage(message);
if (sessionId != null) {
long[] clientHeartbeat = SimpMessageHeaderAccessor.getHeartbeat(headers);
long[] serverHeartbeat = getHeartbeatValue();
long[] heartbeatIn = SimpMessageHeaderAccessor.getHeartbeat(headers);
long[] heartbeatOut = getHeartbeatValue();
Principal user = SimpMessageHeaderAccessor.getUser(headers);
this.sessions.put(sessionId, new SessionInfo(sessionId, user, clientHeartbeat, serverHeartbeat));
MessageChannel outChannel = getClientOutboundChannelForSession(sessionId);
this.sessions.put(sessionId, new SessionInfo(sessionId, user, outChannel, heartbeatIn, heartbeatOut));
SimpMessageHeaderAccessor connectAck = SimpMessageHeaderAccessor.create(SimpMessageType.CONNECT_ACK);
initHeaders(connectAck);
connectAck.setSessionId(sessionId);
@@ -317,7 +318,7 @@ public class SimpleBrokerMessageHandler extends AbstractBrokerMessageHandler {
connectAck.setUser(user);
}
connectAck.setHeader(SimpMessageHeaderAccessor.CONNECT_MESSAGE_HEADER, message);
connectAck.setHeader(SimpMessageHeaderAccessor.HEART_BEAT_HEADER, serverHeartbeat);
connectAck.setHeader(SimpMessageHeaderAccessor.HEART_BEAT_HEADER, heartbeatOut);
Message<byte[]> messageOut = MessageBuilder.createMessage(EMPTY_PAYLOAD, connectAck.getMessageHeaders());
getClientOutboundChannel().send(messageOut);
}
@@ -391,19 +392,20 @@ public class SimpleBrokerMessageHandler extends AbstractBrokerMessageHandler {
headerAccessor.setSessionId(sessionId);
headerAccessor.setSubscriptionId(subscriptionId);
headerAccessor.copyHeadersIfAbsent(message.getHeaders());
headerAccessor.setLeaveMutable(true);
Object payload = message.getPayload();
Message<?> reply = MessageBuilder.createMessage(payload, headerAccessor.getMessageHeaders());
try {
getClientOutboundChannel().send(reply);
}
catch (Throwable ex) {
if (logger.isErrorEnabled()) {
logger.error("Failed to send " + message, ex);
SessionInfo info = this.sessions.get(sessionId);
if (info != null) {
try {
info.getClientOutboundChannel().send(reply);
}
}
finally {
SessionInfo info = this.sessions.get(sessionId);
if (info != null) {
catch (Throwable ex) {
if (logger.isErrorEnabled()) {
logger.error("Failed to send " + message, ex);
}
}
finally {
info.setLastWriteTime(now);
}
}
@@ -427,6 +429,8 @@ public class SimpleBrokerMessageHandler extends AbstractBrokerMessageHandler {
@Nullable
private final Principal user;
private final MessageChannel clientOutboundChannel;
private final long readInterval;
private final long writeInterval;
@@ -435,11 +439,13 @@ public class SimpleBrokerMessageHandler extends AbstractBrokerMessageHandler {
private volatile long lastWriteTime;
public SessionInfo(String sessionId, @Nullable Principal user,
public SessionInfo(String sessionId, @Nullable Principal user, MessageChannel outboundChannel,
@Nullable long[] clientHeartbeat, @Nullable long[] serverHeartbeat) {
this.sessionId = sessionId;
this.user = user;
this.clientOutboundChannel = outboundChannel;
if (clientHeartbeat != null && serverHeartbeat != null) {
this.readInterval = (clientHeartbeat[0] > 0 && serverHeartbeat[1] > 0 ?
Math.max(clientHeartbeat[0], serverHeartbeat[1]) * HEARTBEAT_MULTIPLIER : 0);
@@ -462,6 +468,10 @@ public class SimpleBrokerMessageHandler extends AbstractBrokerMessageHandler {
return this.user;
}
public MessageChannel getClientOutboundChannel() {
return this.clientOutboundChannel;
}
public long getReadInterval() {
return this.readInterval;
}
@@ -505,8 +515,9 @@ public class SimpleBrokerMessageHandler extends AbstractBrokerMessageHandler {
accessor.setUser(user);
}
initHeaders(accessor);
accessor.setLeaveMutable(true);
MessageHeaders headers = accessor.getMessageHeaders();
getClientOutboundChannel().send(MessageBuilder.createMessage(EMPTY_PAYLOAD, headers));
info.getClientOutboundChannel().send(MessageBuilder.createMessage(EMPTY_PAYLOAD, headers));
}
}
}

31
spring-messaging/src/main/java/org/springframework/messaging/simp/config/MessageBrokerRegistry.java
View File

@@ -54,6 +54,8 @@ public class MessageBrokerRegistry {
@Nullable
private String userDestinationPrefix;
private boolean preservePublishOrder;
@Nullable
private PathMatcher pathMatcher;
@@ -160,6 +162,30 @@ public class MessageBrokerRegistry {
return this.userDestinationPrefix;
}
/**
* Whether the client must receive messages in the order of publication.
* <p>By default messages sent to the {@code "clientOutboundChannel"} may
* not be processed in the same order because the channel is backed by a
* ThreadPoolExecutor that in turn does not guarantee processing in order.
* <p>When this flag is set to {@code true} messages within the same session
* will be sent to the {@code "clientOutboundChannel"} one at a time in
* order to preserve the order of publication. Enable this only if needed
* since there is some performance overhead to keep messages in order.
* @param preservePublishOrder whether to publish in order
* @since 5.1
*/
public void setPreservePublishOrder(boolean preservePublishOrder) {
this.preservePublishOrder = preservePublishOrder;
}
/**
* Whether to ensure messages are received in the order of publication.
* @since 5.1
*/
protected boolean isPreservePublishOrder() {
return this.preservePublishOrder;
}
/**
* Configure the PathMatcher to use to match the destinations of incoming
* messages to {@code @MessageMapping} and {@code @SubscribeMapping} methods.
@@ -209,6 +235,7 @@ public class MessageBrokerRegistry {
SimpleBrokerMessageHandler handler = this.simpleBrokerRegistration.getMessageHandler(brokerChannel);
handler.setPathMatcher(this.pathMatcher);
handler.setCacheLimit(this.cacheLimit);
handler.setPreservePublishOrder(this.preservePublishOrder);
return handler;
}
return null;
@@ -217,7 +244,9 @@ public class MessageBrokerRegistry {
@Nullable
protected StompBrokerRelayMessageHandler getStompBrokerRelay(SubscribableChannel brokerChannel) {
if (this.brokerRelayRegistration != null) {
return this.brokerRelayRegistration.getMessageHandler(brokerChannel);
StompBrokerRelayMessageHandler relay = this.brokerRelayRegistration.getMessageHandler(brokerChannel);
relay.setPreservePublishOrder(this.preservePublishOrder);
return relay;
}
return null;
}

10
spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java
View File

@@ -578,6 +578,8 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
private final StompHeaderAccessor connectHeaders;
private final MessageChannel outboundChannel;
@Nullable
private volatile TcpConnection<byte[]> tcpConnection;
@@ -594,6 +596,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
this.sessionId = sessionId;
this.connectHeaders = connectHeaders;
this.isRemoteClientSession = isClientSession;
this.outboundChannel = getClientOutboundChannelForSession(sessionId);
}
public String getSessionId() {
@@ -660,6 +663,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
accessor.setUser(user);
}
accessor.setMessage(errorText);
accessor.setLeaveMutable(true);
Message<?> errorMessage = MessageBuilder.createMessage(EMPTY_PAYLOAD, accessor.getMessageHeaders());
handleInboundMessage(errorMessage);
}
@@ -667,11 +671,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
protected void handleInboundMessage(Message<?> message) {
if (this.isRemoteClientSession) {
MessageHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, null);
if (accessor != null) {
accessor.setImmutable();
}
StompBrokerRelayMessageHandler.this.getClientOutboundChannel().send(message);
this.outboundChannel.send(message);
}
}

8
spring-messaging/src/main/java/org/springframework/messaging/support/ExecutorChannelInterceptor.java
View File

@@ -44,7 +44,9 @@ public interface ExecutorChannelInterceptor extends ChannelInterceptor {
* @return the input message, or a new instance, or {@code null}
*/
@Nullable
Message<?> beforeHandle(Message<?> message, MessageChannel channel, MessageHandler handler);
default Message<?> beforeHandle(Message<?> message, MessageChannel channel, MessageHandler handler) {
return message;
}
/**
* Invoked inside the {@link Runnable} submitted to the Executor after calling
@@ -57,6 +59,8 @@ public interface ExecutorChannelInterceptor extends ChannelInterceptor {
* @param handler the target handler that handled the message
* @param ex any exception that may been raised by the handler
*/
void afterMessageHandled(Message<?> message, MessageChannel channel, MessageHandler handler, @Nullable Exception ex);
default void afterMessageHandled(Message<?> message, MessageChannel channel, MessageHandler handler,
@Nullable Exception ex) {
}
}

18
spring-messaging/src/main/java/org/springframework/messaging/support/ExecutorSubscribableChannel.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 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.
@@ -70,16 +70,22 @@ public class ExecutorSubscribableChannel extends AbstractSubscribableChannel {
public void setInterceptors(List<ChannelInterceptor> interceptors) {
super.setInterceptors(interceptors);
this.executorInterceptors.clear();
for (ChannelInterceptor interceptor : interceptors) {
if (interceptor instanceof ExecutorChannelInterceptor) {
this.executorInterceptors.add((ExecutorChannelInterceptor) interceptor);
}
}
interceptors.forEach(this::updateExecutorInterceptorsFor);
}
@Override
public void addInterceptor(ChannelInterceptor interceptor) {
super.addInterceptor(interceptor);
updateExecutorInterceptorsFor(interceptor);
}
@Override
public void addInterceptor(int index, ChannelInterceptor interceptor) {
super.addInterceptor(index, interceptor);
updateExecutorInterceptorsFor(interceptor);
}
private void updateExecutorInterceptorsFor(ChannelInterceptor interceptor) {
if (interceptor instanceof ExecutorChannelInterceptor) {
this.executorInterceptors.add((ExecutorChannelInterceptor) interceptor);
}