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

INT-3685: Introduce STOMP Adapters

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3685

* Add `StompInboundChannelAdapter` to subscribe to STOMP destination and receive messages from them.
Destinations can be added/removed (hence subscribed/unsubscribed) at runtime.
* Add `stompMessageHandler` to the send messages to STOMP destinations.
* `destination` can be extracted from `MessageHeaders`, the `destination` and `destinationExpression` are also supported.
* `RECEIPT` Frame is also supported and emitted as `StompReceiptEvent`
* `StompExceptionEvent` is emitted when we have a `failure` for `CONNECT` Frame or as a reaction to the `ERROR` Frame.
* Introduce `StompSessionManager` abstraction to manage the single `StompSession` and allow to share it between different adapters.
* Add `WebSocketStompSessionManager` implementation over `WebSocketStompClient`
* Add `Reactor2TcpStompSessionManager` implementation for target Broker connections
* Add `StompHeaderMapper`

Address PR comments and JavaDocs

Polishing according PR comments

Polishing
This commit is contained in:
Artem Bilan
2015-04-25 18:09:22 +03:00
committed by Gary Russell
parent a0160e50d4
commit 9e4c7e00db
31 changed files with 2210 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
build.gradle
View File

@@ -110,6 +110,7 @@ subprojects { subproject ->
log4jVersion = '1.2.17'
mockitoVersion = '1.9.5'
mysqlVersion = '5.1.34'
nettyVersion = '4.0.27.Final'
openJpaVersion = '2.3.0'
pahoMqttClientVersion = '1.0.2'
postgresVersion = '9.1-901-1.jdbc4'
@@ -171,6 +172,8 @@ subprojects { subproject ->
testCompile project(":spring-integration-test")
}
jacoco group: "org.jacoco", name: "org.jacoco.agent", version: "0.7.2.201409121644", classifier: "runtime"
testRuntime "org.slf4j:slf4j-log4j12:$slf4jVersion"
}
// enable all compiler warnings; individual projects may customize further
@@ -460,7 +463,6 @@ project('spring-integration-mongodb') {
exclude group: 'org.springframework', module: 'spring-expression'
exclude group: 'org.springframework', module: 'spring-tx'
}
testCompile "org.slf4j:slf4j-log4j12:$slf4jVersion"
}
}
@@ -484,7 +486,6 @@ project('spring-integration-redis') {
exclude group: 'org.springframework', module: 'spring-tx'
}
testCompile "redis.clients:jedis:$jedisVersion"
testCompile "org.slf4j:slf4j-log4j12:$slf4jVersion"
}
}
@@ -530,6 +531,24 @@ project('spring-integration-sftp') {
}
}
project('spring-integration-stomp') {
description = 'Spring Integration STOMP Support'
dependencies {
compile project(":spring-integration-core")
compile ("org.springframework:spring-websocket:$springVersion", optional)
testCompile project(":spring-integration-websocket")
testCompile project(":spring-integration-websocket").sourceSets.test.output
testCompile project(":spring-integration-event")
testCompile "org.apache.activemq:activemq-stomp:$activeMqVersion"
testCompile "org.apache.tomcat.embed:tomcat-embed-websocket:$tomcatVersion"
testRuntime "org.apache.tomcat.embed:tomcat-embed-logging-log4j:$tomcatVersion"
testRuntime "io.projectreactor:reactor-net:$reactorVersion"
testRuntime "io.netty:netty-all:$nettyVersion"
}
}
project('spring-integration-stream') {
description = 'Spring Integration Stream Support'
dependencies {
@@ -590,7 +609,8 @@ project('spring-integration-websocket') {
testCompile project(":spring-integration-event")
testCompile "org.apache.tomcat.embed:tomcat-embed-websocket:$tomcatVersion"
testCompile("org.apache.tomcat.embed:tomcat-embed-logging-juli:$tomcatVersion")
testRuntime "org.apache.tomcat.embed:tomcat-embed-logging-log4j:$tomcatVersion"
}
}

36
settings.gradle
View File

@@ -1,33 +1,7 @@
rootProject.name = 'spring-integration'
include 'spring-integration-amqp'
include 'spring-integration-core'
include 'spring-integration-event'
include 'spring-integration-feed'
include 'spring-integration-file'
include 'spring-integration-ftp'
include 'spring-integration-gemfire'
include 'spring-integration-groovy'
include 'spring-integration-http'
include 'spring-integration-ip'
include 'spring-integration-jdbc'
include 'spring-integration-jms'
include 'spring-integration-jmx'
include 'spring-integration-jpa'
include 'spring-integration-mail'
include 'spring-integration-mongodb'
include 'spring-integration-mqtt'
include 'spring-integration-redis'
include 'spring-integration-rmi'
include 'spring-integration-scripting'
include 'spring-integration-security'
include 'spring-integration-sftp'
include 'spring-integration-stream'
include 'spring-integration-syslog'
include 'spring-integration-test'
include 'spring-integration-twitter'
include 'spring-integration-websocket'
include 'spring-integration-ws'
include 'spring-integration-xml'
include 'spring-integration-xmpp'
include 'spring-integration-bom'
rootDir.eachDir { dir ->
if (dir.name.startsWith('spring-integration-')) {
include ":${dir.name}"
}
}

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

@@ -66,6 +66,10 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements
this.shouldTrack = shouldTrack;
}
protected MessagingTemplate getMessagingTemplate() {
return messagingTemplate;
}
@Override
protected void onInit() {
Assert.notNull(this.outputChannel, "outputChannel is required");

48
spring-integration-core/src/main/java/org/springframework/integration/support/converter/PassThruMessageConverter.java Normal file
View File

@@ -0,0 +1,48 @@
/*
* Copyright 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.integration.support.converter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.Assert;
/**
* The simple {@link MessageConverter} implementation which contact is to return
* {@link Message} as is for both {@code from/to} operations.
* <p>
* It is useful in cases of some protocol implementations (e.g. STOMP),
* which is based on the "Spring Messaging Foundation" and the further logic
* operates only with {@link Message}s, e.g. Spring Integration Adapters.
* @author Artem Bilan
* @since 4.2
*/
public class PassThruMessageConverter implements MessageConverter {
@Override
public Object fromMessage(Message<?> message, Class<?> targetClass) {
return message;
}
@Override
public Message<?> toMessage(Object payload, MessageHeaders headers) {
Assert.isInstanceOf(byte[].class, payload, "'payload' must be of 'byte[]' type.");
return MessageBuilder.createMessage(payload, headers);
}
}

231
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java Normal file
View File

@@ -0,0 +1,231 @@
/*
* Copyright 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.integration.stomp;
import java.util.ArrayList;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.integration.stomp.event.StompExceptionEvent;
import org.springframework.messaging.simp.stomp.StompClientSupport;
import org.springframework.messaging.simp.stomp.StompCommand;
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.simp.stomp.StompSessionHandlerAdapter;
import org.springframework.util.Assert;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
/**
* Base {@link StompSessionManager} implementation to manage a single {@link StompSession}
* over its {@link ListenableFuture} from the target implementation of this class.
* <p>
* The connection to the {@link StompSession} is made during {@link #afterPropertiesSet()}.
* <p>
* The {@link #destroy()} lifecycle method manages {@link StompSession#disconnect()}.
* <p>
* The {@link #connect(StompSessionHandler)} and {@link #disconnect(StompSessionHandler)} method
* implementations populate/remove the provided {@link StompSessionHandler} to/from an internal
* {@link AbstractStompSessionManager.CompositeStompSessionHandler}, which delegates all operations
* to the provided {@link StompSessionHandler}s.
* This {@link AbstractStompSessionManager.CompositeStompSessionHandler} is used for the
* {@link StompSession} connection.
*
* @author Artem Bilan
* @since 4.2
*/
public abstract class AbstractStompSessionManager implements StompSessionManager, ApplicationEventPublisherAware,
InitializingBean, DisposableBean, BeanNameAware {
protected final Log logger = LogFactory.getLog(getClass());
private final CompositeStompSessionHandler compositeStompSessionHandler = new CompositeStompSessionHandler();
protected final StompClientSupport stompClient;
private ApplicationEventPublisher applicationEventPublisher;
private volatile StompHeaders connectHeaders;
private volatile ListenableFuture<StompSession> stompSessionListenableFuture;
private volatile boolean autoReceipt;
private volatile boolean connected;
private String name;
public AbstractStompSessionManager(StompClientSupport stompClient) {
Assert.notNull(stompClient, "'stompClient' is required.");
this.stompClient = stompClient;
}
public void setConnectHeaders(StompHeaders connectHeaders) {
this.connectHeaders = connectHeaders;
}
public void setAutoReceipt(boolean autoReceipt) {
this.autoReceipt = autoReceipt;
}
@Override
public boolean isAutoReceiptEnabled() {
return this.autoReceipt;
}
@Override
public boolean isConnected() {
return this.connected;
}
@Override
public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
this.applicationEventPublisher = applicationEventPublisher;
}
@Override
public void setBeanName(String name) {
this.name = name;
}
@Override
public void afterPropertiesSet() throws Exception {
this.stompSessionListenableFuture = doConnect(this.compositeStompSessionHandler);
this.stompSessionListenableFuture.addCallback(new ListenableFutureCallback<StompSession>() {
@Override
public void onFailure(Throwable e) {
logger.error("STOMP connect error.", e);
if (applicationEventPublisher != null) {
applicationEventPublisher.publishEvent(
new StompExceptionEvent(AbstractStompSessionManager.this, e));
}
}
@Override
public void onSuccess(StompSession stompSession) {
stompSession.setAutoReceipt(autoReceipt);
AbstractStompSessionManager.this.connected = true;
}
});
}
@Override
public void destroy() throws Exception {
this.stompSessionListenableFuture.addCallback(new ListenableFutureCallback<StompSession>() {
@Override
public void onFailure(Throwable ex) {
AbstractStompSessionManager.this.connected = false;
}
@Override
public void onSuccess(StompSession session) {
session.disconnect();
AbstractStompSessionManager.this.connected = false;
}
});
}
@Override
public void connect(StompSessionHandler handler) {
this.compositeStompSessionHandler.addHandler(handler);
}
@Override
public void disconnect(StompSessionHandler handler) {
this.compositeStompSessionHandler.removeHandler(handler);
}
protected StompHeaders getConnectHeaders() {
return connectHeaders;
}
@Override
public String toString() {
return "StompSessionManager{" +
"connected=" + connected +
", name='" + name + '\'' +
'}';
}
protected abstract ListenableFuture<StompSession> doConnect(StompSessionHandler handler);
private static class CompositeStompSessionHandler extends StompSessionHandlerAdapter {
private final List<StompSessionHandler> delegates = new ArrayList<StompSessionHandler>();
private volatile StompSession session;
private volatile StompHeaders connectedHeaders;
void addHandler(StompSessionHandler delegate) {
if (this.session != null) {
delegate.afterConnected(this.session, this.connectedHeaders);
}
this.delegates.add(delegate);
}
void removeHandler(StompSessionHandler delegate) {
this.delegates.remove(delegate);
}
@Override
public void afterConnected(StompSession session, StompHeaders connectedHeaders) {
this.session = session;
this.connectedHeaders = connectedHeaders;
for (StompSessionHandler delegate : this.delegates) {
delegate.afterConnected(session, connectedHeaders);
}
}
@Override
public void handleException(StompSession session, StompCommand command, StompHeaders headers, byte[] payload,
Throwable exception) {
for (StompSessionHandler delegate : this.delegates) {
delegate.handleException(session, command, headers, payload, exception);
}
}
@Override
public void handleTransportError(StompSession session, Throwable exception) {
for (StompSessionHandler delegate : this.delegates) {
delegate.handleTransportError(session, exception);
}
}
@Override
public void handleFrame(StompHeaders headers, Object payload) {
for (StompSessionHandler delegate : this.delegates) {
delegate.handleFrame(headers, payload);
}
}
}
}

