diff --git a/build.gradle b/build.gradle index ddbb749220..7f67346e6c 100644 --- a/build.gradle +++ b/build.gradle @@ -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" } } diff --git a/settings.gradle b/settings.gradle index 1ac37962f9..3a818d2084 100644 --- a/settings.gradle +++ b/settings.gradle @@ -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}" + } +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java index 39bfb8329c..7c5a332f48 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java @@ -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"); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/converter/PassThruMessageConverter.java b/spring-integration-core/src/main/java/org/springframework/integration/support/converter/PassThruMessageConverter.java new file mode 100644 index 0000000000..b6997066b7 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/converter/PassThruMessageConverter.java @@ -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. + *

+ * 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); + } + +} diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java new file mode 100644 index 0000000000..3b9dc1852f --- /dev/null +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java @@ -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. + *

+ * The connection to the {@link StompSession} is made during {@link #afterPropertiesSet()}. + *

+ * The {@link #destroy()} lifecycle method manages {@link StompSession#disconnect()}. + *

+ * 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 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() { + + @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() { + + @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 doConnect(StompSessionHandler handler); + + + private static class CompositeStompSessionHandler extends StompSessionHandlerAdapter { + + private final List delegates = new ArrayList(); + + 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); + } + } + + } + +} diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/Reactor2TcpStompSessionManager.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/Reactor2TcpStompSessionManager.java new file mode 100644 index 0000000000..593a5d7f7d --- /dev/null +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/Reactor2TcpStompSessionManager.java @@ -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 doConnect(StompSessionHandler handler) { + return ((Reactor2TcpStompClient) this.stompClient).connect(getConnectHeaders(), handler); + } + +} diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/StompSessionManager.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/StompSessionManager.java new file mode 100644 index 0000000000..db5b355b3d --- /dev/null +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/StompSessionManager.java @@ -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(); + +} diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/WebSocketStompSessionManager.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/WebSocketStompSessionManager.java new file mode 100644 index 0000000000..db6aa5d8dd --- /dev/null +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/WebSocketStompSessionManager.java @@ -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 doConnect(StompSessionHandler handler) { + return ((WebSocketStompClient) this.stompClient).connect(this.url, handler, this.handshakeHeaders, + getConnectHeaders(), this.uriVariables); + } + +} diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/config/StompNamespaceHandler.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/config/StompNamespaceHandler.java new file mode 100644 index 0000000000..e783a8efd6 --- /dev/null +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/config/StompNamespaceHandler.java @@ -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() { + + } + +} diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/config/package-info.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/config/package-info.java new file mode 100644 index 0000000000..3243c8b99f --- /dev/null +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/config/package-info.java @@ -0,0 +1,4 @@ +/** + * Contains parser classes for the STOMP namespace support. + */ +package org.springframework.integration.stomp.config; diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompExceptionEvent.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompExceptionEvent.java new file mode 100644 index 0000000000..ec4719309e --- /dev/null +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompExceptionEvent.java @@ -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); + } + +} diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompIntegrationEvent.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompIntegrationEvent.java new file mode 100644 index 0000000000..7554d20860 --- /dev/null +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompIntegrationEvent.java @@ -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); + } +} diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompReceiptEvent.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompReceiptEvent.java new file mode 100644 index 0000000000..8cea26a37c --- /dev/null +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompReceiptEvent.java @@ -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; + } +} diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java new file mode 100644 index 0000000000..3f708f865b --- /dev/null +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java @@ -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}. + *

+ * Destinations can be added and removed at runtime. + *

