diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/ConnectionHandlingStompSession.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/ConnectionHandlingStompSession.java new file mode 100644 index 0000000000..e2abc57194 --- /dev/null +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/ConnectionHandlingStompSession.java @@ -0,0 +1,40 @@ +/* + * Copyright 2002-2015 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.stomp; + +import org.springframework.messaging.tcp.TcpConnectionHandler; +import org.springframework.util.concurrent.ListenableFuture; + + +/** + * A {@link StompSession} that implements + * {@link org.springframework.messaging.tcp.TcpConnectionHandler + * TcpConnectionHandler} in order to send and receive messages. + * + *

A ConnectionHandlingStompSession can be used with any TCP or WebSocket + * library that is adapted to the {@code TcpConnectionHandler} contract. + * + * @author Rossen Stoyanchev + * @since 4.2 + */ +public interface ConnectionHandlingStompSession extends StompSession, TcpConnectionHandler { + + /** + * Return a future that will complete when the session is ready for use. + */ + ListenableFuture getSessionFuture(); + +} diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/ConnectionLostException.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/ConnectionLostException.java new file mode 100644 index 0000000000..eeb5c96c77 --- /dev/null +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/ConnectionLostException.java @@ -0,0 +1,33 @@ +/* + * Copyright 2002-2015 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.stomp; + + +/** + * Raised when the connection for a STOMP session is lost rather than closed. + * + * @author Rossen Stoyanchev + * @since 4.2 + */ +@SuppressWarnings("serial") +public class ConnectionLostException extends Exception { + + + public ConnectionLostException(String message) { + super(message); + } + +} diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/DefaultStompSession.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/DefaultStompSession.java new file mode 100644 index 0000000000..2d22bc9717 --- /dev/null +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/DefaultStompSession.java @@ -0,0 +1,656 @@ +/* + * Copyright 2002-2015 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.stomp; + +import java.lang.reflect.Type; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Date; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.core.ResolvableType; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageDeliveryException; +import org.springframework.messaging.converter.MessageConversionException; +import org.springframework.messaging.converter.MessageConverter; +import org.springframework.messaging.converter.StringMessageConverter; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.messaging.support.MessageHeaderAccessor; +import org.springframework.messaging.tcp.TcpConnection; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.util.AlternativeJdkIdGenerator; +import org.springframework.util.Assert; +import org.springframework.util.IdGenerator; +import org.springframework.util.StringUtils; +import org.springframework.util.concurrent.ListenableFuture; +import org.springframework.util.concurrent.ListenableFutureCallback; +import org.springframework.util.concurrent.SettableListenableFuture; + + +/** + * Default implementation of {@link ConnectionHandlingStompSession}. + * + * @author Rossen Stoyanchev + * @since 4.2 + */ +public class DefaultStompSession implements ConnectionHandlingStompSession { + + private static Log logger = LogFactory.getLog(DefaultStompSession.class); + + private static final IdGenerator idGenerator = new AlternativeJdkIdGenerator(); + + public static final byte[] EMPTY_PAYLOAD = new byte[0]; + + /* STOMP spec: receiver SHOULD take into account an error margin */ + private static final long HEARTBEAT_MULTIPLIER = 3; + + private static final Message HEARTBEAT; + + + static { + StompHeaderAccessor accessor = StompHeaderAccessor.createForHeartbeat(); + HEARTBEAT = MessageBuilder.createMessage(StompDecoder.HEARTBEAT_PAYLOAD, accessor.getMessageHeaders()); + } + + + private final String sessionId; + + private final StompSessionHandler sessionHandler; + + private final StompHeaders connectHeaders; + + private final SettableListenableFuture sessionFuture = new SettableListenableFuture(); + + private MessageConverter converter = new StringMessageConverter(); + + private TaskScheduler taskScheduler; + + private long receiptTimeLimit = 15 * 1000; + + private volatile boolean autoReceiptEnabled; + + + private volatile TcpConnection connection; + + private final AtomicInteger subscriptionIndex = new AtomicInteger(); + + private final Map subscriptions = new ConcurrentHashMap(4); + + private final AtomicInteger receiptIndex = new AtomicInteger(); + + private final Map receiptHandlers = new ConcurrentHashMap(4); + + /* Whether the client is willfully closing the connection */ + private volatile boolean closing = false; + + + /** + * Create a new session. + * + * @param sessionHandler the application handler for the session + * @param connectHeaders headers for the STOMP CONNECT frame + */ + public DefaultStompSession(StompSessionHandler sessionHandler, StompHeaders connectHeaders) { + Assert.notNull(sessionHandler, "'sessionHandler' is required."); + Assert.notNull(connectHeaders, "'connectHeaders' is required."); + this.sessionId = idGenerator.generateId().toString(); + this.sessionHandler = sessionHandler; + this.connectHeaders = connectHeaders; + } + + + @Override + public String getSessionId() { + return this.sessionId; + } + + /** + * Return the configured session handler. + */ + public StompSessionHandler getSessionHandler() { + return this.sessionHandler; + } + + @Override + public ListenableFuture getSessionFuture() { + return this.sessionFuture; + } + + /** + * Set the {@link MessageConverter} to use to convert the payload of incoming + * and outgoing messages to and from {@code byte[]} based on object type, or + * expected object type, and the "content-type" header. + *

By default, {@link StringMessageConverter} is configured. + * @param messageConverter the message converter to use + */ + public void setMessageConverter(MessageConverter messageConverter) { + Assert.notNull(messageConverter, "'messageConverter' must not be null"); + this.converter = messageConverter; + } + + /** + * Return the configured {@link MessageConverter}. + */ + public MessageConverter getMessageConverter() { + return this.converter; + } + + /** + * Configure the TaskScheduler to use for receipt tracking. + */ + public void setTaskScheduler(TaskScheduler taskScheduler) { + this.taskScheduler = taskScheduler; + } + + /** + * Return the configured TaskScheduler to use for receipt tracking. + */ + public TaskScheduler getTaskScheduler() { + return this.taskScheduler; + } + + /** + * Configure the time in milliseconds before a receipt expires. + *

By default set to 15,000 (15 seconds). + */ + public void setReceiptTimeLimit(long receiptTimeLimit) { + Assert.isTrue(receiptTimeLimit > 0); + this.receiptTimeLimit = receiptTimeLimit; + } + + /** + * Return the configured time limit before a receipt expires. + */ + public long getReceiptTimeLimit() { + return this.receiptTimeLimit; + } + + @Override + public void setAutoReceipt(boolean autoReceiptEnabled) { + this.autoReceiptEnabled = autoReceiptEnabled; + } + + /** + * Whether receipt headers should be automatically added. + */ + public boolean isAutoReceiptEnabled() { + return this.autoReceiptEnabled; + } + + + @Override + public boolean isConnected() { + return this.connection != null; + } + + @Override + public Receiptable send(String destination, Object payload) { + StompHeaders stompHeaders = new StompHeaders(); + stompHeaders.setDestination(destination); + return send(stompHeaders, payload); + } + + @Override + public Receiptable send(StompHeaders stompHeaders, Object payload) { + Assert.hasText(stompHeaders.getDestination(), "'destination' header is required"); + + String receiptId = checkOrAddReceipt(stompHeaders); + Receiptable receiptable = new ReceiptHandler(receiptId); + + StompHeaderAccessor accessor = createHeaderAccessor(StompCommand.SEND); + accessor.addNativeHeaders(stompHeaders); + Message message = createMessage(accessor, payload); + execute(message); + + return receiptable; + } + + private String checkOrAddReceipt(StompHeaders stompHeaders) { + String receiptId = stompHeaders.getReceipt(); + if (isAutoReceiptEnabled() && receiptId == null) { + receiptId = String.valueOf(DefaultStompSession.this.receiptIndex.getAndIncrement()); + stompHeaders.setReceipt(receiptId); + } + return receiptId; + } + + private StompHeaderAccessor createHeaderAccessor(StompCommand command) { + StompHeaderAccessor accessor = StompHeaderAccessor.create(command); + accessor.setSessionId(this.sessionId); + accessor.setLeaveMutable(true); + return accessor; + } + + @SuppressWarnings("unchecked") + private Message createMessage(StompHeaderAccessor accessor, Object payload) { + accessor.updateSimpMessageHeadersFromStompHeaders(); + Message message; + if (payload == null) { + message = MessageBuilder.createMessage(EMPTY_PAYLOAD, accessor.getMessageHeaders()); + } + else if (payload instanceof byte[]) { + message = MessageBuilder.createMessage((byte[]) payload, accessor.getMessageHeaders()); + } + else { + message = (Message) getMessageConverter().toMessage(payload, accessor.getMessageHeaders()); + accessor.updateStompHeadersFromSimpMessageHeaders(); + if (message == null) { + throw new MessageConversionException("Unable to convert payload with type='" + + payload.getClass().getName() + "', contentType='" + accessor.getContentType() + + "', converter=[" + getMessageConverter() + "]"); + } + } + return message; + } + + private void execute(Message message) { + StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class); + if (logger.isTraceEnabled()) { + logger.trace("Sending " + accessor.getDetailedLogMessage(message.getPayload())); + } + TcpConnection conn = this.connection; + Assert.state(conn != null, "Connection closed"); + try { + conn.send(message).get(); + } + catch (ExecutionException ex) { + throw new MessageDeliveryException(message, ex.getCause()); + } + catch (Throwable ex) { + throw new MessageDeliveryException(message, ex); + } + } + + @Override + public Subscription subscribe(String destination, StompFrameHandler handler) { + StompHeaders stompHeaders = new StompHeaders(); + stompHeaders.setDestination(destination); + return subscribe(stompHeaders, handler); + } + + @Override + public Subscription subscribe(StompHeaders stompHeaders, StompFrameHandler handler) { + String destination = stompHeaders.getDestination(); + Assert.hasText(destination, "'destination' is required"); + Assert.notNull(handler, "'handler' is required"); + + String subscriptionId = stompHeaders.getId(); + if (!StringUtils.hasText(subscriptionId)) { + subscriptionId = String.valueOf(DefaultStompSession.this.subscriptionIndex.getAndIncrement()); + stompHeaders.setId(subscriptionId); + } + String receiptId = checkOrAddReceipt(stompHeaders); + Subscription subscription = new DefaultSubscription(subscriptionId, destination, receiptId, handler); + + StompHeaderAccessor accessor = createHeaderAccessor(StompCommand.SUBSCRIBE); + accessor.addNativeHeaders(stompHeaders); + Message message = createMessage(accessor, EMPTY_PAYLOAD); + execute(message); + + return subscription; + } + + private void unsubscribe(String id) { + StompHeaderAccessor accessor = createHeaderAccessor(StompCommand.UNSUBSCRIBE); + accessor.setSubscriptionId(id); + Message message = createMessage(accessor, EMPTY_PAYLOAD); + execute(message); + } + + @Override + public void disconnect() { + this.closing = true; + try { + StompHeaderAccessor accessor = createHeaderAccessor(StompCommand.DISCONNECT); + Message message = createMessage(accessor, EMPTY_PAYLOAD); + execute(message); + } + finally { + resetConnection(); + } + } + + // TcpConnectionHandler + + @Override + public void afterConnected(TcpConnection connection) { + this.connection = connection; + if (logger.isDebugEnabled()) { + logger.debug("Connection established in session id=" + this.sessionId); + } + StompHeaderAccessor accessor = createHeaderAccessor(StompCommand.CONNECT); + accessor.addNativeHeaders(this.connectHeaders); + accessor.setAcceptVersion("1.1,1.2"); + Message message = createMessage(accessor, EMPTY_PAYLOAD); + execute(message); + } + + @Override + public void afterConnectFailure(Throwable ex) { + if (logger.isDebugEnabled()) { + logger.debug("Failed to connect session id=" + this.sessionId, ex); + } + this.sessionFuture.setException(ex); + this.sessionHandler.handleTransportError(this, ex); + } + + @Override + public void handleMessage(Message message) { + StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class); + accessor.setSessionId(this.sessionId); + StompCommand command = accessor.getCommand(); + Map> nativeHeaders = accessor.getNativeHeaders(); + StompHeaders stompHeaders = StompHeaders.readOnlyStompHeaders(nativeHeaders); + boolean isHeartbeat = accessor.isHeartbeat(); + if (logger.isTraceEnabled()) { + logger.trace("Received " + accessor.getDetailedLogMessage(message.getPayload())); + } + try { + if (StompCommand.MESSAGE.equals(command)) { + DefaultSubscription subscription = this.subscriptions.get(stompHeaders.getSubscription()); + if (subscription != null) { + invokeHandler(subscription.getHandler(), message, stompHeaders); + } + else if (logger.isDebugEnabled()) { + logger.debug("No handler for: " + accessor.getDetailedLogMessage(message.getPayload()) + + ". Perhaps just unscubscribed?"); + } + } + else { + if (StompCommand.RECEIPT.equals(command)) { + String receiptId = stompHeaders.getReceiptId(); + ReceiptHandler handler = this.receiptHandlers.get(receiptId); + if (handler != null) { + handler.handleReceiptReceived(); + } + else if (logger.isDebugEnabled()) { + logger.debug("No matching receipt: " + accessor.getDetailedLogMessage(message.getPayload())); + } + } + else if (StompCommand.CONNECTED.equals(command)) { + initHeartbeats(stompHeaders); + this.sessionFuture.set(this); + this.sessionHandler.afterConnected(this, stompHeaders); + } + else if (StompCommand.ERROR.equals(command)) { + invokeHandler(this.sessionHandler, message, stompHeaders); + } + else if (!isHeartbeat && logger.isTraceEnabled()) { + logger.trace("Message not handled."); + } + } + } + catch (Throwable ex) { + this.sessionHandler.handleException(this, command, stompHeaders, message.getPayload(), ex); + } + } + + private void invokeHandler(StompFrameHandler handler, Message message, StompHeaders stompHeaders) { + if (message.getPayload().length == 0) { + handler.handleFrame(stompHeaders, null); + return; + } + Type type = handler.getPayloadType(stompHeaders); + Class payloadType = ResolvableType.forType(type).getRawClass(); + Object object = getMessageConverter().fromMessage(message, payloadType); + if (object == null) { + throw new MessageConversionException("No suitable converter, payloadType=" + payloadType); + } + handler.handleFrame(stompHeaders, object); + } + + private void initHeartbeats(StompHeaders connectedHeaders) { + long clientRead = this.connectHeaders.getHeartbeat()[0]; + long serverWrite = connectedHeaders.getHeartbeat()[1]; + + if (clientRead > 0 && serverWrite > 0) { + long interval = Math.max(clientRead, serverWrite); + this.connection.onWriteInactivity(new WriteInactivityTask(), interval); + } + + long clientWrite = this.connectHeaders.getHeartbeat()[1]; + long serverRead = connectedHeaders.getHeartbeat()[0]; + + if (clientWrite > 0 && serverRead > 0) { + final long interval = Math.max(clientWrite, serverRead) * HEARTBEAT_MULTIPLIER; + this.connection.onReadInactivity(new ReadInactivityTask(), interval); + } + } + + @Override + public void handleFailure(Throwable ex) { + try { + this.sessionFuture.setException(ex); // no-op if already set + this.sessionHandler.handleTransportError(this, ex); + } + catch (Throwable ex2) { + if (logger.isDebugEnabled()) { + logger.debug("Uncaught failure while handling transport failure", ex2); + } + } + } + + @Override + public void afterConnectionClosed() { + if (logger.isDebugEnabled()) { + logger.debug("Connection closed session id=" + this.sessionId); + } + if (!this.closing) { + handleFailure(new ConnectionLostException("Connection closed")); + resetConnection(); + } + } + + private void resetConnection() { + TcpConnection conn = this.connection; + this.connection = null; + if (conn != null) { + try { + conn.close(); + } + catch (Throwable ex) { + // Ignore + } + } + } + + + + private class ReceiptHandler implements Receiptable { + + private final String receiptId; + + private final List receiptCallbacks = new ArrayList(2); + + private final List receiptLostCallbacks = new ArrayList(2); + + private ScheduledFuture future; + + private Boolean result; + + + public ReceiptHandler(String receiptId) { + this.receiptId = receiptId; + if (this.receiptId != null) { + initReceiptHandling(); + } + } + + private void initReceiptHandling() { + Assert.notNull(getTaskScheduler(), "To track receipts a TaskScheduler must be configured"); + DefaultStompSession.this.receiptHandlers.put(this.receiptId, this); + Date startTime = new Date(System.currentTimeMillis() + getReceiptTimeLimit()); + this.future = getTaskScheduler().schedule(new Runnable() { + @Override + public void run() { + handleReceiptNotReceived(); + } + }, startTime); + } + + @Override + public String getReceiptId() { + return this.receiptId; + } + + @Override + public void addReceiptTask(Runnable task) { + addTask(task, true); + } + + @Override + public void addReceiptLostTask(Runnable task) { + addTask(task, false); + } + + private void addTask(Runnable task, boolean successTask) { + Assert.notNull(this.receiptId, "To track receipts, set autoReceiptEnabled=true or add 'receiptId' header"); + synchronized (this) { + if (this.result != null && this.result == successTask) { + invoke(Arrays.asList(task)); + } + else { + if (successTask) { + this.receiptCallbacks.add(task); + } + else { + this.receiptLostCallbacks.add(task); + } + } + } + } + + private void invoke(List callbacks) { + for (Runnable runnable : callbacks) { + try { + runnable.run(); + } + catch (Throwable ex) { + // Ignore + } + } + } + + public void handleReceiptReceived() { + handleInternal(true); + } + + public void handleReceiptNotReceived() { + handleInternal(false); + } + + private void handleInternal(boolean result) { + synchronized (this) { + if (this.result != null) { + return; + } + this.result = result; + invoke(result ? this.receiptCallbacks : this.receiptLostCallbacks); + DefaultStompSession.this.receiptHandlers.remove(this.receiptId); + if (this.future != null) { + this.future.cancel(true); + } + } + } + } + + private class DefaultSubscription extends ReceiptHandler implements Subscription { + + private final String id; + + private final String destination; + + private final StompFrameHandler handler; + + + public DefaultSubscription(String id, String destination, String receiptId, StompFrameHandler handler) { + super(receiptId); + Assert.notNull(destination, "'destination' is required"); + Assert.notNull(handler, "'handler' handler is required"); + this.id = id; + this.destination = destination; + this.handler = handler; + + DefaultStompSession.this.subscriptions.put(id, this); + } + + @Override + public String getSubscriptionId() { + return this.id; + } + + public String getDestination() { + return this.destination; + } + + public StompFrameHandler getHandler() { + return this.handler; + } + + @Override + public void unsubscribe() { + DefaultStompSession.this.subscriptions.remove(getSubscriptionId()); + DefaultStompSession.this.unsubscribe(getSubscriptionId()); + } + + @Override + public String toString() { + return "Subscription [id=" + getSubscriptionId() + ", destination='" + getDestination() + + "', receiptId='" + getReceiptId() + "', handler=" + getHandler() + "]"; + } + } + + private class WriteInactivityTask implements Runnable { + + @Override + public void run() { + TcpConnection conn = connection; + if (conn != null) { + conn.send(HEARTBEAT).addCallback( + new ListenableFutureCallback() { + public void onSuccess(Void result) { + } + public void onFailure(Throwable ex) { + handleFailure(ex); + } + }); + } + } + } + + private class ReadInactivityTask implements Runnable { + + @Override + public void run() { + closing = true; + String error = "Server has gone quite. Closing connection in session id=" + sessionId + "."; + if (logger.isDebugEnabled()) { + logger.debug(error); + } + handleFailure(new IllegalStateException(error)); + resetConnection(); + } + } + +} \ No newline at end of file diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/Reactor11TcpStompClient.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/Reactor11TcpStompClient.java new file mode 100644 index 0000000000..0f04c5330b --- /dev/null +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/Reactor11TcpStompClient.java @@ -0,0 +1,127 @@ +/* + * Copyright 2002-2015 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.stomp; + +import java.util.Arrays; +import java.util.Properties; + +import reactor.core.Environment; +import reactor.core.configuration.ConfigurationReader; +import reactor.core.configuration.DispatcherConfiguration; +import reactor.core.configuration.DispatcherType; +import reactor.core.configuration.ReactorConfiguration; +import reactor.net.netty.tcp.NettyTcpClient; +import reactor.net.tcp.TcpClient; +import reactor.net.tcp.spec.TcpClientSpec; + +import org.springframework.messaging.Message; +import org.springframework.messaging.tcp.TcpOperations; +import org.springframework.messaging.tcp.reactor.Reactor11TcpClient; +import org.springframework.util.concurrent.ListenableFuture; +import org.springframework.util.concurrent.SettableListenableFuture; + +/** + * A STOMP over TCP client that uses + * {@link org.springframework.messaging.tcp.reactor.Reactor11TcpClient + * Reactor11TcpClient}. + * + * @author Rossen Stoyanchev + * @since 4.2 + */ +public class Reactor11TcpStompClient extends StompClientSupport { + + private final TcpOperations tcpClient; + + + /** + * Create an instance with host "127.0.0.1" and port 61613. + */ + public Reactor11TcpStompClient() { + this("127.0.0.1", 61613); + } + + /** + * Create an instance with the given host and port. + * @param host the host + * @param port the port + */ + public Reactor11TcpStompClient(String host, int port) { + this.tcpClient = new Reactor11TcpClient(createNettyTcpClient(host, port)); + } + + private TcpClient, Message> createNettyTcpClient(String host, int port) { + return new TcpClientSpec, Message>(NettyTcpClient.class) + .env(new Environment(new StompClientDispatcherConfigReader())) + .codec(new Reactor11StompCodec(new StompEncoder(), new StompDecoder())) + .connect(host, port) + .get(); + } + + /** + * Create an instance with a pre-configured TCP client. + * @param tcpClient the client to use + */ + public Reactor11TcpStompClient(TcpOperations tcpClient) { + this.tcpClient = tcpClient; + } + + + /** + * Connect and notify the given {@link StompSessionHandler} when connected + * on the STOMP level, + * @param handler the handler for the STOMP session + * @return ListenableFuture for access to the session when ready for use + */ + public ListenableFuture connect(StompSessionHandler handler) { + return connect(null, handler); + } + + /** + * An overloaded version of {@link #connect(StompSessionHandler)} that + * accepts headers to use for the STOMP CONNECT frame. + * @param connectHeaders headers to add to the CONNECT frame + * @param handler the handler for the STOMP session + * @return ListenableFuture for access to the session when ready for use + */ + public ListenableFuture connect(StompHeaders connectHeaders, StompSessionHandler handler) { + ConnectionHandlingStompSession session = createSession(connectHeaders, handler); + this.tcpClient.connect(session); + return session.getSessionFuture(); + } + + /** + * Shut down the client and release resources. + */ + public void shutdown() { + this.tcpClient.shutdown(); + } + + + /** + * A ConfigurationReader with a thread pool-based dispatcher. + */ + private static class StompClientDispatcherConfigReader implements ConfigurationReader { + + @Override + public ReactorConfiguration read() { + String dispatcherName = "StompClient"; + DispatcherType dispatcherType = DispatcherType.THREAD_POOL_EXECUTOR; + DispatcherConfiguration config = new DispatcherConfiguration(dispatcherName, dispatcherType, 128, 0); + return new ReactorConfiguration(Arrays.asList(config), dispatcherName, new Properties()); + } + } + +} diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompClientSupport.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompClientSupport.java new file mode 100644 index 0000000000..ecd6d6c6ae --- /dev/null +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompClientSupport.java @@ -0,0 +1,172 @@ +/* + * Copyright 2002-2015 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.stomp; + +import java.util.Arrays; + +import org.springframework.messaging.converter.MessageConverter; +import org.springframework.messaging.converter.StringMessageConverter; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.util.Assert; + + +/** + * Base class for STOMP client implementations. + * + *