42
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/Reactor2TcpStompSessionManager.java Normal file
View File

@@ -0,0 +1,42 @@
/*
* Copyright 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.integration.stomp;
import org.springframework.messaging.simp.stomp.Reactor2TcpStompClient;
import org.springframework.messaging.simp.stomp.StompSession;
import org.springframework.messaging.simp.stomp.StompSessionHandler;
import org.springframework.util.concurrent.ListenableFuture;
/**
* The {@link Reactor2TcpStompClient} based {@link AbstractStompSessionManager} implementation.
*
* @author Artem Bilan
* @see Reactor2TcpStompClient
* @since 4.2
*/
public class Reactor2TcpStompSessionManager extends AbstractStompSessionManager {
public Reactor2TcpStompSessionManager(Reactor2TcpStompClient reactor2TcpStompClient) {
super(reactor2TcpStompClient);
}
@Override
protected ListenableFuture<StompSession> doConnect(StompSessionHandler handler) {
return ((Reactor2TcpStompClient) this.stompClient).connect(getConnectHeaders(), handler);
}
}

38
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/StompSessionManager.java Normal file
View File

@@ -0,0 +1,38 @@
/*
* Copyright 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.integration.stomp;
import org.springframework.messaging.simp.stomp.StompSessionHandler;
/**
* An abstraction to manage the STOMP Session and connection/disconnection
* for {@link StompSessionHandler}.
*
* @author Artem Bilan
* @since 4.2
*/
public interface StompSessionManager {
void connect(StompSessionHandler handler);
void disconnect(StompSessionHandler handler);
boolean isAutoReceiptEnabled();
boolean isConnected();
}

58
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/WebSocketStompSessionManager.java Normal file
View File

@@ -0,0 +1,58 @@
/*
* Copyright 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.integration.stomp;
import org.springframework.messaging.simp.stomp.StompSession;
import org.springframework.messaging.simp.stomp.StompSessionHandler;
import org.springframework.util.Assert;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.socket.WebSocketHttpHeaders;
import org.springframework.web.socket.messaging.WebSocketStompClient;
/**
* The {@link WebSocketStompClient} based {@link AbstractStompSessionManager} implementation.
*
* @author Artem Bilan
* @see WebSocketStompClient
* @since 4.2
*/
public class WebSocketStompSessionManager extends AbstractStompSessionManager {
private final String url;
private final Object[] uriVariables;
private volatile WebSocketHttpHeaders handshakeHeaders;
public WebSocketStompSessionManager(WebSocketStompClient webSocketStompClient, String url, Object... uriVariables) {
super(webSocketStompClient);
Assert.hasText(url, "'url' must not be empty.");
this.url = url;
this.uriVariables = uriVariables;
}
public void setHandshakeHeaders(WebSocketHttpHeaders handshakeHeaders) {
this.handshakeHeaders = handshakeHeaders;
}
@Override
protected ListenableFuture<StompSession> doConnect(StompSessionHandler handler) {
return ((WebSocketStompClient) this.stompClient).connect(this.url, handler, this.handshakeHeaders,
getConnectHeaders(), this.uriVariables);
}
}

31
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/config/StompNamespaceHandler.java Normal file
View File

@@ -0,0 +1,31 @@
/*
* Copyright 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.integration.stomp.config;
import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHandler;
/**
* @author Artem Bilan
* @since 4.2
*/
public class StompNamespaceHandler extends AbstractIntegrationNamespaceHandler {
public void init() {
}
}

4
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/config/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* Contains parser classes for the STOMP namespace support.
*/
package org.springframework.integration.stomp.config;

32
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompExceptionEvent.java Normal file
View File

@@ -0,0 +1,32 @@
/*
* Copyright 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.integration.stomp.event;
/**
* The {@link StompIntegrationEvent} implementation for the exception from STOMP Adapters.
*
* @author Artem Bilan
* @since 4.2
*/
@SuppressWarnings("serial")
public class StompExceptionEvent extends StompIntegrationEvent {
public StompExceptionEvent(Object source, Throwable cause) {
super(source, cause);
}
}

38
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompIntegrationEvent.java Normal file
View File

@@ -0,0 +1,38 @@
/*
* Copyright 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.integration.stomp.event;
import org.springframework.context.ApplicationEvent;
import org.springframework.integration.event.IntegrationEvent;
/**
* Base class for all {@link ApplicationEvent}s generated by the STOMP Adapters.
*
* @author Artem Bilan
* @since 4.2
*/
@SuppressWarnings("serial")
public abstract class StompIntegrationEvent extends IntegrationEvent {
public StompIntegrationEvent(Object source) {
super(source);
}
public StompIntegrationEvent(Object source, Throwable cause) {
super(source, cause);
}
}

75
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompReceiptEvent.java Normal file
View File