+ * 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 destinations = new LinkedHashSet(); + + private final StompSessionManager stompSessionManager; + + private final Map subscriptions = + new HashMap(); + + private final Lock destinationLock = new ReentrantLock(); + + private ApplicationEventPublisher applicationEventPublisher; + + private volatile StompSession stompSession; + + private volatile Class payloadType = String.class; + + private volatile HeaderMapper 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 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 failedMessage = MessageBuilder.createMessage(payload, + headerAccessor.getMessageHeaders()); + getMessagingTemplate().send(errorChannel, + new ErrorMessage(new MessageHandlingException(failedMessage, exception))); + } + else { + logger.error("STOMP Frame handling error.", exception); + } + } + + } + +} diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/package-info.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/package-info.java new file mode 100644 index 0000000000..3d2106b625 --- /dev/null +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides classes which represent inbound STOMP components. + */ +package org.springframework.integration.stomp.inbound; diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/outbound/StompMessageHandler.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/outbound/StompMessageHandler.java new file mode 100644 index 0000000000..60ac124238 --- /dev/null +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/outbound/StompMessageHandler.java @@ -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 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(destination); + } + + public void setDestinationExpression(Expression destinationExpression) { + Assert.notNull(destinationExpression, "'destinationExpression' must not be null."); + this.destinationExpression = destinationExpression; + } + + public void setHeaderMapper(HeaderMapper 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)); + } + } + } + + } + +} diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/outbound/package-info.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/outbound/package-info.java new file mode 100644 index 0000000000..88566a6aed --- /dev/null +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/outbound/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides classes which represent outbound STOMP components. + */ +package org.springframework.integration.stomp.outbound; diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/package-info.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/package-info.java new file mode 100644 index 0000000000..586c163c08 --- /dev/null +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides core classes STOMP components. + */ +package org.springframework.integration.stomp; diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/support/IntegrationStompHeaders.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/support/IntegrationStompHeaders.java new file mode 100644 index 0000000000..b6ddf01e81 --- /dev/null +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/support/IntegrationStompHeaders.java @@ -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 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)); + +} diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/support/StompHeaderMapper.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/support/StompHeaderMapper.java new file mode 100644 index 0000000000..bd10a5ff41 --- /dev/null +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/support/StompHeaderMapper.java @@ -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 { + + 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 STOMP_INBOUND_HEADER_NAMES_LIST = + Arrays.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 STOMP_OUTBOUND_HEADER_NAMES_LIST = + Arrays.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 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 multiValueMap = + headers.get(StompHeaderAccessor.NATIVE_HEADERS, MultiValueMap.class); + for (Map.Entry> 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 toHeaders(StompHeaders source) { + Map target = new HashMap(); + 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; + } + +} diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/support/package-info.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/support/package-info.java new file mode 100644 index 0000000000..9c357804dc --- /dev/null +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/support/package-info.java @@ -0,0 +1,7 @@ +/** + * Provides classes to support STOMP components. + + * @author Artem Bilan + * @since 4.2 + */ +package org.springframework.integration.stomp.support; diff --git a/spring-integration-stomp/src/main/resources/META-INF/spring.handlers b/spring-integration-stomp/src/main/resources/META-INF/spring.handlers new file mode 100644 index 0000000000..a3ce0c2dcf --- /dev/null +++ b/spring-integration-stomp/src/main/resources/META-INF/spring.handlers @@ -0,0 +1 @@ +http\://www.springframework.org/schema/integration/stomp=org.springframework.integration.stomp.config.StompNamespaceHandler diff --git a/spring-integration-stomp/src/main/resources/META-INF/spring.schemas b/spring-integration-stomp/src/main/resources/META-INF/spring.schemas new file mode 100644 index 0000000000..fb6d6a33a8 --- /dev/null +++ b/spring-integration-stomp/src/main/resources/META-INF/spring.schemas @@ -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 diff --git a/spring-integration-stomp/src/main/resources/META-INF/spring.tooling b/spring-integration-stomp/src/main/resources/META-INF/spring.tooling new file mode 100644 index 0000000000..e5bf8837a9 --- /dev/null +++ b/spring-integration-stomp/src/main/resources/META-INF/spring.tooling @@ -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 diff --git a/spring-integration-stomp/src/main/resources/org/springframework/integration/stomp/config/spring-integration-stomp-4.2.xsd b/spring-integration-stomp/src/main/resources/org/springframework/integration/stomp/config/spring-integration-stomp-4.2.xsd new file mode 100644 index 0000000000..ac68551303 --- /dev/null +++ b/spring-integration-stomp/src/main/resources/org/springframework/integration/stomp/config/spring-integration-stomp-4.2.xsd @@ -0,0 +1,23 @@ + + + + + + + + + + + + + diff --git a/spring-integration-stomp/src/main/resources/org/springframework/integration/stomp/config/spring-integration-stomp.gif b/spring-integration-stomp/src/main/resources/org/springframework/integration/stomp/config/spring-integration-stomp.gif new file mode 100644 index 0000000000..750667e608 Binary files /dev/null and b/spring-integration-stomp/src/main/resources/org/springframework/integration/stomp/config/spring-integration-stomp.gif differ diff --git a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java new file mode 100644 index 0000000000..01e822d250 --- /dev/null +++ b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java @@ -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("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("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("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("???".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; + } + + } + +} diff --git a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapterWebSocketIntegrationTests.java b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapterWebSocketIntegrationTests.java new file mode 100644 index 0000000000..2408df3a8f --- /dev/null +++ b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapterWebSocketIntegrationTests.java @@ -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 payload = (Map) 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.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 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 webSocketEventListener( + final AbstractSubscribableChannel clientOutboundChannel) { + return new ApplicationListener() { + + @Override + public void onApplicationEvent(SessionSubscribeEvent event) { + Message 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())); + } + } + + }; + } + + } + +} diff --git a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/outbound/StompMessageHandlerWebSocketIntegrationTests.java b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/outbound/StompMessageHandlerWebSocketIntegrationTests.java new file mode 100644 index 0000000000..b5b5606837 --- /dev/null +++ b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/outbound/StompMessageHandlerWebSocketIntegrationTests.java @@ -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 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.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 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"); + } + + }); + } + + } + +} diff --git a/spring-integration-stomp/src/test/resources/log4j.properties b/spring-integration-stomp/src/test/resources/log4j.properties new file mode 100644 index 0000000000..5ac3c22110 --- /dev/null +++ b/spring-integration-stomp/src/test/resources/log4j.properties @@ -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 diff --git a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/inbound/WebSocketInboundChannelAdapter.java b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/inbound/WebSocketInboundChannelAdapter.java index 2478119372..20bc4d0eec 100644 --- a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/inbound/WebSocketInboundChannelAdapter.java +++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/inbound/WebSocketInboundChannelAdapter.java @@ -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; }