Sub-classes can connect over WebSocket or TCP using any library. + * When creating a new connection a sub-class can create an instance of + * {@link DefaultStompSession} which extends + * {@link org.springframework.messaging.tcp.TcpConnectionHandler + * TcpConnectionHandler} whose lifecycle methods the sub-class must then invoke. + * + *

In effect {@code TcpConnectionHandler} and {@code TcpConnection} are the + * contracts any sub-class must adapt to while using {@link StompEncoder} and + * {@link StompDecoder} to encode and decode STOMP messages. + * + * @author Rossen Stoyanchev + * @since 4.2 + */ +public abstract class StompClientSupport { + + private MessageConverter messageConverter = new StringMessageConverter(); + + private TaskScheduler taskScheduler; + + private long[] defaultHeartbeat = new long[] {10000, 10000}; + + private long receiptTimeLimit = 15 * 1000; + + + /** + * Set the {@link MessageConverter} to use to convert the payload of incoming + * and outgoing messages to and from {@code byte[]} based on object type + * and the "content-type" header. + *

By default, {@link StringMessageConverter} is configured. + * @param messageConverter the message converter to use + */ + public void setMessageConverter(MessageConverter messageConverter) { + Assert.notNull(messageConverter, "'messageConverter' must not be null"); + this.messageConverter = messageConverter; + } + + /** + * Return the configured {@link MessageConverter}. + */ + public MessageConverter getMessageConverter() { + return this.messageConverter; + } + + /** + * Configure a scheduler to use for heartbeats and for receipt tracking. + * + *

Note: some transports have built-in support to work + * with heartbeats and therefore do not require a TaskScheduler. + * Receipts however, if needed, do require a TaskScheduler to be configured. + * + *

By default this is not set. + */ + public void setTaskScheduler(TaskScheduler taskScheduler) { + this.taskScheduler = taskScheduler; + } + + /** + * The configured TaskScheduler. + */ + public TaskScheduler getTaskScheduler() { + return this.taskScheduler; + } + + /** + * Configure the default value for the "heart-beat" header of the STOMP + * CONNECT frame. The first number represents how often the client will write + * or send a heart-beat. The second is how often the server should write. + * A value of 0 means no heart-beats. + *

By default this is set to "10000,10000" but sub-classes may override + * that default and for example set it to "0,0" if they require a + * TaskScheduler to be configured first. + * @param heartbeat the value for the CONNECT "heart-beat" header + * @see + * http://stomp.github.io/stomp-specification-1.2.html#Heart-beating + */ + public void setDefaultHeartbeat(long[] heartbeat) { + Assert.notNull(heartbeat); + Assert.isTrue(heartbeat[0] >= 0 && heartbeat[1] >=0 , "Invalid heart-beat: " + Arrays.toString(heartbeat)); + this.defaultHeartbeat = heartbeat; + } + + /** + * Return the configured default heart-beat value, never {@code null}. + */ + public long[] getDefaultHeartbeat() { + return this.defaultHeartbeat; + } + + /** + * Whether heartbeats are enabled. Returns {@code false} if + * {@link #setDefaultHeartbeat defaultHeartbeat} is set to "0,0", and + * {@code true} otherwise. + */ + public boolean isDefaultHeartbeatEnabled() { + return (getDefaultHeartbeat() != null && getDefaultHeartbeat()[0] != 0 && getDefaultHeartbeat()[1] != 0); + } + + /** + * Configure the number of milliseconds before a receipt is considered expired. + *

By default set to 15,000 (15 seconds). + */ + public void setReceiptTimeLimit(long receiptTimeLimit) { + Assert.isTrue(receiptTimeLimit > 0); + this.receiptTimeLimit = receiptTimeLimit; + } + + /** + * Return the configured receipt time limit. + */ + public long getReceiptTimeLimit() { + return receiptTimeLimit; + } + + + /** + * Factory method for create and configure a new session. + * @param connectHeaders headers for the STOMP CONNECT frame + * @param handler the handler for the STOMP session + * @return the created session + */ + protected ConnectionHandlingStompSession createSession(StompHeaders connectHeaders, StompSessionHandler handler) { + connectHeaders = processConnectHeaders(connectHeaders); + DefaultStompSession session = new DefaultStompSession(handler, connectHeaders); + session.setMessageConverter(getMessageConverter()); + session.setTaskScheduler(getTaskScheduler()); + session.setReceiptTimeLimit(getReceiptTimeLimit()); + return session; + } + + /** + * Further initialize the StompHeaders, for example setting the heart-beat + * header if necessary. + * @param connectHeaders the headers to modify + * @return the modified headers + */ + protected StompHeaders processConnectHeaders(StompHeaders connectHeaders) { + connectHeaders = (connectHeaders != null ? connectHeaders : new StompHeaders()); + if (connectHeaders.getHeartbeat() == null) { + connectHeaders.setHeartbeat(getDefaultHeartbeat()); + } + return connectHeaders; + } + +} diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompEncoder.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompEncoder.java index bc197db52a..228fbcdb57 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompEncoder.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompEncoder.java @@ -73,6 +73,9 @@ public final class StompEncoder { DataOutputStream output = new DataOutputStream(baos); if (SimpMessageType.HEARTBEAT.equals(SimpMessageHeaderAccessor.getMessageType(headers))) { + if (logger.isTraceEnabled()) { + logger.trace("Encoding heartbeat"); + } output.write(StompDecoder.HEARTBEAT_PAYLOAD); } else { diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompFrameHandler.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompFrameHandler.java new file mode 100644 index 0000000000..3a5cff050c --- /dev/null +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompFrameHandler.java @@ -0,0 +1,45 @@ +/* + * Copyright 2002-2015 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.stomp; + +import java.lang.reflect.Type; + + +/** + * Contract to handle a STOMP frame. + * + * @author Rossen Stoyanchev + * @since 4.2 + */ +public interface StompFrameHandler { + + /** + * Invoked before {@link #handleFrame(StompHeaders, Object)} to determine the + * type of Object the payload should be converted to. + * @param headers the headers of a message + */ + Type getPayloadType(StompHeaders headers); + + /** + * Handle a STOMP frame with the payload converted to the target type returned + * from {@link #getPayloadType(StompHeaders)}. + * + * @param headers the headers of the frame + * @param payload the payload or {@code null} if there was no payload + */ + void handleFrame(StompHeaders headers, Object payload); + +} diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompHeaderAccessor.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompHeaderAccessor.java index 4747c9dfdf..3538a63935 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompHeaderAccessor.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompHeaderAccessor.java @@ -162,7 +162,7 @@ public class StompHeaderAccessor extends SimpMessageHeaderAccessor { } } - private void updateStompHeadersFromSimpMessageHeaders() { + void updateStompHeadersFromSimpMessageHeaders() { if (getDestination() != null) { setNativeHeader(STOMP_DESTINATION_HEADER, getDestination()); } diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompHeaders.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompHeaders.java new file mode 100644 index 0000000000..166fcd125b --- /dev/null +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompHeaders.java @@ -0,0 +1,528 @@ +/* + * Copyright 2002-2015 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.stomp; + +import java.io.Serializable; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.springframework.util.Assert; +import org.springframework.util.LinkedMultiValueMap; +import org.springframework.util.MimeType; +import org.springframework.util.MimeTypeUtils; +import org.springframework.util.MultiValueMap; +import org.springframework.util.StringUtils; + +/** + * Represents STOMP frame headers. + * + *

In addition to the normal methods defined by {@link Map}, this class offers + * the following convenience methods: + *

+ * + * @author Rossen Stoyanchev + * @since 4.2 + * @see + * http://stomp.github.io/stomp-specification-1.2.html#Frames_and_Headers + */ +public class StompHeaders implements MultiValueMap, Serializable { + + private static final long serialVersionUID = 7514642206528452544L; + + + // Standard headers (as defined in the spec) + + public static final String CONTENT_TYPE = "content-type"; // SEND, MESSAGE, ERROR + + public static final String CONTENT_LENGTH = "content-length"; // SEND, MESSAGE, ERROR + + public static final String RECEIPT = "receipt"; // any client frame other than CONNECT + + // CONNECT + + public static final String HOST = "host"; + + public static final String LOGIN = "login"; + + public static final String PASSCODE = "passcode"; + + public static final String HEARTBEAT = "heart-beat"; + + // CONNECTED + + public static final String SESSION = "session"; + + public static final String SERVER = "server"; + + // SEND + + public static final String DESTINATION = "destination"; + + // SUBSCRIBE, UNSUBSCRIBE + + public static final String ID = "id"; + + public static final String ACK = "ack"; + + // MESSAGE + + public static final String SUBSCRIPTION = "subscription"; + + public static final String MESSAGE_ID = "message-id"; + + // RECEIPT + + public static final String RECEIPT_ID = "receipt-id"; + + + private final Map> headers; + + + /** + * Create a new instance to be populated with new header values. + */ + public StompHeaders() { + this(new LinkedMultiValueMap(4), false); + } + + private StompHeaders(Map> headers, boolean readOnly) { + Assert.notNull(headers, "'headers' must not be null"); + if (readOnly) { + Map> map = new LinkedMultiValueMap(headers.size()); + for (Entry> entry : headers.entrySet()) { + List values = Collections.unmodifiableList(entry.getValue()); + map.put(entry.getKey(), values); + } + this.headers = Collections.unmodifiableMap(map); + } + else { + this.headers = headers; + } + } + + + /** + * Set the content-type header. + * Applies to the SEND, MESSAGE, and ERROR frames. + */ + public void setContentType(MimeType mimeType) { + Assert.isTrue(!mimeType.isWildcardType(), "'Content-Type' cannot contain wildcard type '*'"); + Assert.isTrue(!mimeType.isWildcardSubtype(), "'Content-Type' cannot contain wildcard subtype '*'"); + set(CONTENT_TYPE, mimeType.toString()); + } + + /** + * Return the content-type header value. + */ + public MimeType getContentType() { + String value = getFirst(CONTENT_TYPE); + return (StringUtils.hasLength(value) ? MimeTypeUtils.parseMimeType(value) : null); + } + + /** + * Set the content-length header. + * Applies to the SEND, MESSAGE, and ERROR frames. + */ + public void setContentLength(long contentLength) { + set(CONTENT_LENGTH, Long.toString(contentLength)); + } + + /** + * Return the content-length header or -1 if unknown. + */ + public long getContentLength() { + String value = getFirst(CONTENT_LENGTH); + return (value != null ? Long.parseLong(value) : -1); + } + + /** + * Set the receipt header. + * Applies to any client frame other than CONNECT. + */ + public void setReceipt(String receipt) { + set(RECEIPT, receipt); + } + + /** + * Get the receipt header. + */ + public String getReceipt() { + return getFirst(RECEIPT); + } + + /** + * Set the host header. + * Applies to the CONNECT frame. + */ + public void setHost(String host) { + set(HOST, host); + } + + /** + * Get the host header. + */ + public String getHost() { + return getFirst(HOST); + } + + /** + * Set the login header. + * Applies to the CONNECT frame. + */ + public void setLogin(String login) { + set(LOGIN, login); + } + + /** + * Get the login header. + */ + public String getLogin() { + return getFirst(LOGIN); + } + + /** + * Set the passcode header. + * Applies to the CONNECT frame. + */ + public void setPasscode(String passcode) { + set(PASSCODE, passcode); + } + + /** + * Get the passcode header. + */ + public String getPasscode() { + return getFirst(PASSCODE); + } + + /** + * Set the heartbeat header. + * Applies to the CONNECT and CONNECTED frames. + */ + public void setHeartbeat(long[] heartbeat) { + Assert.notNull(heartbeat); + String value = heartbeat[0] + "," + heartbeat[1]; + Assert.isTrue(heartbeat[0] >= 0 && heartbeat[1] >= 0, "Heart-beat values cannot be negative: " + value); + set(HEARTBEAT, value); + } + + /** + * Get the heartbeat header. + */ + public long[] getHeartbeat() { + String rawValue = getFirst(HEARTBEAT); + if (!StringUtils.hasText(rawValue)) { + return null; + } + String[] rawValues = StringUtils.commaDelimitedListToStringArray(rawValue); + return new long[] {Long.valueOf(rawValues[0]), Long.valueOf(rawValues[1])}; + } + + /** + * Whether heartbeats are enabled. Returns {@code false} if + * {@link #setHeartbeat} is set to "0,0", and {@code true} otherwise. + */ + public boolean isHeartbeatEnabled() { + long[] heartbeat = getHeartbeat(); + return (heartbeat != null && heartbeat[0] != 0 && heartbeat[1] != 0); + } + + /** + * Set the session header. + * Applies to the CONNECTED frame. + */ + public void setSession(String session) { + set(SESSION, session); + } + + /** + * Get the session header. + */ + public String getSession() { + return getFirst(SESSION); + } + + /** + * Set the server header. + * Applies to the CONNECTED frame. + */ + public void setServer(String server) { + set(SERVER, server); + } + + /** + * Get the server header. + * Applies to the CONNECTED frame. + */ + public String getServer() { + return getFirst(SERVER); + } + + /** + * Set the destination header. + */ + public void setDestination(String destination) { + set(DESTINATION, destination); + } + + /** + * Get the destination header. + * Applies to the SEND, SUBSCRIBE, and MESSAGE frames. + */ + public String getDestination() { + return getFirst(DESTINATION); + } + + /** + * Set the id header. + * Applies to the SUBSCR0BE, UNSUBSCRIBE, and ACK or NACK frames. + */ + public void setId(String id) { + set(ID, id); + } + + /** + * Get the id header. + */ + public String getId() { + return getFirst(ID); + } + + /** + * Set the ack header to one of "auto", "client", or "client-individual". + * Applies to the SUBSCRIBE and MESSAGE frames. + */ + public void setAck(String ack) { + set(ACK, ack); + } + + /** + * Get the ack header. + */ + public String getAck() { + return getFirst(ACK); + } + + /** + * Set the login header. + * Applies to the MESSAGE frame. + */ + public void setSubscription(String subscription) { + set(SUBSCRIPTION, subscription); + } + + /** + * Get the subscription header. + */ + public String getSubscription() { + return getFirst(SUBSCRIPTION); + } + + /** + * Set the message-id header. + * Applies to the MESSAGE frame. + */ + public void setMessageId(String messageId) { + set(MESSAGE_ID, messageId); + } + + /** + * Get the message-id header. + */ + public String getMessageId() { + return getFirst(MESSAGE_ID); + } + + /** + * Set the receipt-id header. + * Applies to the RECEIPT frame. + */ + public void setReceiptId(String receiptId) { + set(RECEIPT_ID, receiptId); + } + + /** + * Get the receipt header. + */ + public String getReceiptId() { + return getFirst(RECEIPT_ID); + } + + /** + * Return the first header value for the given header name, if any. + * @param headerName the header name + * @return the first header value, or {@code null} if none + */ + @Override + public String getFirst(String headerName) { + List headerValues = headers.get(headerName); + return headerValues != null ? headerValues.get(0) : null; + } + + /** + * Add the given, single header value under the given name. + * @param headerName the header name + * @param headerValue the header value + * @throws UnsupportedOperationException if adding headers is not supported + * @see #put(String, List) + * @see #set(String, String) + */ + @Override + public void add(String headerName, String headerValue) { + List headerValues = headers.get(headerName); + if (headerValues == null) { + headerValues = new LinkedList(); + this.headers.put(headerName, headerValues); + } + headerValues.add(headerValue); + } + + /** + * Set the given, single header value under the given name. + * @param headerName the header name + * @param headerValue the header value + * @throws UnsupportedOperationException if adding headers is not supported + * @see #put(String, List) + * @see #add(String, String) + */ + @Override + public void set(String headerName, String headerValue) { + List headerValues = new LinkedList(); + headerValues.add(headerValue); + headers.put(headerName, headerValues); + } + + @Override + public void setAll(Map values) { + for (Entry entry : values.entrySet()) { + set(entry.getKey(), entry.getValue()); + } + } + + @Override + public Map toSingleValueMap() { + LinkedHashMap singleValueMap = new LinkedHashMap(this.headers.size()); + for (Entry> entry : headers.entrySet()) { + singleValueMap.put(entry.getKey(), entry.getValue().get(0)); + } + return singleValueMap; + } + + + // Map implementation + + @Override + public int size() { + return this.headers.size(); + } + + @Override + public boolean isEmpty() { + return this.headers.isEmpty(); + } + + @Override + public boolean containsKey(Object key) { + return this.headers.containsKey(key); + } + + @Override + public boolean containsValue(Object value) { + return this.headers.containsValue(value); + } + + @Override + public List get(Object key) { + return this.headers.get(key); + } + + @Override + public List put(String key, List value) { + return this.headers.put(key, value); + } + + @Override + public List remove(Object key) { + return this.headers.remove(key); + } + + @Override + public void putAll(Map> map) { + this.headers.putAll(map); + } + + @Override + public void clear() { + this.headers.clear(); + } + + @Override + public Set keySet() { + return this.headers.keySet(); + } + + @Override + public Collection> values() { + return this.headers.values(); + } + + @Override + public Set>> entrySet() { + return this.headers.entrySet(); + } + + + @Override + public boolean equals(Object other) { + if (this == other) { + return true; + } + if (!(other instanceof StompHeaders)) { + return false; + } + StompHeaders otherHeaders = (StompHeaders) other; + return this.headers.equals(otherHeaders.headers); + } + + @Override + public int hashCode() { + return this.headers.hashCode(); + } + + @Override + public String toString() { + return this.headers.toString(); + } + + + /** + * Return a {@code StompHeaders} object that can only be read, not written to. + */ + public static StompHeaders readOnlyStompHeaders(Map> headers) { + return new StompHeaders(headers, true); + } + +} diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompSession.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompSession.java new file mode 100644 index 0000000000..5df7ca1d7f --- /dev/null +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompSession.java @@ -0,0 +1,140 @@ +/* + * Copyright 2002-2015 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.stomp; + + +/** + * Represents a STOMP session with operations to send messages, create + * subscriptions and receive messages on those subscriptions. + * + * @author Rossen Stoyanchev + * @since 4.2 + */ +public interface StompSession { + + /** + * Return the id for the session. + */ + String getSessionId(); + + /** + * Whether the session is connected. + */ + boolean isConnected(); + + /** + * When enabled, a receipt header is automatically added to future + * {@code send} and {@code subscribe} operations on this session, which causes + * the server to return a RECEIPT. An application can then use the + * {@link StompSession.Receiptable + * Receiptable} returned from the operation to track the receipt. + * + *

A receipt header can also be added manually through the overloaded + * methods that accept {@code StompHeaders}. + */ + void setAutoReceipt(boolean enabled); + + /** + * Send a message to the specified destination, converting the payload to a + * {@code byte[]} with the help of a + * {@link org.springframework.messaging.converter.MessageConverter MessageConverter}. + * @param destination the destination to send a message to + * @param payload the message payload + * @return a Receiptable for tracking receipts + */ + Receiptable send(String destination, Object payload); + + /** + * An overloaded version of {@link #send(String, Object)} that accepts + * full {@link StompHeaders} instead of a destination. The headers must + * contain a destination and may also have other headers such as + * "content-type" or custom headers for the broker to propagate to subscribers, + * or broker-specific, non-standard headers.. + * @param headers the message headers + * @param payload the message payload + * @return a Receiptable for tracking receipts + */ + Receiptable send(StompHeaders headers, Object payload); + + /** + * Subscribe to the given destination by sending a SUBSCRIBE frame and handle + * received messages with the specified {@link StompFrameHandler}. + * @param destination the destination to subscribe to + * @param handler the handler for received messages + * @return a handle to use to unsubscribe and/or track receipts + */ + Subscription subscribe(String destination, StompFrameHandler handler); + + /** + * An overloaded version of {@link #subscribe(String, StompFrameHandler)} + * that accepts full {@link StompHeaders} rather instead of a destination. + * @param headers the headers for the subscribe message frame + * @param handler the handler for received messages + * @return a handle to use to unsubscribe and/or track receipts + */ + Subscription subscribe(StompHeaders headers, StompFrameHandler handler); + + /** + * Disconnect the session by sending a DISCONNECT frame. + */ + void disconnect(); + + + /** + * A handle to use to track receipts. + * @see #setAutoReceipt(boolean) + */ + interface Receiptable { + + /** + * Return the receipt id, or {@code null} if the STOMP frame for which + * the handle was returned did not have a "receipt" header. + */ + String getReceiptId(); + + /** + * Task to invoke when a receipt is received. + * @throws java.lang.IllegalArgumentException if the receiptId is {@code null} + */ + void addReceiptTask(Runnable runnable); + + /** + * Task to invoke when a receipt is not received in the configured time. + * @throws java.lang.IllegalArgumentException if the receiptId is {@code null} + * @see org.springframework.messaging.simp.stomp.StompClientSupport#setReceiptTimeLimit(long) + */ + void addReceiptLostTask(Runnable runnable); + + } + + /** + * A handle to use to unsubscribe or to track a receipt. + */ + interface Subscription extends Receiptable { + + /** + * Return the id for the subscription. + */ + String getSubscriptionId(); + + /** + * Remove the subscription by sending an UNSUBSCRIBE frame. + */ + void unsubscribe(); + + } + +} diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompSessionHandler.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompSessionHandler.java new file mode 100644 index 0000000000..e52ee0ce99 --- /dev/null +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompSessionHandler.java @@ -0,0 +1,71 @@ +/* + * Copyright 2002-2015 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.stomp; + + +/** + * A contract for client STOMP session lifecycle events including a callback + * when the session is established and notifications of transport or message + * handling failures. + * + *

This contract also extends {@link StompFrameHandler} in order to handle + * STOMP ERROR frames received from the broker. + * + *

Implementations of this interface should consider extending + * {@link StompSessionHandlerAdapter}. + * + * @author Rossen Stoyanchev + * @since 4.2 + * @see StompSessionHandlerAdapter + */ +public interface StompSessionHandler extends StompFrameHandler { + + /** + * Invoked when the session is ready to use, i.e. after the underlying + * transport (TCP, WebSocket) is connected and a STOMP CONNECTED frame is + * received from the broker. + * @param session the client STOMP session + * @param connectedHeaders the STOMP CONNECTED frame headers + */ + void afterConnected(StompSession session, StompHeaders connectedHeaders); + + /** + * Handle any exception arising while processing a STOMP frame such as a + * failure to convert the payload or an unhandled exception in the + * application {@code StompFrameHandler}. + * @param session the client STOMP session + * @param command the STOMP command of the frame + * @param headers the headers + * @param payload the raw payload + * @param exception the exception + */ + void handleException(StompSession session, StompCommand command, StompHeaders headers, + byte[] payload, Throwable exception); + + /** + * Handle a low level transport error which could be an I/O error or a + * failure to encode or decode a STOMP message. + *

Note that + * {@link org.springframework.messaging.simp.stomp.ConnectionLostException + * ConnectionLostException} will be passed into this method when the + * connection is lost rather than closed normally via + * {@link StompSession#disconnect()}. + * @param session the client STOMP session + * @param exception the exception that occurred + */ + void handleTransportError(StompSession session, Throwable exception); + +} diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompSessionHandlerAdapter.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompSessionHandlerAdapter.java new file mode 100644 index 0000000000..d4137806ec --- /dev/null +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompSessionHandlerAdapter.java @@ -0,0 +1,68 @@ +/* + * Copyright 2002-2015 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.stomp; + +import java.lang.reflect.Type; + +/** + * Abstract adapter class for {@link StompSessionHandler} with mostly empty + * implementation methods except for {@link #getPayloadType} which returns String + * as the default Object type expected for STOMP ERROR frame payloads. + * + * @author Rossen Stoyanchev + * @since 4.2 + */ +public abstract class StompSessionHandlerAdapter implements StompSessionHandler { + + /** + * This implementation is empty. + */ + @Override + public void afterConnected(StompSession session, StompHeaders connectedHeaders) { + } + + /** + * This implementation returns String as the expected payload type + * for STOMP ERROR frames. + */ + @Override + public Type getPayloadType(StompHeaders headers) { + return String.class; + } + + /** + * This implementation is empty. + */ + @Override + public void handleFrame(StompHeaders headers, Object payload) { + } + + /** + * This implementation is empty. + */ + @Override + public void handleException(StompSession session, StompCommand command, StompHeaders headers, + byte[] payload, Throwable exception) { + } + + /** + * This implementation is empty. + */ + @Override + public void handleTransportError(StompSession session, Throwable exception) { + } + +} diff --git a/spring-messaging/src/main/java/org/springframework/messaging/support/NativeMessageHeaderAccessor.java b/spring-messaging/src/main/java/org/springframework/messaging/support/NativeMessageHeaderAccessor.java index db8a534a17..881d38e1c2 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/support/NativeMessageHeaderAccessor.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/support/NativeMessageHeaderAccessor.java @@ -25,6 +25,7 @@ import org.springframework.messaging.Message; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; import org.springframework.util.LinkedMultiValueMap; +import org.springframework.util.MultiValueMap; import org.springframework.util.ObjectUtils; /** @@ -186,6 +187,17 @@ public class NativeMessageHeaderAccessor extends MessageHeaderAccessor { setModified(true); } + public void addNativeHeaders(MultiValueMap headers) { + if (headers == null) { + return; + } + for (String header : headers.keySet()) { + for (String value : headers.get(header)) { + addNativeHeader(header, value); + } + } + } + public List removeNativeHeader(String name) { Assert.state(isMutable(), "Already immutable"); Map> nativeHeaders = getNativeHeaders(); diff --git a/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/Reactor11TcpClient.java b/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/Reactor11TcpClient.java index 093147fb57..a38eb324ff 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/Reactor11TcpClient.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/Reactor11TcpClient.java @@ -196,7 +196,9 @@ public class Reactor11TcpClient

implements TcpOperations

{ }; } finally { - this.environment.shutdown(); + if (this.environment != null) { + this.environment.shutdown(); + } } } diff --git a/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/DefaultStompSessionTests.java b/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/DefaultStompSessionTests.java new file mode 100644 index 0000000000..77533c1a94 --- /dev/null +++ b/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/DefaultStompSessionTests.java @@ -0,0 +1,586 @@ +/* + * Copyright 2002-2015 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.stomp; + +import static org.hamcrest.Matchers.*; +import static org.junit.Assert.*; +import static org.mockito.Matchers.any; +import static org.mockito.Mockito.eq; +import static org.mockito.Mockito.*; +import static org.mockito.Mockito.same; + +import java.nio.charset.Charset; +import java.util.Date; +import java.util.Map; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.atomic.AtomicReference; + +import org.hamcrest.Matchers; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.ExpectedException; +import org.mockito.ArgumentCaptor; +import org.mockito.Captor; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; + +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageDeliveryException; +import org.springframework.messaging.converter.MessageConversionException; +import org.springframework.messaging.simp.stomp.StompSession.Receiptable; +import org.springframework.messaging.simp.stomp.StompSession.Subscription; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.messaging.support.MessageHeaderAccessor; +import org.springframework.messaging.tcp.TcpConnection; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.util.MimeType; +import org.springframework.util.MimeTypeUtils; +import org.springframework.util.concurrent.SettableListenableFuture; + +/** + * Unit tests for {@link DefaultStompSession}. + * + * @author Rossen Stoyanchev + */ +public class DefaultStompSessionTests { + + private static Charset UTF_8 = Charset.forName("UTF-8"); + + + private DefaultStompSession session; + + @Mock + private StompSessionHandler sessionHandler; + + private StompHeaders connectHeaders; + + @Mock + private TcpConnection connection; + + @Captor + private ArgumentCaptor> messageCaptor; + + @Rule + public ExpectedException expected = ExpectedException.none(); + + + @Before + public void setUp() throws Exception { + + MockitoAnnotations.initMocks(this); + + this.sessionHandler = mock(StompSessionHandler.class); + this.connectHeaders = new StompHeaders(); + this.session = new DefaultStompSession(this.sessionHandler, this.connectHeaders); + + SettableListenableFuture future = new SettableListenableFuture<>(); + future.set(null); + when(this.connection.send(this.messageCaptor.capture())).thenReturn(future); + } + + + @Test + public void afterConnected() throws Exception { + + assertFalse(this.session.isConnected()); + this.connectHeaders.setHost("my-host"); + this.connectHeaders.setHeartbeat(new long[] {11, 12}); + + this.session.afterConnected(this.connection); + assertTrue(this.session.isConnected()); + + Message message = this.messageCaptor.getValue(); + StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class); + assertEquals(StompCommand.CONNECT, accessor.getCommand()); + assertEquals("my-host", accessor.getHost()); + assertThat(accessor.getAcceptVersion(), containsInAnyOrder("1.1", "1.2")); + assertArrayEquals(new long[] {11, 12}, accessor.getHeartbeat()); + } + + @Test + public void afterConnectFailure() throws Exception { + IllegalStateException exception = new IllegalStateException("simulated exception"); + this.session.afterConnectFailure(exception); + verify(this.sessionHandler).handleTransportError(this.session, exception); + verifyNoMoreInteractions(this.sessionHandler); + } + + @Test + public void handleConnectedFrame() throws Exception { + + this.session.afterConnected(this.connection); + assertTrue(this.session.isConnected()); + + this.connectHeaders.setHeartbeat(new long[] {10000, 10000}); + + StompHeaderAccessor accessor = StompHeaderAccessor.create(StompCommand.CONNECTED); + accessor.setVersion("1.2"); + accessor.setHeartbeat(10000, 10000); + accessor.setLeaveMutable(true); + this.session.handleMessage(MessageBuilder.createMessage(new byte[0], accessor.getMessageHeaders())); + + StompHeaders stompHeaders = StompHeaders.readOnlyStompHeaders(accessor.getNativeHeaders()); + verify(this.sessionHandler).afterConnected(this.session, stompHeaders); + verifyNoMoreInteractions(this.sessionHandler); + } + + @Test + public void heartbeatValues() throws Exception { + + this.session.afterConnected(this.connection); + assertTrue(this.session.isConnected()); + + this.connectHeaders.setHeartbeat(new long[] {10000, 10000}); + + StompHeaderAccessor accessor = StompHeaderAccessor.create(StompCommand.CONNECTED); + accessor.setVersion("1.2"); + accessor.setHeartbeat(20000, 20000); + accessor.setLeaveMutable(true); + this.session.handleMessage(MessageBuilder.createMessage(new byte[0], accessor.getMessageHeaders())); + + ArgumentCaptor writeInterval = ArgumentCaptor.forClass(Long.class); + verify(this.connection).onWriteInactivity(any(Runnable.class), writeInterval.capture()); + assertEquals(20000, (long) writeInterval.getValue()); + + ArgumentCaptor readInterval = ArgumentCaptor.forClass(Long.class); + verify(this.connection).onReadInactivity(any(Runnable.class), readInterval.capture()); + assertEquals(60000, (long) readInterval.getValue()); + } + + @Test + public void heartbeatNotSupportedByServer() throws Exception { + + this.session.afterConnected(this.connection); + verify(this.connection).send(any()); + + this.connectHeaders.setHeartbeat(new long[] {10000, 10000}); + + StompHeaderAccessor accessor = StompHeaderAccessor.create(StompCommand.CONNECTED); + accessor.setVersion("1.2"); + accessor.setHeartbeat(0, 0); + accessor.setLeaveMutable(true); + this.session.handleMessage(MessageBuilder.createMessage(new byte[0], accessor.getMessageHeaders())); + + verifyNoMoreInteractions(this.connection); + } + + @Test + public void heartbeatTasks() throws Exception { + + this.session.afterConnected(this.connection); + verify(this.connection).send(any()); + + this.connectHeaders.setHeartbeat(new long[] {10000, 10000}); + + StompHeaderAccessor connected = StompHeaderAccessor.create(StompCommand.CONNECTED); + connected.setVersion("1.2"); + connected.setHeartbeat(10000, 10000); + connected.setLeaveMutable(true); + this.session.handleMessage(MessageBuilder.createMessage(new byte[0], connected.getMessageHeaders())); + + ArgumentCaptor writeTaskCaptor = ArgumentCaptor.forClass(Runnable.class); + ArgumentCaptor readTaskCaptor = ArgumentCaptor.forClass(Runnable.class); + verify(this.connection).onWriteInactivity(writeTaskCaptor.capture(), any(Long.class)); + verify(this.connection).onReadInactivity(readTaskCaptor.capture(), any(Long.class)); + + Runnable writeTask = writeTaskCaptor.getValue(); + Runnable readTask = readTaskCaptor.getValue(); + assertNotNull(writeTask); + assertNotNull(readTask); + + writeTask.run(); + StompHeaderAccessor accessor = StompHeaderAccessor.createForHeartbeat(); + Message message = MessageBuilder.createMessage(new byte[] {'\n'}, accessor.getMessageHeaders()); + verify(this.connection).send(eq(message)); + verifyNoMoreInteractions(this.connection); + + reset(this.sessionHandler); + readTask.run(); + verify(this.sessionHandler).handleTransportError(same(this.session), any(IllegalStateException.class)); + verifyNoMoreInteractions(this.sessionHandler); + } + + @Test + public void handleErrorFrame() throws Exception { + + StompHeaderAccessor accessor = StompHeaderAccessor.create(StompCommand.ERROR); + accessor.setContentType(new MimeType("text", "plain", UTF_8)); + accessor.addNativeHeader("foo", "bar"); + accessor.setLeaveMutable(true); + String payload = "Oops"; + + StompHeaders stompHeaders = StompHeaders.readOnlyStompHeaders(accessor.getNativeHeaders()); + when(this.sessionHandler.getPayloadType(stompHeaders)).thenReturn(String.class); + + this.session.handleMessage(MessageBuilder.createMessage(payload.getBytes(UTF_8), accessor.getMessageHeaders())); + + verify(this.sessionHandler).getPayloadType(stompHeaders); + verify(this.sessionHandler).handleFrame(stompHeaders, payload); + verifyNoMoreInteractions(this.sessionHandler); + } + + @Test + public void handleErrorFrameWithEmptyPayload() throws Exception { + + StompHeaderAccessor accessor = StompHeaderAccessor.create(StompCommand.ERROR); + accessor.addNativeHeader("foo", "bar"); + accessor.setLeaveMutable(true); + this.session.handleMessage(MessageBuilder.createMessage(new byte[0], accessor.getMessageHeaders())); + + StompHeaders stompHeaders = StompHeaders.readOnlyStompHeaders(accessor.getNativeHeaders()); + verify(this.sessionHandler).handleFrame(stompHeaders, null); + verifyNoMoreInteractions(this.sessionHandler); + } + + @Test + public void handleErrorFrameWithConversionException() throws Exception { + + StompHeaderAccessor accessor = StompHeaderAccessor.create(StompCommand.ERROR); + accessor.setContentType(MimeTypeUtils.APPLICATION_JSON); + accessor.addNativeHeader("foo", "bar"); + accessor.setLeaveMutable(true); + byte[] payload = "{'foo':'bar'}".getBytes(UTF_8); + + StompHeaders stompHeaders = StompHeaders.readOnlyStompHeaders(accessor.getNativeHeaders()); + when(this.sessionHandler.getPayloadType(stompHeaders)).thenReturn(Map.class); + + this.session.handleMessage(MessageBuilder.createMessage(payload, accessor.getMessageHeaders())); + + verify(this.sessionHandler).getPayloadType(stompHeaders); + verify(this.sessionHandler).handleException(same(this.session), same(StompCommand.ERROR), + eq(stompHeaders), same(payload), any(MessageConversionException.class)); + verifyNoMoreInteractions(this.sessionHandler); + } + + @Test + public void handleMessageFrame() throws Exception { + + this.session.afterConnected(this.connection); + + StompFrameHandler frameHandler = mock(StompFrameHandler.class); + String destination = "/topic/foo"; + Subscription subscription = this.session.subscribe(destination, frameHandler); + + StompHeaderAccessor accessor = StompHeaderAccessor.create(StompCommand.MESSAGE); + accessor.setDestination(destination); + accessor.setSubscriptionId(subscription.getSubscriptionId()); + accessor.setContentType(MimeTypeUtils.TEXT_PLAIN); + accessor.setMessageId("1"); + accessor.setLeaveMutable(true); + String payload = "sample payload"; + + StompHeaders stompHeaders = StompHeaders.readOnlyStompHeaders(accessor.getNativeHeaders()); + when(frameHandler.getPayloadType(stompHeaders)).thenReturn(String.class); + + this.session.handleMessage(MessageBuilder.createMessage(payload.getBytes(UTF_8), accessor.getMessageHeaders())); + + verify(frameHandler).getPayloadType(stompHeaders); + verify(frameHandler).handleFrame(stompHeaders, payload); + verifyNoMoreInteractions(frameHandler); + } + + @Test + public void handleMessageFrameWithConversionException() throws Exception { + + this.session.afterConnected(this.connection); + assertTrue(this.session.isConnected()); + + StompFrameHandler frameHandler = mock(StompFrameHandler.class); + String destination = "/topic/foo"; + Subscription subscription = this.session.subscribe(destination, frameHandler); + + StompHeaderAccessor accessor = StompHeaderAccessor.create(StompCommand.MESSAGE); + accessor.setDestination(destination); + accessor.setSubscriptionId(subscription.getSubscriptionId()); + accessor.setContentType(MimeTypeUtils.APPLICATION_JSON); + accessor.setMessageId("1"); + accessor.setLeaveMutable(true); + byte[] payload = "{'foo':'bar'}".getBytes(UTF_8); + + StompHeaders stompHeaders = StompHeaders.readOnlyStompHeaders(accessor.getNativeHeaders()); + when(frameHandler.getPayloadType(stompHeaders)).thenReturn(Map.class); + + this.session.handleMessage(MessageBuilder.createMessage(payload, accessor.getMessageHeaders())); + + verify(frameHandler).getPayloadType(stompHeaders); + verifyNoMoreInteractions(frameHandler); + + verify(this.sessionHandler).handleException(same(this.session), same(StompCommand.MESSAGE), + eq(stompHeaders), same(payload), any(MessageConversionException.class)); + verifyNoMoreInteractions(this.sessionHandler); + } + + @Test + public void handleFailure() throws Exception { + IllegalStateException exception = new IllegalStateException("simulated exception"); + this.session.handleFailure(exception); + + verify(this.sessionHandler).handleTransportError(this.session, exception); + verifyNoMoreInteractions(this.sessionHandler); + } + + @Test + public void afterConnectionClosed() throws Exception { + this.session.afterConnectionClosed(); + + verify(this.sessionHandler).handleTransportError(same(this.session), any(ConnectionLostException.class)); + verifyNoMoreInteractions(this.sessionHandler); + } + + @Test + public void send() throws Exception { + + this.session.afterConnected(this.connection); + assertTrue(this.session.isConnected()); + + String destination = "/topic/foo"; + String payload = "sample payload"; + this.session.send(destination, payload); + + Message message = this.messageCaptor.getValue(); + StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class); + assertEquals(StompCommand.SEND, accessor.getCommand()); + + StompHeaders stompHeaders = StompHeaders.readOnlyStompHeaders(accessor.getNativeHeaders()); + assertEquals(stompHeaders.toString(), 2, stompHeaders.size()); + + assertEquals(destination, stompHeaders.getDestination()); + assertEquals(new MimeType("text", "plain", UTF_8), stompHeaders.getContentType()); + assertEquals(-1, stompHeaders.getContentLength()); // StompEncoder isn't involved + assertEquals(payload, new String(message.getPayload(), UTF_8)); + } + + @Test + public void sendWithReceipt() throws Exception { + + this.session.afterConnected(this.connection); + assertTrue(this.session.isConnected()); + + this.session.setTaskScheduler(mock(TaskScheduler.class)); + this.session.setAutoReceipt(true); + this.session.send("/topic/foo", "sample payload"); + + Message message = this.messageCaptor.getValue(); + StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class); + assertNotNull(accessor.getReceipt()); + + StompHeaders stompHeaders = new StompHeaders(); + stompHeaders.setDestination("/topic/foo"); + stompHeaders.setReceipt("my-receipt"); + this.session.send(stompHeaders, "sample payload"); + + message = this.messageCaptor.getValue(); + accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class); + assertEquals("my-receipt", accessor.getReceipt()); + } + + @Test + public void sendWithConversionException() throws Exception { + + this.session.afterConnected(this.connection); + assertTrue(this.session.isConnected()); + + StompHeaders stompHeaders = new StompHeaders(); + stompHeaders.setDestination("/topic/foo"); + stompHeaders.setContentType(MimeTypeUtils.APPLICATION_JSON); + String payload = "{'foo':'bar'}"; + + this.expected.expect(MessageConversionException.class); + this.session.send(stompHeaders, payload); + verifyNoMoreInteractions(this.connection); + } + + @Test + public void sendWithExecutionException() throws Exception { + + this.session.afterConnected(this.connection); + assertTrue(this.session.isConnected()); + + IllegalStateException exception = new IllegalStateException("simulated exception"); + SettableListenableFuture future = new SettableListenableFuture<>(); + future.setException(exception); + + when(this.connection.send(any())).thenReturn(future); + this.expected.expect(MessageDeliveryException.class); + this.expected.expectCause(Matchers.sameInstance(exception)); + + this.session.send("/topic/foo", "sample payload".getBytes(UTF_8)); + + verifyNoMoreInteractions(this.connection); + } + + @Test + public void subscribe() throws Exception { + + this.session.afterConnected(this.connection); + assertTrue(this.session.isConnected()); + + String destination = "/topic/foo"; + StompFrameHandler frameHandler = mock(StompFrameHandler.class); + Subscription subscription = this.session.subscribe(destination, frameHandler); + + Message message = this.messageCaptor.getValue(); + StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class); + assertEquals(StompCommand.SUBSCRIBE, accessor.getCommand()); + + StompHeaders stompHeaders = StompHeaders.readOnlyStompHeaders(accessor.getNativeHeaders()); + assertEquals(stompHeaders.toString(), 2, stompHeaders.size()); + assertEquals(destination, stompHeaders.getDestination()); + assertEquals(subscription.getSubscriptionId(), stompHeaders.getId()); + } + + @Test + public void subscribeWithHeaders() throws Exception { + + this.session.afterConnected(this.connection); + assertTrue(this.session.isConnected()); + + String subscriptionId = "123"; + String destination = "/topic/foo"; + + StompHeaders stompHeaders = new StompHeaders(); + stompHeaders.setId(subscriptionId); + stompHeaders.setDestination(destination); + StompFrameHandler frameHandler = mock(StompFrameHandler.class); + + Subscription subscription = this.session.subscribe(stompHeaders, frameHandler); + assertEquals(subscriptionId, subscription.getSubscriptionId()); + + Message message = this.messageCaptor.getValue(); + StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class); + assertEquals(StompCommand.SUBSCRIBE, accessor.getCommand()); + + stompHeaders = StompHeaders.readOnlyStompHeaders(accessor.getNativeHeaders()); + assertEquals(stompHeaders.toString(), 2, stompHeaders.size()); + assertEquals(destination, stompHeaders.getDestination()); + assertEquals(subscriptionId, stompHeaders.getId()); + } + + @Test + public void unsubscribe() throws Exception { + + this.session.afterConnected(this.connection); + assertTrue(this.session.isConnected()); + + String destination = "/topic/foo"; + StompFrameHandler frameHandler = mock(StompFrameHandler.class); + Subscription subscription = this.session.subscribe(destination, frameHandler); + subscription.unsubscribe(); + + Message message = this.messageCaptor.getValue(); + StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class); + assertEquals(StompCommand.UNSUBSCRIBE, accessor.getCommand()); + + StompHeaders stompHeaders = StompHeaders.readOnlyStompHeaders(accessor.getNativeHeaders()); + assertEquals(stompHeaders.toString(), 1, stompHeaders.size()); + assertEquals(subscription.getSubscriptionId(), stompHeaders.getId()); + } + + @Test + public void receiptReceived() throws Exception { + + this.session.afterConnected(this.connection); + this.session.setTaskScheduler(mock(TaskScheduler.class)); + + AtomicReference received = new AtomicReference<>(); + + StompHeaders headers = new StompHeaders(); + headers.setDestination("/topic/foo"); + headers.setReceipt("my-receipt"); + Subscription subscription = this.session.subscribe(headers, mock(StompFrameHandler.class)); + subscription.addReceiptTask(() -> received.set(true)); + + assertNull(received.get()); + + StompHeaderAccessor accessor = StompHeaderAccessor.create(StompCommand.RECEIPT); + accessor.setReceiptId("my-receipt"); + accessor.setLeaveMutable(true); + this.session.handleMessage(MessageBuilder.createMessage(new byte[0], accessor.getMessageHeaders())); + + assertNotNull(received.get()); + assertTrue(received.get()); + } + + @Test + public void receiptReceivedBeforeTaskAdded() throws Exception { + + this.session.afterConnected(this.connection); + this.session.setTaskScheduler(mock(TaskScheduler.class)); + + AtomicReference received = new AtomicReference<>(); + + StompHeaders headers = new StompHeaders(); + headers.setDestination("/topic/foo"); + headers.setReceipt("my-receipt"); + Subscription subscription = this.session.subscribe(headers, mock(StompFrameHandler.class)); + + StompHeaderAccessor accessor = StompHeaderAccessor.create(StompCommand.RECEIPT); + accessor.setReceiptId("my-receipt"); + accessor.setLeaveMutable(true); + this.session.handleMessage(MessageBuilder.createMessage(new byte[0], accessor.getMessageHeaders())); + + subscription.addReceiptTask(() -> received.set(true)); + + assertNotNull(received.get()); + assertTrue(received.get()); + } + + @SuppressWarnings("unchecked") + @Test + public void receiptNotReceived() throws Exception { + + TaskScheduler taskScheduler = mock(TaskScheduler.class); + + this.session.afterConnected(this.connection); + this.session.setTaskScheduler(taskScheduler); + + AtomicReference notReceived = new AtomicReference<>(); + + ScheduledFuture future = mock(ScheduledFuture.class); + when(taskScheduler.schedule(any(Runnable.class), any(Date.class))).thenReturn(future); + + StompHeaders headers = new StompHeaders(); + headers.setDestination("/topic/foo"); + headers.setReceipt("my-receipt"); + Receiptable receiptable = this.session.send(headers, "payload"); + receiptable.addReceiptLostTask(() -> notReceived.set(true)); + + ArgumentCaptor taskCaptor = ArgumentCaptor.forClass(Runnable.class); + verify(taskScheduler).schedule(taskCaptor.capture(), notNull(Date.class)); + Runnable scheduledTask = taskCaptor.getValue(); + assertNotNull(scheduledTask); + + assertNull(notReceived.get()); + + scheduledTask.run(); + assertTrue(notReceived.get()); + verify(future).cancel(true); + verifyNoMoreInteractions(future); + } + + @Test + public void disconnect() throws Exception { + + this.session.afterConnected(this.connection); + assertTrue(this.session.isConnected()); + + this.session.disconnect(); + assertFalse(this.session.isConnected()); + verifyNoMoreInteractions(this.sessionHandler); + } + +} diff --git a/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/Reactor11TcpStompClientTests.java b/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/Reactor11TcpStompClientTests.java new file mode 100644 index 0000000000..97f4532875 --- /dev/null +++ b/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/Reactor11TcpStompClientTests.java @@ -0,0 +1,236 @@ +/* + * Copyright 2002-2015 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.stomp; + +import static org.hamcrest.Matchers.*; +import static org.junit.Assert.*; + +import java.lang.reflect.Type; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.apache.activemq.broker.BrokerService; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TestName; + +import org.springframework.messaging.converter.StringMessageConverter; +import org.springframework.messaging.simp.stomp.StompSession.Subscription; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import org.springframework.util.Assert; +import org.springframework.util.SocketUtils; +import org.springframework.util.concurrent.ListenableFuture; + +/** + * Integration tests for {@link Reactor11TcpStompClient}. + * + * @author Rossen Stoyanchev + */ +public class Reactor11TcpStompClientTests { + + private static final Log logger = LogFactory.getLog(Reactor11TcpStompClientTests.class); + + @Rule + public final TestName testName = new TestName(); + + private BrokerService activeMQBroker; + + private Reactor11TcpStompClient client; + + + @Before + public void setUp() throws Exception { + + logger.debug("Setting up before '" + this.testName.getMethodName() + "'"); + + int port = SocketUtils.findAvailableTcpPort(61613); + + this.activeMQBroker = new BrokerService(); + this.activeMQBroker.addConnector("stomp://127.0.0.1:" + port); + this.activeMQBroker.setStartAsync(false); + this.activeMQBroker.setPersistent(false); + this.activeMQBroker.setUseJmx(false); + this.activeMQBroker.getSystemUsage().getMemoryUsage().setLimit(1024 * 1024 * 5); + this.activeMQBroker.getSystemUsage().getTempUsage().setLimit(1024 * 1024 * 5); + this.activeMQBroker.start(); + + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.afterPropertiesSet(); + + this.client = new Reactor11TcpStompClient("127.0.0.1", port); + this.client.setMessageConverter(new StringMessageConverter()); + this.client.setTaskScheduler(taskScheduler); + } + + @After + public void tearDown() throws Exception { + try { + this.client.shutdown(); + } + catch (Throwable ex) { + logger.error("Failed to shut client", ex); + } + final CountDownLatch latch = new CountDownLatch(1); + this.activeMQBroker.addShutdownHook(latch::countDown); + logger.debug("Stopping ActiveMQ broker and will await shutdown"); + this.activeMQBroker.stop(); + if (!latch.await(5, TimeUnit.SECONDS)) { + logger.debug("ActiveMQ broker did not shut in the expected time."); + } + } + + @Test + public void publishSubscribe() throws Exception { + + String destination = "/topic/foo"; + ConsumingHandler consumingHandler1 = new ConsumingHandler(destination); + ListenableFuture consumerFuture1 = this.client.connect(consumingHandler1); + + ConsumingHandler consumingHandler2 = new ConsumingHandler(destination); + ListenableFuture consumerFuture2 = this.client.connect(consumingHandler2); + + assertTrue(consumingHandler1.awaitForSubscriptions(5000)); + assertTrue(consumingHandler2.awaitForSubscriptions(5000)); + + ProducingHandler producingHandler = new ProducingHandler(); + producingHandler.addToSend(destination, "foo1"); + producingHandler.addToSend(destination, "foo2"); + ListenableFuture producerFuture = this.client.connect(producingHandler); + + assertTrue(consumingHandler1.awaitForMessageCount(2, 5000)); + assertThat(consumingHandler1.getReceived(), containsInAnyOrder("foo1", "foo2")); + + assertTrue(consumingHandler2.awaitForMessageCount(2, 5000)); + assertThat(consumingHandler2.getReceived(), containsInAnyOrder("foo1", "foo2")); + + consumerFuture1.get().disconnect(); + consumerFuture2.get().disconnect(); + producerFuture.get().disconnect(); + } + + + private static class LoggingSessionHandler extends StompSessionHandlerAdapter { + + @Override + public void handleException(StompSession session, StompCommand command, + StompHeaders headers, byte[] payload, Throwable ex) { + + logger.error(command + " " + headers, ex); + } + + @Override + public void handleFrame(StompHeaders headers, Object payload) { + logger.error("STOMP error frame " + headers + " payload=" + payload); + } + + @Override + public void handleTransportError(StompSession session, Throwable exception) { + logger.error(exception); + } + + } + + private static class ConsumingHandler extends LoggingSessionHandler { + + private final List topics; + + private final CountDownLatch subscriptionLatch; + + private final List received = new ArrayList<>(); + + + public ConsumingHandler(String... topics) { + Assert.notEmpty(topics); + this.topics = Arrays.asList(topics); + this.subscriptionLatch = new CountDownLatch(this.topics.size()); + } + + + public List getReceived() { + return this.received; + } + + + @Override + public void afterConnected(StompSession session, StompHeaders connectedHeaders) { + for (String topic : this.topics) { + session.setAutoReceipt(true); + Subscription subscription = session.subscribe(topic, new StompFrameHandler() { + @Override + public Type getPayloadType(StompHeaders headers) { + return String.class; + } + @Override + public void handleFrame(StompHeaders headers, Object payload) { + received.add((String) payload); + } + }); + subscription.addReceiptTask(subscriptionLatch::countDown); + } + } + + public boolean awaitForSubscriptions(long millisToWait) throws InterruptedException { + if (logger.isDebugEnabled()) { + logger.debug("Awaiting for subscription receipts"); + } + return this.subscriptionLatch.await(millisToWait, TimeUnit.MILLISECONDS); + } + + public boolean awaitForMessageCount(int expected, long millisToWait) throws InterruptedException { + if (logger.isDebugEnabled()) { + logger.debug("Awaiting for message count: " + expected); + } + long startTime = System.currentTimeMillis(); + while (this.received.size() < expected) { + Thread.sleep(500); + if ((System.currentTimeMillis() - startTime) > millisToWait) { + return false; + } + } + return true; + } + + } + + private static class ProducingHandler extends LoggingSessionHandler { + + private final List topics = new ArrayList<>(); + + private final List payloads = new ArrayList<>(); + + + public ProducingHandler addToSend(String topic, Object payload) { + this.topics.add(topic); + this.payloads.add(payload); + return this; + } + + @Override + public void afterConnected(StompSession session, StompHeaders connectedHeaders) { + for (int i=0; i < this.topics.size(); i++) { + session.send(this.topics.get(i), this.payloads.get(i)); + } + } + } + +} diff --git a/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/StompClientSupportTests.java b/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/StompClientSupportTests.java new file mode 100644 index 0000000000..7059ab44ec --- /dev/null +++ b/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/StompClientSupportTests.java @@ -0,0 +1,92 @@ +/* + * Copyright 2002-2015 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.stomp; + +import static org.junit.Assert.*; +import static org.mockito.Matchers.same; + +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.ExpectedException; +import org.mockito.MockitoAnnotations; + + +/** + * Unit tests for {@@link StompClientSupport}. + * @author Rossen Stoyanchev + */ +public class StompClientSupportTests { + + private StompClientSupport stompClient; + + + @Before + public void setUp() throws Exception { + this.stompClient = new StompClientSupport() {}; + } + + @Test + public void defaultHearbeatValidation() throws Exception { + trySetDefaultHeartbeat(null); + trySetDefaultHeartbeat(new long[] {-1, 0}); + trySetDefaultHeartbeat(new long[] {0, -1}); + } + + private void trySetDefaultHeartbeat(long[] heartbeat) { + try { + this.stompClient.setDefaultHeartbeat(heartbeat); + fail("Expected exception"); + } + catch (IllegalArgumentException ex) { + // Ignore + } + } + + @Test + public void defaultHeartbeatValue() throws Exception { + assertArrayEquals(new long[] {10000, 10000}, this.stompClient.getDefaultHeartbeat()); + } + + @Test + public void isDefaultHeartbeatEnabled() throws Exception { + assertArrayEquals(new long[] {10000, 10000}, this.stompClient.getDefaultHeartbeat()); + assertTrue(this.stompClient.isDefaultHeartbeatEnabled()); + + this.stompClient.setDefaultHeartbeat(new long[] {0, 0}); + assertFalse(this.stompClient.isDefaultHeartbeatEnabled()); + } + + @Test + public void processConnectHeadersDefault() throws Exception { + StompHeaders connectHeaders = this.stompClient.processConnectHeaders(null); + + assertNotNull(connectHeaders); + assertArrayEquals(new long[] {10000, 10000}, connectHeaders.getHeartbeat()); + } + + @Test + public void processConnectHeadersWithExplicitHeartbeat() throws Exception { + + StompHeaders connectHeaders = new StompHeaders(); + connectHeaders.setHeartbeat(new long[] {15000, 15000}); + connectHeaders = this.stompClient.processConnectHeaders(connectHeaders); + + assertNotNull(connectHeaders); + assertArrayEquals(new long[] {15000, 15000}, connectHeaders.getHeartbeat()); + } + +} diff --git a/spring-messaging/src/test/resources/log4j.properties b/spring-messaging/src/test/resources/log4j.properties index 9108ba5605..06597aa7b5 100644 --- a/spring-messaging/src/test/resources/log4j.properties +++ b/spring-messaging/src/test/resources/log4j.properties @@ -4,13 +4,10 @@ log4j.appender.console.layout.ConversionPattern=%d{HH:mm:ss,SSS} [%c] - %m%n log4j.rootCategory=WARN, console log4j.logger.org.springframework.messaging=DEBUG -log4j.logger.org.springframework.web=DEBUG log4j.logger.org.apache.activemq=TRACE + # Enable TRACE level to chase integration test issues on CI servers log4j.logger.org.springframework.messaging.simp.stomp=TRACE log4j.logger.reactor.net=TRACE log4j.logger.io.netty=TRACE - -#log4j.logger.org.springframework.messaging=TRACE - diff --git a/spring-websocket/src/main/java/org/springframework/web/socket/messaging/WebSocketStompClient.java b/spring-websocket/src/main/java/org/springframework/web/socket/messaging/WebSocketStompClient.java new file mode 100644 index 0000000000..48475cc35d --- /dev/null +++ b/spring-websocket/src/main/java/org/springframework/web/socket/messaging/WebSocketStompClient.java @@ -0,0 +1,512 @@ +/* + * Copyright 2002-2015 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.web.socket.messaging; + +import java.io.IOException; +import java.net.URI; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.ScheduledFuture; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.context.Lifecycle; +import org.springframework.context.SmartLifecycle; +import org.springframework.messaging.Message; +import org.springframework.messaging.simp.stomp.BufferingStompDecoder; +import org.springframework.messaging.simp.stomp.ConnectionHandlingStompSession; +import org.springframework.messaging.simp.stomp.StompClientSupport; +import org.springframework.messaging.simp.stomp.StompDecoder; +import org.springframework.messaging.simp.stomp.StompEncoder; +import org.springframework.messaging.simp.stomp.StompHeaderAccessor; +import org.springframework.messaging.simp.stomp.StompHeaders; +import org.springframework.messaging.simp.stomp.StompSession; +import org.springframework.messaging.simp.stomp.StompSessionHandler; +import org.springframework.messaging.support.MessageHeaderAccessor; +import org.springframework.messaging.tcp.TcpConnection; +import org.springframework.messaging.tcp.TcpConnectionHandler; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.util.Assert; +import org.springframework.util.MimeTypeUtils; +import org.springframework.util.concurrent.ListenableFuture; +import org.springframework.util.concurrent.ListenableFutureCallback; +import org.springframework.util.concurrent.SettableListenableFuture; +import org.springframework.web.socket.BinaryMessage; +import org.springframework.web.socket.CloseStatus; +import org.springframework.web.socket.TextMessage; +import org.springframework.web.socket.WebSocketHandler; +import org.springframework.web.socket.WebSocketHttpHeaders; +import org.springframework.web.socket.WebSocketMessage; +import org.springframework.web.socket.WebSocketSession; +import org.springframework.web.socket.client.WebSocketClient; +import org.springframework.web.socket.sockjs.transport.SockJsSession; +import org.springframework.web.util.UriComponentsBuilder; + +/** + * A STOMP over WebSocket client that connects using an implementation of + * {@link org.springframework.web.socket.client.WebSocketClient WebSocketClient} + * including {@link org.springframework.web.socket.sockjs.client.SockJsClient + * SockJsClient}. + * + * @author Rossen Stoyanchev + * @since 4.2 + */ +public class WebSocketStompClient extends StompClientSupport implements SmartLifecycle { + + private static Log logger = LogFactory.getLog(WebSocketStompClient.class); + + + private final WebSocketClient webSocketClient; + + private int inboundMessageSizeLimit = 64 * 1024; + + private boolean autoStartup = true; + + private boolean running = false; + + private int phase = Integer.MAX_VALUE; + + + /** + * Class constructor. Sets {@link #setDefaultHeartbeat} to "0,0" but will + * reset it back to the preferred "10000,10000" when a + * {@link #setTaskScheduler} is configured. + * + * @param webSocketClient the WebSocket client to connect with + */ + public WebSocketStompClient(WebSocketClient webSocketClient) { + Assert.notNull(webSocketClient, "'webSocketClient' is required."); + this.webSocketClient = webSocketClient; + setDefaultHeartbeat(new long[] {0, 0}); + } + + + /** + * Return the configured WebSocketClient. + */ + public WebSocketClient getWebSocketClient() { + return this.webSocketClient; + } + + /** + * {@inheritDoc} + *

Also automatically sets the {@link #setDefaultHeartbeat defaultHeartbeat} + * property to "10000,10000" if it is currently set to "0,0". + */ + @Override + public void setTaskScheduler(TaskScheduler taskScheduler) { + if (taskScheduler != null && !isDefaultHeartbeatEnabled()) { + setDefaultHeartbeat(new long[] {10000, 10000}); + } + super.setTaskScheduler(taskScheduler); + } + + /** + * Configure the maximum size allowed for inbound STOMP message. + * Since a STOMP message can be received in multiple WebSocket messages, + * buffering may be required and this property determines the maximum buffer + * size per message. + *

By default this is set to 64 * 1024 (64K). + */ + public void setInboundMessageSizeLimit(int inboundMessageSizeLimit) { + this.inboundMessageSizeLimit = inboundMessageSizeLimit; + } + + /** + * Get the configured inbound message buffer size in bytes. + */ + public int getInboundMessageSizeLimit() { + return this.inboundMessageSizeLimit; + } + + /** + * Set whether to auto-start the contained WebSocketClient when the Spring + * context has been refreshed. + *

Default is "true". + */ + public void setAutoStartup(boolean autoStartup) { + this.autoStartup = autoStartup; + } + + /** + * Return the value for the 'autoStartup' property. If "true", this client + * will automatically start and stop the contained WebSocketClient. + */ + @Override + public boolean isAutoStartup() { + return this.autoStartup; + } + + @Override + public boolean isRunning() { + return this.running; + } + + /** + * Specify the phase in which the WebSocket client should be started and + * subsequently closed. The startup order proceeds from lowest to highest, + * and the shutdown order is the reverse of that. + *

By default this is Integer.MAX_VALUE meaning that the WebSocket client + * is started as late as possible and stopped as soon as possible. + */ + public void setPhase(int phase) { + this.phase = phase; + } + + /** + * Return the configured phase. + */ + @Override + public int getPhase() { + return this.phase; + } + + + @Override + public void start() { + if (!isRunning()) { + this.running = true; + if (getWebSocketClient() instanceof Lifecycle) { + ((Lifecycle) getWebSocketClient()).start(); + } + } + + } + + @Override + public void stop() { + if (isRunning()) { + this.running = false; + if (getWebSocketClient() instanceof Lifecycle) { + ((Lifecycle) getWebSocketClient()).stop(); + } + } + } + + @Override + public void stop(Runnable callback) { + this.stop(); + callback.run(); + } + + /** + * Connect to the given WebSocket URL and notify the given + * {@link org.springframework.messaging.simp.stomp.StompSessionHandler} + * when connected on the STOMP level after the CONNECTED frame is received. + * @param url the url to connect to + * @param handler the session handler + * @param uriVars URI variables to expand into the URL + * @return ListenableFuture for access to the session when ready for use + */ + public ListenableFuture connect(String url, StompSessionHandler handler, Object... uriVars) { + return connect(url, null, handler, uriVars); + } + + /** + * An overloaded version of + * {@link #connect(String, StompSessionHandler, Object...)} that also + * accepts {@link WebSocketHttpHeaders} to use for the WebSocket handshake. + * @param url the url to connect to + * @param handshakeHeaders the headers for the WebSocket handshake + * @param handler the session handler + * @param uriVariables URI variables to expand into the URL + * @return ListenableFuture for access to the session when ready for use + */ + public ListenableFuture connect(String url, WebSocketHttpHeaders handshakeHeaders, + StompSessionHandler handler, Object... uriVariables) { + + return connect(url, handshakeHeaders, null, handler, uriVariables); + } + + /** + * An overloaded version of + * {@link #connect(String, StompSessionHandler, Object...)} that also accepts + * {@link WebSocketHttpHeaders} to use for the WebSocket handshake and + * {@link StompHeaders} for the STOMP CONNECT frame. + * @param url the url to connect to + * @param handshakeHeaders headers for the WebSocket handshake + * @param connectHeaders headers for the STOMP CONNECT frame + * @param handler the session handler + * @param uriVariables URI variables to expand into the URL + * @return ListenableFuture for access to the session when ready for use + */ + public ListenableFuture connect(String url, WebSocketHttpHeaders handshakeHeaders, + StompHeaders connectHeaders, StompSessionHandler handler, Object... uriVariables) { + + Assert.notNull(url, "uriTemplate must not be null"); + URI uri = UriComponentsBuilder.fromUriString(url).buildAndExpand(uriVariables).encode().toUri(); + return connect(uri, handshakeHeaders, connectHeaders, handler); + } + + /** + * An overloaded version of + * {@link #connect(String, WebSocketHttpHeaders, StompSessionHandler, Object...)} + * that accepts a fully prepared {@link java.net.URI}. + * @param url the url to connect to + * @param handshakeHeaders the headers for the WebSocket handshake + * @param connectHeaders headers for the STOMP CONNECT frame + * @param sessionHandler the STOMP session handler + * @return ListenableFuture for access to the session when ready for use + */ + public ListenableFuture connect(URI url, WebSocketHttpHeaders handshakeHeaders, + StompHeaders connectHeaders, StompSessionHandler sessionHandler) { + + Assert.notNull(url, "'uri' must not be null"); + ConnectionHandlingStompSession session = createSession(connectHeaders, sessionHandler); + WebSocketTcpConnectionHandlerAdapter adapter = new WebSocketTcpConnectionHandlerAdapter(session); + getWebSocketClient().doHandshake(adapter, handshakeHeaders, url).addCallback(adapter); + return session.getSessionFuture(); + } + + @Override + protected StompHeaders processConnectHeaders(StompHeaders connectHeaders) { + connectHeaders = super.processConnectHeaders(connectHeaders); + if (connectHeaders.isHeartbeatEnabled()) { + Assert.notNull(getTaskScheduler(), "TaskScheduler cannot be null if heartbeats are enabled."); + } + return connectHeaders; + } + + + /** + * Adapt WebSocket to the TcpConnectionHandler and TcpConnection contracts. + */ + private class WebSocketTcpConnectionHandlerAdapter implements ListenableFutureCallback, + WebSocketHandler, TcpConnection { + + private final TcpConnectionHandler connectionHandler; + + private final StompWebSocketMessageCodec codec = new StompWebSocketMessageCodec(getInboundMessageSizeLimit()); + + private volatile WebSocketSession session; + + private volatile long lastReadTime = -1; + + private volatile long lastWriteTime = -1; + + private final List> inactivityTasks = new ArrayList>(2); + + + public WebSocketTcpConnectionHandlerAdapter(TcpConnectionHandler connectionHandler) { + Assert.notNull(connectionHandler); + this.connectionHandler = connectionHandler; + } + + + // ListenableFutureCallback implementation: handshake outcome + + @Override + public void onSuccess(WebSocketSession webSocketSession) { + } + + @Override + public void onFailure(Throwable ex) { + this.connectionHandler.afterConnectFailure(ex); + } + + + // WebSocketHandler implementation + + @Override + public void afterConnectionEstablished(WebSocketSession session) { + this.session = session; + this.connectionHandler.afterConnected(this); + } + + @Override + public void handleMessage(WebSocketSession session, WebSocketMessage webSocketMessage) { + this.lastReadTime = (this.lastReadTime != -1 ? System.currentTimeMillis() : -1); + List> messages; + try { + messages = this.codec.decode(webSocketMessage); + } + catch (Throwable ex) { + this.connectionHandler.handleFailure(ex); + return; + } + for (Message message : messages) { + this.connectionHandler.handleMessage(message); + } + } + + @Override + public void handleTransportError(WebSocketSession session, Throwable ex) throws Exception { + this.connectionHandler.handleFailure(ex); + } + + @Override + public void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) throws Exception { + cancelInactivityTasks(); + this.connectionHandler.afterConnectionClosed(); + } + + private void cancelInactivityTasks() { + for (ScheduledFuture task : this.inactivityTasks) { + try { + task.cancel(true); + } + catch (Throwable ex) { + // Ignore + } + } + this.lastReadTime = -1; + this.lastWriteTime = -1; + this.inactivityTasks.clear(); + } + + @Override + public boolean supportsPartialMessages() { + return false; + } + + + // TcpConnection implementation + + @Override + public ListenableFuture send(Message message) { + updateLastWriteTime(); + SettableListenableFuture future = new SettableListenableFuture(); + try { + this.session.sendMessage(this.codec.encode(message, this.session.getClass())); + future.set(null); + } + catch (Throwable ex) { + future.setException(ex); + } + finally { + updateLastWriteTime(); + } + return future; + } + + private void updateLastWriteTime() { + this.lastWriteTime = (this.lastWriteTime != -1 ? System.currentTimeMillis() : -1); + } + + @Override + public void onReadInactivity(final Runnable runnable, final long duration) { + Assert.notNull(getTaskScheduler(), "No scheduler configured."); + this.lastReadTime = System.currentTimeMillis(); + this.inactivityTasks.add(getTaskScheduler().scheduleWithFixedDelay(new Runnable() { + @Override + public void run() { + if (System.currentTimeMillis() - lastReadTime > duration) { + try { + runnable.run(); + } + catch (Throwable ex) { + if (logger.isDebugEnabled()) { + logger.debug("ReadInactivityTask failure", ex); + } + } + } + } + }, duration / 2)); + } + + @Override + public void onWriteInactivity(final Runnable runnable, final long duration) { + Assert.notNull(getTaskScheduler(), "No scheduler configured."); + this.lastWriteTime = System.currentTimeMillis(); + this.inactivityTasks.add(getTaskScheduler().scheduleWithFixedDelay(new Runnable() { + @Override + public void run() { + if (System.currentTimeMillis() - lastWriteTime > duration) { + try { + runnable.run(); + } + catch (Throwable ex) { + if (logger.isDebugEnabled()) { + logger.debug("WriteInactivityTask failure", ex); + } + } + } + } + }, duration / 2)); + } + + @Override + public void close() { + try { + this.session.close(); + } + catch (IOException ex) { + if (logger.isDebugEnabled()) { + logger.debug("Failed to close session: " + this.session.getId(), ex); + } + } + } + } + + + /** + * Encode and decode STOMP WebSocket messages. + */ + private static class StompWebSocketMessageCodec { + + private static final StompEncoder ENCODER = new StompEncoder(); + + private static final StompDecoder DECODER = new StompDecoder(); + + private final BufferingStompDecoder bufferingDecoder; + + + public StompWebSocketMessageCodec(int messageSizeLimit) { + this.bufferingDecoder = new BufferingStompDecoder(DECODER, messageSizeLimit); + } + + + public List> decode(WebSocketMessage webSocketMessage) { + List> result = Collections.>emptyList(); + ByteBuffer byteBuffer; + if (webSocketMessage instanceof TextMessage) { + byteBuffer = ByteBuffer.wrap(((TextMessage) webSocketMessage).asBytes()); + } + else if (webSocketMessage instanceof BinaryMessage) { + byteBuffer = ((BinaryMessage) webSocketMessage).getPayload(); + } + else { + return result; + } + result = this.bufferingDecoder.decode(byteBuffer); + if (result.isEmpty()) { + if (logger.isTraceEnabled()) { + logger.trace("Incomplete STOMP frame content received, bufferSize=" + + this.bufferingDecoder.getBufferSize() + ", bufferSizeLimit=" + + this.bufferingDecoder.getBufferSizeLimit() + "."); + } + } + return result; + } + + public WebSocketMessage encode(Message message, Class sessionType) { + StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class); + Assert.notNull(accessor); + byte[] payload = message.getPayload(); + byte[] bytes = ENCODER.encode(accessor.getMessageHeaders(), payload); + + boolean useBinary = (payload.length > 0 && + !(SockJsSession.class.isAssignableFrom(sessionType)) && + MimeTypeUtils.APPLICATION_OCTET_STREAM.isCompatibleWith(accessor.getContentType())); + + return (useBinary ? new BinaryMessage(bytes) : new TextMessage(bytes)); + } + } + +} diff --git a/spring-websocket/src/test/java/org/springframework/web/socket/messaging/WebSocketStompClientIntegrationTests.java b/spring-websocket/src/test/java/org/springframework/web/socket/messaging/WebSocketStompClientIntegrationTests.java new file mode 100644 index 0000000000..d589d43e8c --- /dev/null +++ b/spring-websocket/src/test/java/org/springframework/web/socket/messaging/WebSocketStompClientIntegrationTests.java @@ -0,0 +1,213 @@ +/* + * Copyright 2002-2015 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.web.socket.messaging; + +import static org.hamcrest.Matchers.*; +import static org.junit.Assert.*; + +import java.lang.reflect.Type; +import java.util.ArrayList; +import java.util.List; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TestName; + +import org.springframework.context.annotation.Configuration; +import org.springframework.messaging.converter.StringMessageConverter; +import org.springframework.messaging.simp.config.MessageBrokerRegistry; +import org.springframework.messaging.simp.stomp.StompCommand; +import org.springframework.messaging.simp.stomp.StompFrameHandler; +import org.springframework.messaging.simp.stomp.StompHeaders; +import org.springframework.messaging.simp.stomp.StompSession; +import org.springframework.messaging.simp.stomp.StompSessionHandlerAdapter; +import org.springframework.web.context.support.AnnotationConfigWebApplicationContext; +import org.springframework.web.socket.TomcatWebSocketTestServer; +import org.springframework.web.socket.WebSocketTestServer; +import org.springframework.web.socket.client.WebSocketClient; +import org.springframework.web.socket.client.standard.StandardWebSocketClient; +import org.springframework.web.socket.config.annotation.StompEndpointRegistry; +import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurationSupport; +import org.springframework.web.socket.server.RequestUpgradeStrategy; +import org.springframework.web.socket.server.standard.TomcatRequestUpgradeStrategy; +import org.springframework.web.socket.server.support.DefaultHandshakeHandler; + + +/** + * Integration tests for {@link WebSocketStompClient}. + * @author Rossen Stoyanchev + */ +public class WebSocketStompClientIntegrationTests { + + private static final Log logger = LogFactory.getLog(WebSocketStompClientIntegrationTests.class); + + @Rule + public final TestName testName = new TestName(); + + private WebSocketStompClient stompClient; + + private WebSocketTestServer server; + + private AnnotationConfigWebApplicationContext wac; + + + @Before + public void setUp() throws Exception { + + logger.debug("Setting up before '" + this.testName.getMethodName() + "'"); + + this.wac = new AnnotationConfigWebApplicationContext(); + this.wac.register(TestConfig.class); + this.wac.refresh(); + + this.server = new TomcatWebSocketTestServer(); + this.server.setup(); + this.server.deployConfig(this.wac); + this.server.start(); + + WebSocketClient webSocketClient = new StandardWebSocketClient(); + this.stompClient = new WebSocketStompClient(webSocketClient); + this.stompClient.setMessageConverter(new StringMessageConverter()); + } + + @After + public void tearDown() throws Exception { + try { + this.server.undeployConfig(); + } + catch (Throwable t) { + logger.error("Failed to undeploy application config", t); + } + try { + this.server.stop(); + } + catch (Throwable t) { + logger.error("Failed to stop server", t); + } + try { + this.wac.close(); + } + catch (Throwable t) { + logger.error("Failed to close WebApplicationContext", t); + } + } + + + @Test + public void publishSubscribe() throws Exception { + + String url = "ws://127.0.0.1:" + this.server.getPort() + "/stomp"; + + TestHandler testHandler = new TestHandler("/topic/foo", "payload"); + this.stompClient.connect(url, testHandler); + + assertTrue(testHandler.awaitForMessageCount(1, 5000)); + assertThat(testHandler.getReceived(), containsInAnyOrder("payload")); + } + + + @Configuration + static class TestConfig extends WebSocketMessageBrokerConfigurationSupport { + + @Override + protected void registerStompEndpoints(StompEndpointRegistry registry) { + // Can't rely on classpath detection + RequestUpgradeStrategy upgradeStrategy = new TomcatRequestUpgradeStrategy(); + registry.addEndpoint("/stomp") + .setHandshakeHandler(new DefaultHandshakeHandler(upgradeStrategy)) + .setAllowedOrigins("*"); + } + + @Override + public void configureMessageBroker(MessageBrokerRegistry configurer) { + configurer.setApplicationDestinationPrefixes("/app"); + configurer.enableSimpleBroker("/topic", "/queue"); + } + } + + + private static class TestHandler extends StompSessionHandlerAdapter { + + private final String topic; + + private final Object payload; + + private final List received = new ArrayList<>(); + + + public TestHandler(String topic, Object payload) { + this.topic = topic; + this.payload = payload; + } + + + public List getReceived() { + return this.received; + } + + + @Override + public void afterConnected(StompSession session, StompHeaders connectedHeaders) { + session.subscribe(this.topic, new StompFrameHandler() { + @Override + public Type getPayloadType(StompHeaders headers) { + return String.class; + } + @Override + public void handleFrame(StompHeaders headers, Object payload) { + received.add((String) payload); + } + }); + session.send(this.topic, this.payload); + } + + public boolean awaitForMessageCount(int expected, long millisToWait) throws InterruptedException { + if (logger.isDebugEnabled()) { + logger.debug("Awaiting for message count: " + expected); + } + long startTime = System.currentTimeMillis(); + while (this.received.size() < expected) { + Thread.sleep(500); + if ((System.currentTimeMillis() - startTime) > millisToWait) { + return false; + } + } + return true; + } + + @Override + public void handleException(StompSession session, StompCommand command, + StompHeaders headers, byte[] payload, Throwable ex) { + + logger.error(command + " " + headers, ex); + } + + @Override + public void handleFrame(StompHeaders headers, Object payload) { + logger.error("STOMP error frame " + headers + " payload=" + payload); + } + + @Override + public void handleTransportError(StompSession session, Throwable exception) { + logger.error(exception); + } + } + +} \ No newline at end of file diff --git a/spring-websocket/src/test/java/org/springframework/web/socket/messaging/WebSocketStompClientTests.java b/spring-websocket/src/test/java/org/springframework/web/socket/messaging/WebSocketStompClientTests.java new file mode 100644 index 0000000000..003ca57733 --- /dev/null +++ b/spring-websocket/src/test/java/org/springframework/web/socket/messaging/WebSocketStompClientTests.java @@ -0,0 +1,381 @@ +/* + * Copyright 2002-2015 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.web.socket.messaging; + + +import static org.junit.Assert.*; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.isNotNull; +import static org.mockito.Matchers.same; +import static org.mockito.Mockito.*; + +import java.net.URI; +import java.nio.charset.Charset; +import java.util.concurrent.ScheduledFuture; + +import org.junit.Before; +import org.junit.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; + +import org.springframework.messaging.Message; +import org.springframework.messaging.simp.stomp.ConnectionHandlingStompSession; +import org.springframework.messaging.simp.stomp.StompCommand; +import org.springframework.messaging.simp.stomp.StompHeaderAccessor; +import org.springframework.messaging.simp.stomp.StompHeaders; +import org.springframework.messaging.simp.stomp.StompSessionHandler; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.messaging.support.MessageHeaderAccessor; +import org.springframework.messaging.tcp.TcpConnection; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.util.MimeTypeUtils; +import org.springframework.util.concurrent.SettableListenableFuture; +import org.springframework.web.socket.BinaryMessage; +import org.springframework.web.socket.CloseStatus; +import org.springframework.web.socket.PongMessage; +import org.springframework.web.socket.TextMessage; +import org.springframework.web.socket.WebSocketHandler; +import org.springframework.web.socket.WebSocketSession; +import org.springframework.web.socket.client.WebSocketClient; + +/** + * Unit tests for {@link \WebSocketStompClient}. + * + * @author Rossen Stoyanchev + */ +public class WebSocketStompClientTests { + + private static final Charset UTF_8 = Charset.forName("UTF-8"); + + + private TestWebSocketStompClient stompClient; + + @Mock + private TaskScheduler taskScheduler; + + @Mock + private ConnectionHandlingStompSession stompSession; + + private ArgumentCaptor webSocketHandlerCaptor; + + private SettableListenableFuture handshakeFuture; + + @Mock + private WebSocketSession webSocketSession; + + + @Before + public void setUp() throws Exception { + MockitoAnnotations.initMocks(this); + + WebSocketClient webSocketClient = mock(WebSocketClient.class); + this.stompClient = new TestWebSocketStompClient(webSocketClient); + this.stompClient.setTaskScheduler(this.taskScheduler); + this.stompClient.setStompSession(this.stompSession); + + this.webSocketHandlerCaptor = ArgumentCaptor.forClass(WebSocketHandler.class); + this.handshakeFuture = new SettableListenableFuture<>(); + when(webSocketClient.doHandshake(this.webSocketHandlerCaptor.capture(), any(), any(URI.class))) + .thenReturn(this.handshakeFuture); + } + + + @SuppressWarnings("unchecked") + @Test + public void webSocketHandshakeFailure() throws Exception { + + connect(); + + IllegalStateException handshakeFailure = new IllegalStateException("simulated exception"); + this.handshakeFuture.setException(handshakeFailure); + + verify(this.stompSession).afterConnectFailure(same(handshakeFailure)); + } + + @SuppressWarnings("unchecked") + @Test + public void webSocketConnectionEstablished() throws Exception { + connect().afterConnectionEstablished(this.webSocketSession); + verify(this.stompSession).afterConnected(isNotNull(TcpConnection.class)); + } + + @Test + public void webSocketTransportError() throws Exception { + + IllegalStateException exception = new IllegalStateException("simulated exception"); + connect().handleTransportError(this.webSocketSession, exception); + + verify(this.stompSession).handleFailure(same(exception)); + } + + @Test + public void webSocketConnectionClosed() throws Exception { + connect().afterConnectionClosed(this.webSocketSession, CloseStatus.NORMAL); + verify(this.stompSession).afterConnectionClosed(); + } + + @Test + public void handleWebSocketMessage() throws Exception { + + String text = "SEND\na:alpha\n\nMessage payload\0"; + connect().handleMessage(this.webSocketSession, new TextMessage(text)); + + ArgumentCaptor> captor = ArgumentCaptor.forClass(Message.class); + verify(this.stompSession).handleMessage(captor.capture()); + Message message = captor.getValue(); + assertNotNull(message); + + StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class); + StompHeaders headers = StompHeaders.readOnlyStompHeaders(accessor.toNativeHeaderMap()); + assertEquals(StompCommand.SEND, accessor.getCommand()); + assertEquals("alpha", headers.getFirst("a")); + assertEquals("Message payload", new String(message.getPayload(), UTF_8)); + } + + @Test + public void handleWebSocketMessageSplitAcrossTwoMessage() throws Exception { + + WebSocketHandler webSocketHandler = connect(); + + String part1 = "SEND\na:alpha\n\nMessage"; + webSocketHandler.handleMessage(this.webSocketSession, new TextMessage(part1)); + + verifyNoMoreInteractions(this.stompSession); + + String part2 = " payload\0"; + webSocketHandler.handleMessage(this.webSocketSession, new TextMessage(part2)); + + ArgumentCaptor> captor = ArgumentCaptor.forClass(Message.class); + verify(this.stompSession).handleMessage(captor.capture()); + Message message = captor.getValue(); + assertNotNull(message); + + StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class); + StompHeaders headers = StompHeaders.readOnlyStompHeaders(accessor.toNativeHeaderMap()); + assertEquals(StompCommand.SEND, accessor.getCommand()); + assertEquals("alpha", headers.getFirst("a")); + assertEquals("Message payload", new String(message.getPayload(), UTF_8)); + } + + @Test + public void handleWebSocketMessageBinary() throws Exception { + + String text = "SEND\na:alpha\n\nMessage payload\0"; + connect().handleMessage(this.webSocketSession, new BinaryMessage(text.getBytes(UTF_8))); + + ArgumentCaptor> captor = ArgumentCaptor.forClass(Message.class); + verify(this.stompSession).handleMessage(captor.capture()); + Message message = captor.getValue(); + assertNotNull(message); + + StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class); + StompHeaders headers = StompHeaders.readOnlyStompHeaders(accessor.toNativeHeaderMap()); + assertEquals(StompCommand.SEND, accessor.getCommand()); + assertEquals("alpha", headers.getFirst("a")); + assertEquals("Message payload", new String(message.getPayload(), UTF_8)); + } + + @Test + public void handleWebSocketMessagePong() throws Exception { + connect().handleMessage(this.webSocketSession, new PongMessage()); + verifyNoMoreInteractions(this.stompSession); + } + + @Test + public void sendWebSocketMessage() throws Exception { + + StompHeaderAccessor accessor = StompHeaderAccessor.create(StompCommand.SEND); + accessor.setDestination("/topic/foo"); + byte[] payload = "payload".getBytes(UTF_8); + + getTcpConnection().send(MessageBuilder.createMessage(payload, accessor.getMessageHeaders())); + + ArgumentCaptor textMessageCaptor = ArgumentCaptor.forClass(TextMessage.class); + verify(this.webSocketSession).sendMessage(textMessageCaptor.capture()); + TextMessage textMessage = textMessageCaptor.getValue(); + assertNotNull(textMessage); + assertEquals("SEND\ndestination:/topic/foo\ncontent-length:7\n\npayload\0", textMessage.getPayload()); + } + + @Test + public void sendWebSocketBinary() throws Exception { + + StompHeaderAccessor accessor = StompHeaderAccessor.create(StompCommand.SEND); + accessor.setDestination("/b"); + accessor.setContentType(MimeTypeUtils.APPLICATION_OCTET_STREAM); + byte[] payload = "payload".getBytes(UTF_8); + + getTcpConnection().send(MessageBuilder.createMessage(payload, accessor.getMessageHeaders())); + + ArgumentCaptor binaryMessageCaptor = ArgumentCaptor.forClass(BinaryMessage.class); + verify(this.webSocketSession).sendMessage(binaryMessageCaptor.capture()); + BinaryMessage binaryMessage = binaryMessageCaptor.getValue(); + assertNotNull(binaryMessage); + assertEquals("SEND\ndestination:/b\ncontent-type:application/octet-stream\ncontent-length:7\n\npayload\0", + new String(binaryMessage.getPayload().array(), UTF_8)); + } + + @Test + public void heartbeatDefaultValue() throws Exception { + WebSocketStompClient stompClient = new WebSocketStompClient(mock(WebSocketClient.class)); + assertArrayEquals(new long[] {0, 0}, stompClient.getDefaultHeartbeat()); + + StompHeaders connectHeaders = stompClient.processConnectHeaders(null); + assertArrayEquals(new long[] {0, 0}, connectHeaders.getHeartbeat()); + } + + @Test + public void heartbeatDefaultValueWithScheduler() throws Exception { + WebSocketStompClient stompClient = new WebSocketStompClient(mock(WebSocketClient.class)); + stompClient.setTaskScheduler(mock(TaskScheduler.class)); + assertArrayEquals(new long[] {10000, 10000}, stompClient.getDefaultHeartbeat()); + + + StompHeaders connectHeaders = stompClient.processConnectHeaders(null); + assertArrayEquals(new long[] {10000, 10000}, connectHeaders.getHeartbeat()); + } + + @Test + public void heartbeatDefaultValueSetWithoutScheduler() throws Exception { + WebSocketStompClient stompClient = new WebSocketStompClient(mock(WebSocketClient.class)); + stompClient.setDefaultHeartbeat(new long[] {5, 5}); + try { + stompClient.processConnectHeaders(null); + fail("Expected exception"); + } + catch (IllegalArgumentException ex) { + // Ignore + } + } + + @Test + public void readInactivityAfterDelayHasElapsed() throws Exception { + TcpConnection tcpConnection = getTcpConnection(); + Runnable runnable = mock(Runnable.class); + long delay = 2; + tcpConnection.onReadInactivity(runnable, delay); + testInactivityTaskScheduling(runnable, delay, 10); + } + + @Test + public void readInactivityBeforeDelayHasElapsed() throws Exception { + TcpConnection tcpConnection = getTcpConnection(); + Runnable runnable = mock(Runnable.class); + long delay = 10000; + tcpConnection.onReadInactivity(runnable, delay); + testInactivityTaskScheduling(runnable, delay, 0); + } + + @Test + public void writeInactivityAfterDelayHasElapsed() throws Exception { + TcpConnection tcpConnection = getTcpConnection(); + Runnable runnable = mock(Runnable.class); + long delay = 2; + tcpConnection.onWriteInactivity(runnable, delay); + testInactivityTaskScheduling(runnable, delay, 10); + } + + @Test + public void writeInactivityBeforeDelayHasElapsed() throws Exception { + TcpConnection tcpConnection = getTcpConnection(); + Runnable runnable = mock(Runnable.class); + long delay = 1000; + tcpConnection.onWriteInactivity(runnable, delay); + testInactivityTaskScheduling(runnable, delay, 0); + } + + @Test + public void cancelInactivityTasks() throws Exception { + + TcpConnection tcpConnection = getTcpConnection(); + + ScheduledFuture future = mock(ScheduledFuture.class); + when(this.taskScheduler.scheduleWithFixedDelay(any(), eq(1L))).thenReturn(future); + + tcpConnection.onReadInactivity(mock(Runnable.class), 2L); + tcpConnection.onWriteInactivity(mock(Runnable.class), 2L); + + this.webSocketHandlerCaptor.getValue().afterConnectionClosed(this.webSocketSession, CloseStatus.NORMAL); + + verify(future, times(2)).cancel(true); + verifyNoMoreInteractions(future); + } + + + private WebSocketHandler connect() { + + this.stompClient.connect("/foo", mock(StompSessionHandler.class)); + + verify(this.stompSession).getSessionFuture(); + verifyNoMoreInteractions(this.stompSession); + + WebSocketHandler webSocketHandler = this.webSocketHandlerCaptor.getValue(); + assertNotNull(webSocketHandler); + return webSocketHandler; + } + + @SuppressWarnings("unchecked") + private TcpConnection getTcpConnection() throws Exception { + WebSocketHandler webSocketHandler = connect(); + webSocketHandler.afterConnectionEstablished(this.webSocketSession); + return (TcpConnection) webSocketHandler; + } + + private void testInactivityTaskScheduling(Runnable runnable, long delay, long sleepTime) + throws InterruptedException { + + ArgumentCaptor inactivityTaskCaptor = ArgumentCaptor.forClass(Runnable.class); + verify(this.taskScheduler).scheduleWithFixedDelay(inactivityTaskCaptor.capture(), eq(delay/2)); + verifyNoMoreInteractions(this.taskScheduler); + + if (sleepTime > 0) { + Thread.sleep(sleepTime); + } + + Runnable inactivityTask = inactivityTaskCaptor.getValue(); + assertNotNull(inactivityTask); + inactivityTask.run(); + + if (sleepTime > 0) { + verify(runnable).run(); + } + else { + verifyNoMoreInteractions(runnable); + } + } + + + private static class TestWebSocketStompClient extends WebSocketStompClient { + + private ConnectionHandlingStompSession stompSession; + + + public TestWebSocketStompClient(WebSocketClient webSocketClient) { + super(webSocketClient); + } + + public void setStompSession(ConnectionHandlingStompSession stompSession) { + this.stompSession = stompSession; + } + + @Override + protected ConnectionHandlingStompSession createSession(StompHeaders headers, StompSessionHandler handler) { + return this.stompSession; + } + } + +} \ No newline at end of file