@@ -0,0 +1,75 @@
/*
* Copyright 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.integration.stomp.event;
import org.springframework.messaging.Message;
import org.springframework.messaging.simp.stomp.StompCommand;
/**
* The {@link StompIntegrationEvent} for the STOMP {@code RECEIPT} Frames or lost receipts.
*
* @author Artem Bilan
* @since 4.2
* @see org.springframework.integration.stomp.inbound.StompInboundChannelAdapter
* @see org.springframework.integration.stomp.outbound.StompMessageHandler
*/
@SuppressWarnings("serial")
public class StompReceiptEvent extends StompIntegrationEvent {
private final String destination;
private final String receiptId;
private final StompCommand stompCommand;
private final boolean lost;
private Message<?> message;
public StompReceiptEvent(Object source, String destination, String receiptId, StompCommand stompCommand,
boolean lost) {
super(source);
this.destination = destination;
this.receiptId = receiptId;
this.stompCommand = stompCommand;
this.lost = lost;
}
public String getDestination() {
return destination;
}
public String getReceiptId() {
return receiptId;
}
public StompCommand getStompCommand() {
return stompCommand;
}
public boolean isLost() {
return lost;
}
public Message<?> getMessage() {
return message;
}
public void setMessage(Message<?> message) {
this.message = message;
}
}

297
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java Normal file
View File

@@ -0,0 +1,297 @@
/*
* Copyright 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.integration.stomp.inbound;
import java.lang.reflect.Type;
import java.util.HashMap;
import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.mapping.HeaderMapper;
import org.springframework.integration.stomp.StompSessionManager;
import org.springframework.integration.stomp.event.StompReceiptEvent;
import org.springframework.integration.stomp.support.StompHeaderMapper;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.simp.stomp.StompCommand;
import org.springframework.messaging.simp.stomp.StompFrameHandler;
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.simp.stomp.StompSessionHandlerAdapter;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.Assert;
/**
* The {@link MessageProducerSupport} for STOMP protocol to handle STOMP frames from
* provided destination and send messages to the {@code outputChannel}.
* <p>
* Destinations can be added and removed at runtime.
* <p>
* The {@link StompReceiptEvent} is emitted for each {@code Subscribe STOMP frame}
* if provided {@link StompSessionManager} supports {@code autoReceiptEnabled}.
*
* @author Artem Bilan
* @since 4.2
*/
@ManagedResource
@IntegrationManagedResource
public class StompInboundChannelAdapter extends MessageProducerSupport implements ApplicationEventPublisherAware {
private final StompSessionHandler stompSessionHandler = new IntegrationInboundStompSessionHandler();
private final Set<String> destinations = new LinkedHashSet<String>();
private final StompSessionManager stompSessionManager;
private final Map<String, StompSession.Subscription> subscriptions =
new HashMap<String, StompSession.Subscription>();
private final Lock destinationLock = new ReentrantLock();
private ApplicationEventPublisher applicationEventPublisher;
private volatile StompSession stompSession;
private volatile Class<?> payloadType = String.class;
private volatile HeaderMapper<StompHeaders> headerMapper = new StompHeaderMapper();
private volatile MessageChannel errorChannel;
public StompInboundChannelAdapter(StompSessionManager stompSessionManager, String... destinations) {
Assert.notNull(stompSessionManager, "'stompSessionManager' is required.");
if (destinations != null) {
for (String destination : destinations) {
Assert.hasText(destination, "'destinations' must not have empty strings.");
this.destinations.add(destination);
}
}
this.stompSessionManager = stompSessionManager;
}
public void setPayloadType(Class<?> payloadType) {
Assert.notNull(payloadType, "'payloadType' must not be null.");
this.payloadType = payloadType;
}
@Override
public void setErrorChannel(MessageChannel errorChannel) {
super.setErrorChannel(errorChannel);
this.errorChannel = errorChannel;
}
public void setHeaderMapper(HeaderMapper<StompHeaders> headerMapper) {
Assert.notNull(headerMapper, "'headerMapper' must not be null.");
this.headerMapper = headerMapper;
}
@Override
public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
this.applicationEventPublisher = applicationEventPublisher;
}
@ManagedAttribute
public String[] getDestinations() {
this.destinationLock.lock();
try {
return this.destinations.toArray(new String[this.destinations.size()]);
}
finally {
this.destinationLock.unlock();
}
}
/**
* Add a destination (or destinations) to the subscribed list and subscribe it.
* @param destination The destinations.
*/
@ManagedOperation
public void addDestination(String... destination) {
Assert.notNull(destination, "'destination' cannot be null");
this.destinationLock.lock();
try {
for (String d : destination) {
if (this.destinations.add(d)) {
if (this.logger.isDebugEnabled()) {
logger.debug("Subscribe to destination '" + d + "'.");
}
subscribeDestination(d);
}
}
}
finally {
this.destinationLock.unlock();
}
}
/**
* Remove a destination (or destinations) from the subscribed list and unsubscribe it.
* @param destination The destinations.
*/
@ManagedOperation
public void removeDestination(String... destination) {
Assert.notNull(destination, "'destination' cannot be null");
this.destinationLock.lock();
try {
for (String d : destination) {
if (this.destinations.remove(d)) {
if (this.logger.isDebugEnabled()) {
logger.debug("Removed '" + d + "' from subscriptions.");
}
StompSession.Subscription subscription = this.subscriptions.get(d);
if (subscription != null) {
subscription.unsubscribe();
}
else {
if (this.logger.isDebugEnabled()) {
logger.debug("No subscription for destination '" + d + "'.");
}
}
}
}
}
finally {
this.destinationLock.unlock();
}
}
@Override
public String getComponentType() {
return "stomp:inbound-channel-adapter";
}
@Override
protected void doStart() {
this.stompSessionManager.connect(this.stompSessionHandler);
}
@Override
protected void doStop() {
for (StompSession.Subscription subscription : this.subscriptions.values()) {
subscription.unsubscribe();
}
this.subscriptions.clear();
this.stompSessionManager.disconnect(this.stompSessionHandler);
}
private void subscribeDestination(final String destination) {
if (this.stompSession != null) {
final StompSession.Subscription subscription =
this.stompSession.subscribe(destination, new StompFrameHandler() {
@Override
public Type getPayloadType(StompHeaders headers) {
return payloadType;
}
@Override
public void handleFrame(StompHeaders headers, Object body) {
Message<?> message;
if (body instanceof Message) {
message = (Message<?>) body;
}
else {
message = getMessageBuilderFactory().withPayload(body)
.copyHeaders(headerMapper.toHeaders(headers))
.build();
}
sendMessage(message);
}
});
if (this.stompSessionManager.isAutoReceiptEnabled()) {
if (this.applicationEventPublisher != null) {
subscription.addReceiptTask(new Runnable() {
@Override
public void run() {
StompReceiptEvent event = new StompReceiptEvent(StompInboundChannelAdapter.this,
destination, subscription.getReceiptId(), StompCommand.SUBSCRIBE, false);
applicationEventPublisher.publishEvent(event);
}
});
}
subscription.addReceiptLostTask(new Runnable() {
@Override
public void run() {
if (applicationEventPublisher != null) {
StompReceiptEvent event = new StompReceiptEvent(StompInboundChannelAdapter.this,
destination, subscription.getReceiptId(), StompCommand.SUBSCRIBE, true);
applicationEventPublisher.publishEvent(event);
}
else {
logger.error("The receipt [" + subscription.getReceiptId() + "] is lost for [" +
subscription.getSubscriptionId() + "] on destination [" + destination + "]");
}
}
});
}
this.subscriptions.put(destination, subscription);
}
else {
logger.warn("The StompInboundChannelAdapter [" + getComponentName() +
"] hasn't been connected to StompSession. Check the state of [" + this.stompSessionManager + "]");
}
}
private class IntegrationInboundStompSessionHandler extends StompSessionHandlerAdapter {
@Override
public void afterConnected(StompSession session, StompHeaders connectedHeaders) {
StompInboundChannelAdapter.this.stompSession = session;
for (String destination : destinations) {
subscribeDestination(destination);
}
}
@Override
public void handleException(StompSession session, StompCommand command, StompHeaders headers, byte[] payload,
Throwable exception) {
if (errorChannel != null) {
StompHeaderAccessor headerAccessor = StompHeaderAccessor.create(command);
headerAccessor.copyHeaders(headerMapper.toHeaders(headers));
Message<byte[]> failedMessage = MessageBuilder.createMessage(payload,
headerAccessor.getMessageHeaders());
getMessagingTemplate().send(errorChannel,
new ErrorMessage(new MessageHandlingException(failedMessage, exception)));
}
else {
logger.error("STOMP Frame handling error.", exception);
}
}
}
}

4
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* Provides classes which represent inbound STOMP components.
*/
package org.springframework.integration.stomp.inbound;

197
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/outbound/StompMessageHandler.java Normal file
View File

@@ -0,0 +1,197 @@
/*
* Copyright 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.integration.stomp.outbound;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.context.Lifecycle;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.Expression;
import org.springframework.integration.expression.IntegrationEvaluationContextAware;
import org.springframework.integration.expression.ValueExpression;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.mapping.HeaderMapper;
import org.springframework.integration.stomp.StompSessionManager;
import org.springframework.integration.stomp.event.StompExceptionEvent;
import org.springframework.integration.stomp.event.StompReceiptEvent;
import org.springframework.integration.stomp.support.StompHeaderMapper;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessagingException;
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.StompSession;
import org.springframework.messaging.simp.stomp.StompSessionHandler;
import org.springframework.messaging.simp.stomp.StompSessionHandlerAdapter;
import org.springframework.util.Assert;
/**
* The {@link AbstractMessageHandler} implemntation to send messages to STOMP destinations.
*
* @author Artem Bilan
* @since 4.2
*/
public class StompMessageHandler extends AbstractMessageHandler implements IntegrationEvaluationContextAware,
ApplicationEventPublisherAware, Lifecycle {
private final StompSessionHandler sessionHandler = new IntegrationOutboundStompSessionHandler();
private final StompSessionManager stompSessionManager;
private volatile StompSession stompSession;
private volatile boolean running;
private volatile HeaderMapper<StompHeaders> headerMapper = new StompHeaderMapper();
private Expression destinationExpression;
private EvaluationContext evaluationContext;
private ApplicationEventPublisher applicationEventPublisher;
public StompMessageHandler(StompSessionManager stompSessionManager) {
Assert.notNull(stompSessionManager, "'stompSessionManager' is required.");
this.stompSessionManager = stompSessionManager;
}
public void setDestination(String destination) {
Assert.hasText(destination, "'destination' must not be empty.");
this.destinationExpression = new ValueExpression<String>(destination);
}
public void setDestinationExpression(Expression destinationExpression) {
Assert.notNull(destinationExpression, "'destinationExpression' must not be null.");
this.destinationExpression = destinationExpression;
}
public void setHeaderMapper(HeaderMapper<StompHeaders> headerMapper) {
Assert.notNull(headerMapper, "'headerMapper' must not be null.");
this.headerMapper = headerMapper;
}
@Override
public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) {
this.evaluationContext = evaluationContext;
}
@Override
public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
this.applicationEventPublisher = applicationEventPublisher;
}
@Override
protected void handleMessageInternal(final Message<?> message) throws Exception {
if (!this.isRunning()) {
throw new MessageDeliveryException(message, "The StompMessageHandler [" + getComponentName() +
"] hasn't been connected to StompSession. Check the state of [" + this.stompSessionManager + "]");
}
StompHeaders stompHeaders = new StompHeaders();
this.headerMapper.fromHeaders(message.getHeaders(), stompHeaders);
if (stompHeaders.getDestination() == null) {
Assert.state(this.destinationExpression != null, "One of 'destination' or 'destinationExpression' must be" +
" provided, if message header doesn't supply 'destination' STOMP header.");
String destination = this.destinationExpression.getValue(this.evaluationContext, message, String.class);
stompHeaders.setDestination(destination);
}
final StompSession.Receiptable receiptable = this.stompSession.send(stompHeaders, message.getPayload());
if (this.stompSessionManager.isAutoReceiptEnabled()) {
final String destination = stompHeaders.getDestination();
if (this.applicationEventPublisher != null) {
receiptable.addReceiptTask(new Runnable() {
@Override
public void run() {
StompReceiptEvent event = new StompReceiptEvent(StompMessageHandler.this,
destination, receiptable.getReceiptId(), StompCommand.SEND, false);
event.setMessage(message);
applicationEventPublisher.publishEvent(event);
}
});
}
receiptable.addReceiptLostTask(new Runnable() {
@Override
public void run() {
if (applicationEventPublisher != null) {
StompReceiptEvent event = new StompReceiptEvent(StompMessageHandler.this,
destination, receiptable.getReceiptId(), StompCommand.SEND, true);
event.setMessage(message);
applicationEventPublisher.publishEvent(event);
}
else {
logger.error("The receipt [" + receiptable.getReceiptId() + "] is lost for [" +
message + "] on destination [" + destination + "]");
}
}
});
}
}
@Override
public void start() {
this.stompSessionManager.connect(this.sessionHandler);
}
@Override
public void stop() {
this.stompSessionManager.disconnect(this.sessionHandler);
this.running = false;
}
@Override
public boolean isRunning() {
return this.running;
}
private class IntegrationOutboundStompSessionHandler extends StompSessionHandlerAdapter {
@Override
public void afterConnected(StompSession session, StompHeaders connectedHeaders) {
StompMessageHandler.this.stompSession = session;
StompMessageHandler.this.running = true;
}
@Override
public void handleFrame(StompHeaders headers, Object payload) {
Object thePayload = payload;
if (thePayload == null) {
thePayload = headers.getFirst(StompHeaderAccessor.STOMP_MESSAGE_HEADER);
}
if (thePayload != null) {
Message<?> failedMessage = getMessageBuilderFactory().withPayload(thePayload)
.copyHeaders(headerMapper.toHeaders(headers))
.build();
MessagingException exception = new MessageDeliveryException(failedMessage,
"STOMP frame handling error.");
logger.error("STOMP frame handling error.", exception);
if (applicationEventPublisher != null) {
applicationEventPublisher.publishEvent(
new StompExceptionEvent(StompMessageHandler.this, exception));
}
}
}
}
}

4
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/outbound/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* Provides classes which represent outbound STOMP components.
*/
package org.springframework.integration.stomp.outbound;

4
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* Provides core classes STOMP components.
*/
package org.springframework.integration.stomp;

68
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/support/IntegrationStompHeaders.java Normal file
View File

@@ -0,0 +1,68 @@
/*
* Copyright 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.integration.stomp.support;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import org.springframework.messaging.simp.stomp.StompHeaders;
/**
* The STOMP headers with Integration-friendly {@code stomp_} prefix.
*
* @author Artem Bilan
* @see StompHeaders
* @since 4.2
*/
public abstract class IntegrationStompHeaders {
public static final String PREFIX = "stomp_";
public static final String RECEIPT = PREFIX + StompHeaders.RECEIPT;
public static final String HOST = PREFIX + StompHeaders.HOST;
public static final String LOGIN = PREFIX + StompHeaders.LOGIN;
public static final String PASSCODE = PREFIX + StompHeaders.PASSCODE;
public static final String HEARTBEAT = PREFIX + StompHeaders.HEARTBEAT;
public static final String SESSION = PREFIX + StompHeaders.SESSION;
public static final String SERVER = PREFIX + StompHeaders.SERVER;
public static final String DESTINATION = PREFIX + StompHeaders.DESTINATION;
public static final String ID = PREFIX + StompHeaders.ID;
public static final String ACK = PREFIX + StompHeaders.ACK;
public static final String SUBSCRIPTION = PREFIX + StompHeaders.SUBSCRIPTION;
public static final String MESSAGE_ID = PREFIX + StompHeaders.MESSAGE_ID;
public static final String RECEIPT_ID = PREFIX + StompHeaders.RECEIPT_ID;
static final Collection<String> HEADERS =
Collections.unmodifiableList(Arrays.asList(StompHeaders.RECEIPT, StompHeaders.HOST, StompHeaders.LOGIN,
StompHeaders.PASSCODE, StompHeaders.HEARTBEAT, StompHeaders.SESSION, StompHeaders.SERVER,
StompHeaders.DESTINATION, StompHeaders.ID, StompHeaders.ACK, StompHeaders.SUBSCRIPTION,
StompHeaders.MESSAGE_ID, StompHeaders.RECEIPT_ID));
}

244
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/support/StompHeaderMapper.java Normal file
View File

@@ -0,0 +1,244 @@
/*
* Copyright 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.integration.stomp.support;
import java.text.MessageFormat;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.http.MediaType;
import org.springframework.integration.mapping.HeaderMapper;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
import org.springframework.messaging.simp.stomp.StompHeaders;
import org.springframework.util.Assert;
import org.springframework.util.MimeType;
import org.springframework.util.MultiValueMap;
import org.springframework.util.PatternMatchUtils;
import org.springframework.util.StringUtils;
/**
* The STOMP {@link HeaderMapper} implementation.
*
* @author Artem Bilan
* @since 4.2
* @see StompHeaders
*/
public class StompHeaderMapper implements HeaderMapper<StompHeaders> {
private static final Log logger = LogFactory.getLog(StompHeaderMapper.class);
public static final String STOMP_INBOUND_HEADER_NAME_PATTERN = "STOMP_INBOUND_HEADERS";
public static final String STOMP_OUTBOUND_HEADER_NAME_PATTERN = "STOMP_OUTBOUND_HEADERS";
private static final String[] STOMP_INBOUND_HEADER_NAMES = new String[] {
StompHeaders.CONTENT_LENGTH,
StompHeaders.CONTENT_TYPE,
StompHeaders.MESSAGE_ID,
StompHeaders.RECEIPT_ID,
StompHeaders.SUBSCRIPTION,
};
private final static List<String> STOMP_INBOUND_HEADER_NAMES_LIST =
Arrays.<String>asList(STOMP_INBOUND_HEADER_NAMES);
private static final String[] STOMP_OUTBOUND_HEADER_NAMES = new String[] {
StompHeaders.CONTENT_LENGTH,
StompHeaders.CONTENT_TYPE,
StompHeaders.DESTINATION,
StompHeaders.RECEIPT,
IntegrationStompHeaders.DESTINATION,
IntegrationStompHeaders.RECEIPT
};
private final static List<String> STOMP_OUTBOUND_HEADER_NAMES_LIST =
Arrays.<String>asList(STOMP_OUTBOUND_HEADER_NAMES);
private volatile String[] inboundHeaderNames = STOMP_INBOUND_HEADER_NAMES;
private volatile String[] outboundHeaderNames = STOMP_OUTBOUND_HEADER_NAMES;
public void setInboundHeaderNames(String[] inboundHeaderNames) {//NOSONAR - false positive
Assert.notNull(inboundHeaderNames, "'inboundHeaderNames' must not be null.");
Assert.noNullElements(inboundHeaderNames, "'inboundHeaderNames' must not contains null elements.");
Arrays.sort(inboundHeaderNames);
if (!Arrays.equals(STOMP_INBOUND_HEADER_NAMES, inboundHeaderNames)) {
this.inboundHeaderNames = inboundHeaderNames;
}
}
public void setOutboundHeaderNames(String[] outboundHeaderNames) {//NOSONAR - false positive
Assert.notNull(outboundHeaderNames, "'outboundHeaderNames' must not be null.");
Assert.noNullElements(outboundHeaderNames, "'outboundHeaderNames' must not contains null elements.");
Arrays.sort(outboundHeaderNames);
if (!Arrays.equals(STOMP_OUTBOUND_HEADER_NAMES, outboundHeaderNames)) {
this.outboundHeaderNames = outboundHeaderNames;
}
}
@Override
@SuppressWarnings("unchecked")
public void fromHeaders(MessageHeaders headers, StompHeaders target) {
for (Map.Entry<String, Object> entry : headers.entrySet()) {
String name = entry.getKey();
if (shouldMapHeader(name, this.outboundHeaderNames)) {
Object value = entry.getValue();
if (value != null) {
setStompHeader(target, name, value);
}
}
else if (StompHeaderAccessor.NATIVE_HEADERS.equals(name)) {
MultiValueMap<String, String> multiValueMap =
headers.get(StompHeaderAccessor.NATIVE_HEADERS, MultiValueMap.class);
for (Map.Entry<String, List<String>> entry1 : multiValueMap.entrySet()) {
name = entry1.getKey();
if (shouldMapHeader(name, this.outboundHeaderNames)) {
String value = entry1.getValue().get(0);
if (StringUtils.hasText(value)) {
setStompHeader(target, name, value);
}
}
}
}
}
}
private void setStompHeader(StompHeaders target, String name, Object value) {
if (StompHeaders.CONTENT_LENGTH.equals(name)) {
if (value instanceof Number) {
target.setContentLength(((Number) value).longValue());
}
else if (value instanceof String) {
target.setContentLength(Long.parseLong((String) value));
}
else {
Class<?> clazz = (value != null) ? value.getClass() : null;
throw new IllegalArgumentException(
"Expected Number or String value for 'content-length' header value, but received: " + clazz);
}
}
else if (StompHeaders.CONTENT_TYPE.equals(name) || MessageHeaders.CONTENT_TYPE.equals(name)) {
MimeType contentType = target.getContentType();
if (contentType == null || StompHeaders.CONTENT_TYPE.equals(name)) {
if (value instanceof MediaType) {
target.setContentType((MediaType) value);
}
else if (value instanceof String) {
target.setContentType(MediaType.parseMediaType((String) value));
}
else {
Class<?> clazz = (value != null) ? value.getClass() : null;
throw new IllegalArgumentException(
"Expected MediaType or String value for 'content-type' header value, but received: " + clazz);
}
}
}
else if (StompHeaders.DESTINATION.equals(name) || IntegrationStompHeaders.DESTINATION.equals(name)) {
if (value instanceof String) {
target.setDestination((String) value);
}
else {
Class<?> clazz = (value != null) ? value.getClass() : null;
throw new IllegalArgumentException(
"Expected String value for 'destination' header value, but received: " + clazz);
}
}
else if (StompHeaders.RECEIPT.equals(name) || IntegrationStompHeaders.RECEIPT.equals(name)) {
if (value instanceof String) {
target.setReceipt((String) value);
}
else {
Class<?> clazz = (value != null) ? value.getClass() : null;
throw new IllegalArgumentException(
"Expected String value for 'receipt' header value, but received: " + clazz);
}
}
else {
if (value instanceof String) {
target.set(name, (String) value);
}
else {
Class<?> clazz = (value != null) ? value.getClass() : null;
throw new IllegalArgumentException(
"Expected String value for any generic STOMP header value, but received: " + clazz);
}
}
}
@Override
public Map<String, Object> toHeaders(StompHeaders source) {
Map<String, Object> target = new HashMap<String, Object>();
for (String name : source.keySet()) {
if (shouldMapHeader(name, this.inboundHeaderNames)) {
if (StompHeaders.CONTENT_TYPE.equals(name)) {
target.put(MessageHeaders.CONTENT_TYPE, source.getContentType());
}
else {
String key = name;
if (IntegrationStompHeaders.HEADERS.contains(name)) {
key = IntegrationStompHeaders.PREFIX + name;
}
target.put(key, source.getFirst(name));
}
}
}
return target;
}
private boolean shouldMapHeader(String headerName, String[] patterns) {
if (patterns != null && patterns.length > 0) {
for (String pattern : patterns) {
if (PatternMatchUtils.simpleMatch(pattern, headerName)) {
if (logger.isDebugEnabled()) {
logger.debug(MessageFormat.format("headerName=[{0}] WILL be mapped, matched pattern={1}",
headerName, pattern));
}
return true;
}
else if (STOMP_INBOUND_HEADER_NAME_PATTERN.equals(pattern)
&& STOMP_INBOUND_HEADER_NAMES_LIST.contains(headerName)) {
if (logger.isDebugEnabled()) {
logger.debug(MessageFormat.format("headerName=[{0}] WILL be mapped, matched pattern={1}",
headerName, pattern));
}
return true;
}
else if (STOMP_OUTBOUND_HEADER_NAME_PATTERN.equals(pattern)
&& (STOMP_OUTBOUND_HEADER_NAMES_LIST.contains(headerName)
|| MessageHeaders.CONTENT_TYPE.equals(headerName))) {
if (logger.isDebugEnabled()) {
logger.debug(MessageFormat.format("headerName=[{0}] WILL be mapped, matched pattern={1}",
headerName, pattern));
}
return true;
}
}
}
if (logger.isDebugEnabled()) {
logger.debug(MessageFormat.format("headerName=[{0}] WILL NOT be mapped", headerName));
}
return false;
}
}

7
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/support/package-info.java Normal file
View File

@@ -0,0 +1,7 @@
/**
* Provides classes to support STOMP components.
* @author Artem Bilan
* @since 4.2
*/
package org.springframework.integration.stomp.support;

1
spring-integration-stomp/src/main/resources/META-INF/spring.handlers Normal file
View File

@@ -0,0 +1 @@
http\://www.springframework.org/schema/integration/stomp=org.springframework.integration.stomp.config.StompNamespaceHandler

2
spring-integration-stomp/src/main/resources/META-INF/spring.schemas Normal file
View File

@@ -0,0 +1,2 @@
http\://www.springframework.org/schema/integration/stomp/spring-integration-stomp-4.2.xsd=org/springframework/integration/stomp/config/spring-integration-stomp-4.2.xsd
http\://www.springframework.org/schema/integration/stomp/spring-integration-stomp.xsd=org/springframework/integration/stomp/config/spring-integration-stomp-4.2.xsd

4
spring-integration-stomp/src/main/resources/META-INF/spring.tooling Normal file
View File

@@ -0,0 +1,4 @@
# Tooling related information for the integration STOMP namespace
http\://www.springframework.org/schema/integration/stomp@name=integration STOMP Namespace
http\://www.springframework.org/schema/integration/stomp@prefix=int-stomp
http\://www.springframework.org/schema/integration/stomp@icon=org/springframework/integration/stomp/config/spring-integration-stomp.gif

23
spring-integration-stomp/src/main/resources/org/springframework/integration/stomp/config/spring-integration-stomp-4.2.xsd Normal file
View File

@@ -0,0 +1,23 @@
<?xml version="1.0" encoding="UTF-8"?>
<xsd:schema xmlns="http://www.springframework.org/schema/integration/websocket"
xmlns:xsd="http://www.w3.org/2001/XMLSchema"
xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:tool="http://www.springframework.org/schema/tool"
xmlns:integration="http://www.springframework.org/schema/integration"
targetNamespace="http://www.springframework.org/schema/integration/websocket"
elementFormDefault="qualified"
attributeFormDefault="unqualified">
<xsd:import namespace="http://www.springframework.org/schema/beans"/>
<xsd:import namespace="http://www.springframework.org/schema/tool"/>
<xsd:import namespace="http://www.springframework.org/schema/integration"
schemaLocation="http://www.springframework.org/schema/integration/spring-integration-4.2.xsd"/>
<xsd:annotation>
<xsd:documentation><![CDATA[
Defines the configuration elements for Spring Integration's STOMP adapters.
]]></xsd:documentation>
</xsd:annotation>
</xsd:schema>

BIN
spring-integration-stomp/src/main/resources/org/springframework/integration/stomp/config/spring-integration-stomp.gif Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 578 B

167
spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java Normal file
View File

@@ -0,0 +1,167 @@
/*
* Copyright 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.integration.stomp.client;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import org.apache.activemq.broker.BrokerService;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.Lifecycle;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.stomp.Reactor2TcpStompSessionManager;
import org.springframework.integration.stomp.StompSessionManager;
import org.springframework.integration.stomp.inbound.StompInboundChannelAdapter;
import org.springframework.integration.stomp.outbound.StompMessageHandler;
import org.springframework.integration.support.converter.PassThruMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.simp.stomp.Reactor2TcpStompClient;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.util.SocketUtils;
/**
* @author Artem Bilan
* @since 4.2
*/
public class StompServerIntegrationTests {
private static BrokerService activeMQBroker;
private static Reactor2TcpStompClient stompClient;
@BeforeClass
public static void setup() throws Exception {
int port = SocketUtils.findAvailableTcpPort(61613);
activeMQBroker = new BrokerService();
activeMQBroker.addConnector("stomp://127.0.0.1:" + port);
activeMQBroker.setStartAsync(false);
activeMQBroker.setPersistent(false);
activeMQBroker.setUseJmx(false);
activeMQBroker.getSystemUsage().getMemoryUsage().setLimit(1024 * 1024 * 5);
activeMQBroker.getSystemUsage().getTempUsage().setLimit(1024 * 1024 * 5);
activeMQBroker.start();
stompClient = new Reactor2TcpStompClient("127.0.0.1", port);
stompClient.setMessageConverter(new PassThruMessageConverter());
}
@AfterClass
public static void teardown() throws Exception {
activeMQBroker.stop();
}
@Test
public void testStompAdapters() {
ConfigurableApplicationContext context1 = new AnnotationConfigApplicationContext(ContextConfiguration.class);
ConfigurableApplicationContext context2 = new AnnotationConfigApplicationContext(ContextConfiguration.class);
PollableChannel stompInputChannel1 = context1.getBean("stompInputChannel", PollableChannel.class);
PollableChannel stompInputChannel2 = context2.getBean("stompInputChannel", PollableChannel.class);
MessageChannel stompOutputChannel1 = context1.getBean("stompOutputChannel", MessageChannel.class);
MessageChannel stompOutputChannel2 = context2.getBean("stompOutputChannel", MessageChannel.class);
stompOutputChannel1.send(new GenericMessage<byte[]>("Hello, Client#2!".getBytes()));
Message<?> receive11 = stompInputChannel1.receive(10000);
Message<?> receive21 = stompInputChannel2.receive(10000);
assertNotNull(receive11);
assertNotNull(receive21);
assertArrayEquals("Hello, Client#2!".getBytes(), (byte[]) receive11.getPayload());
assertArrayEquals("Hello, Client#2!".getBytes(), (byte[]) receive21.getPayload());
stompOutputChannel2.send(new GenericMessage<byte[]>("Hello, Client#1!".getBytes()));
Message<?> receive12 = stompInputChannel1.receive(10000);
Message<?> receive22 = stompInputChannel2.receive(10000);
assertNotNull(receive12);
assertNotNull(receive22);
assertArrayEquals("Hello, Client#1!".getBytes(), (byte[]) receive12.getPayload());
assertArrayEquals("Hello, Client#1!".getBytes(), (byte[]) receive22.getPayload());
Lifecycle stompInboundChannelAdapter2 = context2.getBean("stompInboundChannelAdapter", Lifecycle.class);
stompInboundChannelAdapter2.stop();
stompOutputChannel1.send(new GenericMessage<byte[]>("How do you do?".getBytes()));
Message<?> receive13 = stompInputChannel1.receive(10000);
assertNotNull(receive13);
Message<?> receive23 = stompInputChannel2.receive(100);
assertNull(receive23);
stompInboundChannelAdapter2.start();
stompOutputChannel1.send(new GenericMessage<byte[]>("???".getBytes()));
Message<?> receive24 = stompInputChannel2.receive(10000);
assertNotNull(receive24);
assertArrayEquals("???".getBytes(), (byte[]) receive24.getPayload());
context1.close();
context2.close();
}
@Configuration
@EnableIntegration
public static class ContextConfiguration {
@Bean
public StompSessionManager stompSessionManager() {
return new Reactor2TcpStompSessionManager(stompClient);
}
@Bean
public PollableChannel stompInputChannel() {
return new QueueChannel();
}
@Bean
public StompInboundChannelAdapter stompInboundChannelAdapter() {
StompInboundChannelAdapter adapter =
new StompInboundChannelAdapter(stompSessionManager(), "/topic/myTopic");
adapter.setOutputChannel(stompInputChannel());
return adapter;
}
@Bean
@ServiceActivator(inputChannel = "stompOutputChannel")
public MessageHandler stompMessageHandler() {
StompMessageHandler handler = new StompMessageHandler(stompSessionManager());
handler.setDestination("/topic/myTopic");
return handler;
}
}
}

267
spring-integration-stomp/src/test/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapterWebSocketIntegrationTests.java Normal file
View File

@@ -0,0 +1,267 @@
/*
* Copyright 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.integration.stomp.inbound;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.core.IsInstanceOf.instanceOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
import java.util.Collections;
import java.util.Map;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.MediaType;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.event.inbound.ApplicationEventListeningMessageProducer;
import org.springframework.integration.stomp.StompSessionManager;
import org.springframework.integration.stomp.WebSocketStompSessionManager;
import org.springframework.integration.stomp.event.StompIntegrationEvent;
import org.springframework.integration.stomp.event.StompReceiptEvent;
import org.springframework.integration.websocket.TomcatWebSocketTestServer;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.converter.MappingJackson2MessageConverter;
import org.springframework.messaging.converter.MessageConversionException;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.messaging.simp.stomp.StompCommand;
import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
import org.springframework.messaging.support.AbstractSubscribableChannel;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.web.socket.client.WebSocketClient;
import org.springframework.web.socket.client.standard.StandardWebSocketClient;
import org.springframework.web.socket.config.annotation.AbstractWebSocketMessageBrokerConfigurer;
import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
import org.springframework.web.socket.messaging.SessionSubscribeEvent;
import org.springframework.web.socket.messaging.WebSocketStompClient;
import org.springframework.web.socket.server.standard.TomcatRequestUpgradeStrategy;
import org.springframework.web.socket.server.support.DefaultHandshakeHandler;
import org.springframework.web.socket.sockjs.client.SockJsClient;
import org.springframework.web.socket.sockjs.client.Transport;
import org.springframework.web.socket.sockjs.client.WebSocketTransport;
/**
* @author Artem Bilan
* @since 4.2
*/
@ContextConfiguration(classes = StompInboundChannelAdapterWebSocketIntegrationTests.ContextConfiguration.class)
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class StompInboundChannelAdapterWebSocketIntegrationTests {
@Value("#{server.serverContext}")
private ApplicationContext serverContext;
@Autowired
@Qualifier("stompInputChannel")
private PollableChannel stompInputChannel;
@Autowired
@Qualifier("errorChannel")
private PollableChannel errorChannel;
@Autowired
@Qualifier("stompEvents")
private PollableChannel stompEvents;
@Autowired
private StompInboundChannelAdapter stompInboundChannelAdapter;
@Test
public void testWebSocketStompClient() throws InterruptedException {
Message<?> receive = this.stompEvents.receive(10000);
assertNotNull(receive);
assertThat(receive.getPayload(), instanceOf(StompReceiptEvent.class));
StompReceiptEvent stompReceiptEvent = (StompReceiptEvent) receive.getPayload();
assertEquals(StompCommand.SUBSCRIBE, stompReceiptEvent.getStompCommand());
assertEquals("/topic/myTopic", stompReceiptEvent.getDestination());
SimpMessagingTemplate messagingTemplate = this.serverContext.getBean("brokerMessagingTemplate",
SimpMessagingTemplate.class);
StompHeaderAccessor stompHeaderAccessor = StompHeaderAccessor.create(StompCommand.MESSAGE);
stompHeaderAccessor.setContentType(MediaType.APPLICATION_JSON);
stompHeaderAccessor.updateStompCommandAsServerMessage();
stompHeaderAccessor.setLeaveMutable(true);
messagingTemplate.send("/topic/myTopic",
MessageBuilder.createMessage("{\"foo\": \"bar\"}".getBytes(), stompHeaderAccessor.getMessageHeaders()));
receive = this.stompInputChannel.receive(10000);
assertNotNull(receive);
assertThat(receive.getPayload(), instanceOf(Map.class));
@SuppressWarnings("unchecked")
Map<String, String> payload = (Map<String, String>) receive.getPayload();
String foo = payload.get("foo");
assertNotNull(foo);
assertEquals("bar", foo);
this.stompInboundChannelAdapter.removeDestination("/topic/myTopic");
messagingTemplate.convertAndSend("/topic/myTopic", "foo");
receive = this.stompInputChannel.receive(1000);
assertNull(receive);
this.stompInboundChannelAdapter.addDestination("/topic/myTopic");
receive = this.stompEvents.receive(10000);
assertNotNull(receive);
messagingTemplate.convertAndSend("/topic/myTopic", "foo");
receive = this.errorChannel.receive(10000);
assertNotNull(receive);
assertThat(receive, instanceOf(ErrorMessage.class));
ErrorMessage errorMessage = (ErrorMessage) receive;
Throwable throwable = errorMessage.getPayload();
assertThat(throwable, instanceOf(MessageHandlingException.class));
assertThat(throwable.getCause(), instanceOf(MessageConversionException.class));
assertThat(throwable.getMessage(), containsString("No suitable converter, payloadType=interface java.util.Map"));
}
// STOMP Client
@Configuration
@EnableIntegration
public static class ContextConfiguration {
@Bean
public TomcatWebSocketTestServer server() {
return new TomcatWebSocketTestServer(ServerConfig.class);
}
@Bean
public WebSocketClient webSocketClient() {
return new SockJsClient(Collections.<Transport>singletonList(new WebSocketTransport(new StandardWebSocketClient())));
}
@Bean
public WebSocketStompClient stompClient(TaskScheduler taskScheduler) {
WebSocketStompClient webSocketStompClient = new WebSocketStompClient(webSocketClient());
webSocketStompClient.setMessageConverter(new MappingJackson2MessageConverter());
webSocketStompClient.setTaskScheduler(taskScheduler);
return webSocketStompClient;
}
@Bean
public StompSessionManager stompSessionManager(WebSocketStompClient stompClient) {
WebSocketStompSessionManager webSocketStompSessionManager =
new WebSocketStompSessionManager(stompClient, server().getWsBaseUrl() + "/ws");
webSocketStompSessionManager.setAutoReceipt(true);
return webSocketStompSessionManager;
}
@Bean
public PollableChannel stompInputChannel() {
return new QueueChannel();
}
@Bean
public PollableChannel errorChannel() {
return new QueueChannel();
}
@Bean
public StompInboundChannelAdapter stompInboundChannelAdapter(StompSessionManager stompSessionFactory) {
StompInboundChannelAdapter adapter = new StompInboundChannelAdapter(stompSessionFactory, "/topic/myTopic");
adapter.setPayloadType(Map.class);
adapter.setOutputChannel(stompInputChannel());
adapter.setErrorChannel(errorChannel());
return adapter;
}
@Bean
public PollableChannel stompEvents() {
return new QueueChannel();
}
@Bean
@SuppressWarnings("unchecked")
public ApplicationListener<ApplicationEvent> stompEventListener() {
ApplicationEventListeningMessageProducer producer = new ApplicationEventListeningMessageProducer();
producer.setEventTypes(StompIntegrationEvent.class);
producer.setOutputChannel(stompEvents());
return producer;
}
}
// WebSocket Server part
@Configuration
@EnableWebSocketMessageBroker
static class ServerConfig extends AbstractWebSocketMessageBrokerConfigurer {
@Bean
public DefaultHandshakeHandler handshakeHandler() {
return new DefaultHandshakeHandler(new TomcatRequestUpgradeStrategy());
}
@Override
public void registerStompEndpoints(StompEndpointRegistry registry) {
registry.addEndpoint("/ws").setHandshakeHandler(handshakeHandler()).withSockJS();
}
@Override
public void configureMessageBroker(MessageBrokerRegistry configurer) {
configurer.setApplicationDestinationPrefixes("/app");
configurer.enableSimpleBroker("/topic", "/queue");
}
//TODO SimpleBrokerMessageHandler doesn't support RECEIPT frame, hence we emulate it this way
@Bean
@SuppressWarnings("unchecked")
public ApplicationListener<SessionSubscribeEvent> webSocketEventListener(
final AbstractSubscribableChannel clientOutboundChannel) {
return new ApplicationListener<SessionSubscribeEvent>() {
@Override
public void onApplicationEvent(SessionSubscribeEvent event) {
Message<byte[]> message = event.getMessage();
StompHeaderAccessor stompHeaderAccessor = StompHeaderAccessor.wrap(message);
if (stompHeaderAccessor.getReceipt() != null) {
stompHeaderAccessor.setHeader("stompCommand", StompCommand.RECEIPT);
stompHeaderAccessor.setReceiptId(stompHeaderAccessor.getReceipt());
clientOutboundChannel.send(
MessageBuilder.createMessage(new byte[0], stompHeaderAccessor.getMessageHeaders()));
}
}
};
}
}
}

283
spring-integration-stomp/src/test/java/org/springframework/integration/stomp/outbound/StompMessageHandlerWebSocketIntegrationTests.java Normal file
View File

@@ -0,0 +1,283 @@
/*
* Copyright 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.integration.stomp.outbound;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.core.IsInstanceOf.instanceOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import java.util.Collections;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.event.inbound.ApplicationEventListeningMessageProducer;
import org.springframework.integration.stomp.StompSessionManager;
import org.springframework.integration.stomp.WebSocketStompSessionManager;
import org.springframework.integration.stomp.event.StompExceptionEvent;
import org.springframework.integration.stomp.event.StompIntegrationEvent;
import org.springframework.integration.stomp.event.StompReceiptEvent;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.websocket.TomcatWebSocketTestServer;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.converter.StringMessageConverter;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.messaging.simp.config.ChannelRegistration;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.messaging.simp.stomp.StompCommand;
import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
import org.springframework.messaging.support.ChannelInterceptorAdapter;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.stereotype.Controller;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.web.socket.client.WebSocketClient;
import org.springframework.web.socket.client.standard.StandardWebSocketClient;
import org.springframework.web.socket.config.annotation.AbstractWebSocketMessageBrokerConfigurer;
import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
import org.springframework.web.socket.messaging.WebSocketStompClient;
import org.springframework.web.socket.server.standard.TomcatRequestUpgradeStrategy;
import org.springframework.web.socket.server.support.DefaultHandshakeHandler;
import org.springframework.web.socket.sockjs.client.SockJsClient;
import org.springframework.web.socket.sockjs.client.Transport;
import org.springframework.web.socket.sockjs.client.WebSocketTransport;
/**
* @author Artem Bilan
* @author Gary Russell
* @since 4.2
*/
@ContextConfiguration(classes = StompMessageHandlerWebSocketIntegrationTests.ContextConfiguration.class)
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class StompMessageHandlerWebSocketIntegrationTests {
@Value("#{server.serverContext}")
private ApplicationContext serverContext;
@Autowired
@Qualifier("stompMessageHandler")
private MessageHandler stompMessageHandler;
@Autowired
@Qualifier("webSocketOutputChannel")
private MessageChannel webSocketOutputChannel;
@Autowired
@Qualifier("stompEvents")
private PollableChannel stompEvents;
@Test
public void testStompMessageHandler() throws InterruptedException {
int n = 0;
while (TestUtils.getPropertyValue(this.stompMessageHandler, "stompSession") == null && n++ < 10) {
Thread.sleep(50);
}
assertTrue(n < 10);
StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.SEND);
headers.setDestination("/app/simple");
Message<String> message = MessageBuilder.withPayload("foo").setHeaders(headers).build();
this.webSocketOutputChannel.send(message);
SimpleController controller = this.serverContext.getBean(SimpleController.class);
assertTrue(controller.latch.await(10, TimeUnit.SECONDS));
// Simple Broker Relay doesn't support RECEIPT Frame, so we check here the 'lost' StompReceiptEvent
Message<?> receive = this.stompEvents.receive(10000);
assertNotNull(receive);
assertThat(receive.getPayload(), instanceOf(StompReceiptEvent.class));
StompReceiptEvent stompReceiptEvent = (StompReceiptEvent) receive.getPayload();
assertEquals(StompCommand.SEND, stompReceiptEvent.getStompCommand());
assertEquals("/app/simple", stompReceiptEvent.getDestination());
assertTrue(stompReceiptEvent.isLost());
assertNotNull(stompReceiptEvent.getMessage());
headers = StompHeaderAccessor.create(StompCommand.SEND);
headers.setDestination("/foo");
message = MessageBuilder.withPayload("bar").setHeaders(headers).build();
this.webSocketOutputChannel.send(message);
receive = this.stompEvents.receive(10000);
assertNotNull(receive);
assertThat(receive.getPayload(), instanceOf(StompExceptionEvent.class));
StompExceptionEvent stompExceptionEvent = (StompExceptionEvent) receive.getPayload();
Throwable cause = stompExceptionEvent.getCause();
assertThat(cause, instanceOf(MessageDeliveryException.class));
MessageDeliveryException messageDeliveryException = (MessageDeliveryException) cause;
Message<?> failedMessage = messageDeliveryException.getFailedMessage();
assertThat((String) failedMessage.getPayload(), containsString("preSend intentional Exception"));
receive = this.stompEvents.receive(10000);
assertNotNull(receive);
assertThat(receive.getPayload(), instanceOf(StompReceiptEvent.class));
stompReceiptEvent = (StompReceiptEvent) receive.getPayload();
assertEquals(StompCommand.SEND, stompReceiptEvent.getStompCommand());
assertEquals("/foo", stompReceiptEvent.getDestination());
assertTrue(stompReceiptEvent.isLost());
}
// STOMP Client
@Configuration
@EnableIntegration
public static class ContextConfiguration {
@Bean
public TomcatWebSocketTestServer server() {
return new TomcatWebSocketTestServer(ServerConfig.class);
}
@Bean
public WebSocketClient webSocketClient() {
return new SockJsClient(Collections.<Transport>singletonList(new WebSocketTransport(new StandardWebSocketClient())));
}
@Bean
public WebSocketStompClient stompClient(TaskScheduler taskScheduler) {
WebSocketStompClient webSocketStompClient = new WebSocketStompClient(webSocketClient());
webSocketStompClient.setTaskScheduler(taskScheduler);
webSocketStompClient.setReceiptTimeLimit(200);
webSocketStompClient.setMessageConverter(new StringMessageConverter());
return webSocketStompClient;
}
@Bean
public StompSessionManager stompSessionManager(WebSocketStompClient stompClient) {
WebSocketStompSessionManager webSocketStompSessionManager =
new WebSocketStompSessionManager(stompClient, server().getWsBaseUrl() + "/ws");
webSocketStompSessionManager.setAutoReceipt(true);
return webSocketStompSessionManager;
}
@Bean
@ServiceActivator(inputChannel = "webSocketOutputChannel")
public MessageHandler stompMessageHandler(StompSessionManager stompSessionManager) {
return new StompMessageHandler(stompSessionManager);
}
@Bean
public PollableChannel stompEvents() {
return new QueueChannel();
}
@Bean
@SuppressWarnings("unchecked")
public ApplicationListener<ApplicationEvent> stompEventListener() {
ApplicationEventListeningMessageProducer producer = new ApplicationEventListeningMessageProducer();
producer.setEventTypes(StompIntegrationEvent.class);
producer.setOutputChannel(stompEvents());
return producer;
}
}
// WebSocket Server part
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Controller
private @interface IntegrationTestController {
}
@IntegrationTestController
static class SimpleController {
private final CountDownLatch latch = new CountDownLatch(1);
@MessageMapping(value = "/simple")
public void handle() {
this.latch.countDown();
}
}
@Configuration
@EnableWebSocketMessageBroker
@ComponentScan(
basePackageClasses = StompMessageHandlerWebSocketIntegrationTests.class,
useDefaultFilters = false,
includeFilters = @ComponentScan.Filter(IntegrationTestController.class))
static class ServerConfig extends AbstractWebSocketMessageBrokerConfigurer {
@Bean
public DefaultHandshakeHandler handshakeHandler() {
return new DefaultHandshakeHandler(new TomcatRequestUpgradeStrategy());
}
@Override
public void registerStompEndpoints(StompEndpointRegistry registry) {
registry.addEndpoint("/ws").setHandshakeHandler(handshakeHandler()).withSockJS();
}
@Override
public void configureMessageBroker(MessageBrokerRegistry configurer) {
configurer.setApplicationDestinationPrefixes("/app");
configurer.enableSimpleBroker("/topic", "/queue");
}
@Override
public void configureClientInboundChannel(ChannelRegistration registration) {
registration.setInterceptors(new ChannelInterceptorAdapter() {
private final AtomicBoolean invoked = new AtomicBoolean();
@Override
public Message<?> preSend(Message<?> message, MessageChannel channel) {
if (StompCommand.CONNECT.equals(StompHeaderAccessor.wrap(message).getCommand()) ||
this.invoked.compareAndSet(false, true)) {
return super.preSend(message, channel);
}
throw new RuntimeException("preSend intentional Exception");
}
});
}
}
}

8
spring-integration-stomp/src/test/resources/log4j.properties Normal file
View File

@@ -0,0 +1,8 @@
log4j.rootCategory=WARN, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %c{1} [%t] : %m%n
log4j.category.org.springframework.integration=WARN
log4j.category.org.springframework.integration.stomp=WARN

2
spring-integration-websocket/src/main/java/org/springframework/integration/websocket/inbound/WebSocketInboundChannelAdapter.java
View File

@@ -266,7 +266,7 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport
private boolean isActive() {
if (!this.active) {
logger.warn("MessageProducer '" + this + "'isn't started to accept WebSocket events");
logger.warn("MessageProducer '" + this + " 'isn't started to accept WebSocket events.");
}
return this.active;
}