diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpInboundEndpointParser.java b/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpInboundEndpointParser.java index a960fbc176..bba7fb3b9a 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpInboundEndpointParser.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpInboundEndpointParser.java @@ -154,7 +154,7 @@ public class HttpInboundEndpointParser extends AbstractSingleBeanDefinitionParse if (StringUtils.hasText(headerMapper)) { if (hasMappedRequestHeaders || hasMappedResponseHeaders) { - parserContext.getReaderContext().error("Neither 'mappped-request-headers' or 'mapped-response-headers' " + + parserContext.getReaderContext().error("Neither 'mapped-request-headers' or 'mapped-response-headers' " + "attributes are allowed when a 'header-mapper' has been specified.", parserContext.extractSource(element)); } builder.addPropertyReference("headerMapper", headerMapper); 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 index 3b9dc1852f..4ec268bff1 100644 --- 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 @@ -17,6 +17,7 @@ package org.springframework.integration.stomp; import java.util.ArrayList; +import java.util.Collections; import java.util.List; import org.apache.commons.logging.Log; @@ -178,7 +179,8 @@ public abstract class AbstractStompSessionManager implements StompSessionManager private static class CompositeStompSessionHandler extends StompSessionHandlerAdapter { - private final List delegates = new ArrayList(); + private final List delegates = + Collections.synchronizedList(new ArrayList()); private volatile StompSession session; @@ -188,41 +190,53 @@ public abstract class AbstractStompSessionManager implements StompSessionManager if (this.session != null) { delegate.afterConnected(this.session, this.connectedHeaders); } - this.delegates.add(delegate); + synchronized (this.delegates) { + this.delegates.add(delegate); + } } void removeHandler(StompSessionHandler delegate) { - this.delegates.remove(delegate); + synchronized (this.delegates) { + 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); + synchronized (this.delegates) { + 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); + synchronized (this.delegates) { + 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); + synchronized (this.delegates) { + 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); + synchronized (this.delegates) { + for (StompSessionHandler delegate : this.delegates) { + delegate.handleFrame(headers, payload); + } } } diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/config/StompAdapterParserUtils.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/config/StompAdapterParserUtils.java new file mode 100644 index 0000000000..b845eefccb --- /dev/null +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/config/StompAdapterParserUtils.java @@ -0,0 +1,66 @@ +/* + * 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.w3c.dom.Element; + +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.stomp.support.StompHeaderMapper; +import org.springframework.util.StringUtils; + +/** + * @author Artem Bilan + * @since 4.2 + */ +abstract class StompAdapterParserUtils { + + static void configureStompAdapter(BeanDefinitionBuilder builder, ParserContext parserContext, Element element) { + String stompSessionManager = element.getAttribute("stomp-session-manager"); + if (!StringUtils.hasText(stompSessionManager)) { + parserContext.getReaderContext().error("The 'stomp-session-manager' is required", element); + } + builder.addConstructorArgReference(stompSessionManager); + + String headerMapper = element.getAttribute("header-mapper"); + + String mappedHeaders = element.getAttribute("mapped-headers"); + + boolean hasMappedHeaders = StringUtils.hasText(mappedHeaders); + + if (StringUtils.hasText(headerMapper)) { + if (hasMappedHeaders) { + parserContext.getReaderContext().error("The 'mapped-headers' " + + "attribute is not allowed when a 'header-mapper' has been specified.", + parserContext.extractSource(element)); + } + builder.addPropertyReference("headerMapper", headerMapper); + } + else if (hasMappedHeaders) { + BeanDefinitionBuilder headerMapperBuilder = + BeanDefinitionBuilder.genericBeanDefinition(StompHeaderMapper.class); + + // This is tricky a bit, but from one side the 'headerMapper' is an internal instance + // and isn't accessible outside and from other side allow us to avoid extra 'boolean' variable + headerMapperBuilder.addPropertyValue("inboundHeaderNames", mappedHeaders); + headerMapperBuilder.addPropertyValue("outboundHeaderNames", mappedHeaders); + + builder.addPropertyValue("headerMapper", headerMapperBuilder.getBeanDefinition()); + } + } + +} diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/config/StompInboundChannelAdapterParser.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/config/StompInboundChannelAdapterParser.java new file mode 100644 index 0000000000..fc7c27bdab --- /dev/null +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/config/StompInboundChannelAdapterParser.java @@ -0,0 +1,50 @@ +/* + * 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.w3c.dom.Element; + +import org.springframework.beans.factory.support.AbstractBeanDefinition; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.AbstractChannelAdapterParser; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.integration.stomp.inbound.StompInboundChannelAdapter; + +/** + * The {@link AbstractSingleBeanDefinitionParser} implementation for + * the {@code } element. + * + * @author Artem Bilan + * @since 4.2 + */ +public class StompInboundChannelAdapterParser extends AbstractChannelAdapterParser { + + @Override + protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) { + BeanDefinitionBuilder builder = BeanDefinitionBuilder.rootBeanDefinition(StompInboundChannelAdapter.class); + StompAdapterParserUtils.configureStompAdapter(builder, parserContext, element); + builder.addConstructorArgValue(element.getAttribute("destinations")); + builder.addPropertyReference("outputChannel", channelName); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "payload-type"); + return builder.getBeanDefinition(); + } + +} 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 index e783a8efd6..506436e08c 100644 --- 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 @@ -25,7 +25,8 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa public class StompNamespaceHandler extends AbstractIntegrationNamespaceHandler { public void init() { - + registerBeanDefinitionParser("inbound-channel-adapter", new StompInboundChannelAdapterParser()); + registerBeanDefinitionParser("outbound-channel-adapter", new StompOutboundChannelAdapterParser()); } } diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/config/StompOutboundChannelAdapterParser.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/config/StompOutboundChannelAdapterParser.java new file mode 100644 index 0000000000..58f260f5cf --- /dev/null +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/config/StompOutboundChannelAdapterParser.java @@ -0,0 +1,52 @@ +/* + * 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.w3c.dom.Element; + +import org.springframework.beans.factory.config.BeanDefinition; +import org.springframework.beans.factory.support.AbstractBeanDefinition; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.integration.stomp.outbound.StompMessageHandler; + +/** + * The {@link AbstractOutboundChannelAdapterParser} implementation for + * the {@code } element. + * + * @author Artem Bilan + * @since 4.2 + */ +public class StompOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser { + + @Override + protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) { + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(StompMessageHandler.class); + StompAdapterParserUtils.configureStompAdapter(builder, parserContext, element); + BeanDefinition expressionDef = + IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("destination", + "destination-expression", parserContext, element, false); + if (expressionDef != null) { + builder.addPropertyValue("destinationExpression", expressionDef); + } + return builder.getBeanDefinition(); + + } + +} 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 index 3f708f865b..9f3b0b0aa6 100644 --- 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 @@ -292,6 +292,11 @@ public class StompInboundChannelAdapter extends MessageProducerSupport implement } } + @Override + public void handleTransportError(StompSession session, Throwable exception) { + logger.error("STOMP transport error for session: [" + session + "]", exception); + } + } } 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 index 8287191412..a2a8537c0e 100644 --- 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 @@ -119,7 +119,7 @@ public class StompMessageHandler extends AbstractMessageHandler implements Appli } final StompSession.Receiptable receiptable = this.stompSession.send(stompHeaders, message.getPayload()); - if (this.stompSessionManager.isAutoReceiptEnabled()) { + if (receiptable.getReceiptId() != null) { final String destination = stompHeaders.getDestination(); if (this.applicationEventPublisher != null) { receiptable.addReceiptTask(new Runnable() { @@ -199,6 +199,11 @@ public class StompMessageHandler extends AbstractMessageHandler implements Appli } } + @Override + public void handleTransportError(StompSession session, Throwable exception) { + logger.error("STOMP transport error for session: [" + session + "]", exception); + } + } } 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 index ac68551303..af517fd77b 100644 --- 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 @@ -1,10 +1,9 @@ - @@ -19,5 +18,152 @@ ]]> + + + + + Configures an endpoint that will receive STOMP Messages using the provided + 'StompSessionManager'. + + + + + + + + + Comma-separated list of STOMP destination names to subscribe. + The list of destinations (and therefore subscriptions) can be modified at runtime + through the 'addDestination()' and 'removeDestination()' '@ManagedOperation's. + + + + + + + Message Channel to which error Messages should be sent. + + + + + + + + + + + + Maximum amount of time in milliseconds to wait when sending a message + to the channel if such channel may block. + For example, a Queue Channel can block until space is available + if its maximum capacity has been reached. + + + + + + + Fully qualified name of the java type for the target 'payload' + to convert from the incoming Stomp Message. + Defaults to 'java.lang.String'. + + + + + + + + + + + + Configures an endpoint that will send a STOMP Message to the provided + 'StompSessionManager'. + + + + + + + + + + + + + Name of the destination to which STOMP Messages will be sent. + Mutually exclusive with the 'destination-expression'. + + + + + + + A SpEL expression to be evaluated at runtime against each Spring Integration Message as + the root object. + Mutually exclusive with the 'destination'. + + + + + + + + + + + + Base type for the 'inbound-channel-adapter' and 'outbound-channel-adapter' elements. + + + + + + The reference to the 'StompSessionManager' bean, which encapsulates the low-level + connection and StompSession handling operations. Required. + + + + + + + + + + + + Reference to a bean implementing 'HeaderMapper' that maps Spring Integration MessageHeaders to/from + STOMP frame headers. + This is mutually exclusive with 'mapped-headers'. + Defaults to StompHeaderMapper. + + + + + + + + + + + + Comma-separated list of names of STOMP Headers to be mapped from/to the STOMP frame headers. + This can only be provided if the 'header-mapper' reference is not being set directly. + The values in this list can also be simple patterns to be matched against + the header names (e.g. "foo*" or "*foo"). + Special tokens 'STOMP_INBOUND_HEADERS' and 'STOMP_OUTBOUND_HEADERS' represent + all the standard STOMP headers (content-length, receipt, heart-beat etc) + for the inbound and outbound channel adapters respectively; + they are included by default. + If you wish to add your own headers, you must also include these tokens if you wish the + standard headers to also be mapped or provide your own 'HeaderMapper' + implementation using 'header-mapper'. + + + + + 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 index 3614b2d896..e90af3e2b5 100644 --- 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 @@ -71,7 +71,6 @@ public class StompServerIntegrationTests { 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); diff --git a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/config/StompAdaptersParserTests-context.xml b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/config/StompAdaptersParserTests-context.xml new file mode 100644 index 0000000000..835d1c3529 --- /dev/null +++ b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/config/StompAdaptersParserTests-context.xml @@ -0,0 +1,67 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/config/StompAdaptersParserTests.java b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/config/StompAdaptersParserTests.java new file mode 100644 index 0000000000..238fabcc8b --- /dev/null +++ b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/config/StompAdaptersParserTests.java @@ -0,0 +1,171 @@ +/* + * 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 static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + +import java.util.Collections; +import java.util.List; + +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.context.SmartLifecycle; +import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.integration.mapping.HeaderMapper; +import org.springframework.integration.stomp.StompSessionManager; +import org.springframework.integration.stomp.inbound.StompInboundChannelAdapter; +import org.springframework.integration.support.SmartLifecycleRoleController; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.util.MultiValueMap; + +/** + * @author Artem Bilan + * @since 4.2 + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class StompAdaptersParserTests { + + @Autowired + private StompSessionManager stompSessionManager; + + @Autowired + private HeaderMapper headerMapper; + + @Autowired + @Qualifier("defaultInboundAdapter") + private MessageChannel defaultInboundAdapterChannel; + + @Autowired + private MessageChannel errorChannel; + + @Autowired + private MessageChannel inboundChannel; + @Autowired + @Qualifier("defaultInboundAdapter.adapter") + private StompInboundChannelAdapter defaultInboundAdapter; + + @Autowired + private StompInboundChannelAdapter customInboundAdapter; + + @Autowired + @Qualifier("defaultOutboundAdapter") + private MessageChannel defaultOutboundAdapterChannel; + + @Autowired + @Qualifier("defaultOutboundAdapter.handler") + private MessageHandler defaultOutboundAdapterHandler; + + @Autowired + @Qualifier("defaultOutboundAdapter.adapter") + private AbstractEndpoint defaultOutboundAdapter; + + @Autowired + private MessageChannel outboundChannel; + + @Autowired + @Qualifier("customOutboundAdapter.handler") + private MessageHandler customOutboundAdapterHandler; + + @Autowired + @Qualifier("customOutboundAdapter") + private AbstractEndpoint customOutboundAdapter; + + @Autowired + private SmartLifecycleRoleController roleController; + + @Test + public void testParsers() { + assertSame(this.defaultInboundAdapterChannel, + TestUtils.getPropertyValue(this.defaultInboundAdapter, "outputChannel")); + assertSame(this.stompSessionManager, + TestUtils.getPropertyValue(this.defaultInboundAdapter, "stompSessionManager")); + assertNull(TestUtils.getPropertyValue(this.defaultInboundAdapter, "errorChannel")); + Object headerMapper = TestUtils.getPropertyValue(this.defaultInboundAdapter, "headerMapper"); + assertNotNull(headerMapper); + assertNotSame(this.headerMapper, headerMapper); + assertEquals(String.class, TestUtils.getPropertyValue(this.defaultInboundAdapter, "payloadType", Class.class)); + assertTrue(TestUtils.getPropertyValue(this.defaultInboundAdapter, "autoStartup", Boolean.class)); + + assertSame(this.inboundChannel, + TestUtils.getPropertyValue(this.customInboundAdapter, "outputChannel")); + assertSame(this.stompSessionManager, + TestUtils.getPropertyValue(this.customInboundAdapter, "stompSessionManager")); + assertSame(this.errorChannel, TestUtils.getPropertyValue(this.customInboundAdapter, "errorChannel")); + assertEquals(Collections.singleton("foo"), + TestUtils.getPropertyValue(this.customInboundAdapter, "destinations")); + headerMapper = TestUtils.getPropertyValue(this.customInboundAdapter, "headerMapper"); + assertNotNull(headerMapper); + assertNotSame(this.headerMapper, headerMapper); + assertArrayEquals(new String[] {"bar", "foo"}, + TestUtils.getPropertyValue(headerMapper, "inboundHeaderNames", String[].class)); + assertEquals(Integer.class, TestUtils.getPropertyValue(this.customInboundAdapter, "payloadType", Class.class)); + assertFalse(TestUtils.getPropertyValue(this.customInboundAdapter, "autoStartup", Boolean.class)); + assertEquals(200, TestUtils.getPropertyValue(this.customInboundAdapter, "phase")); + assertEquals(2000L, TestUtils.getPropertyValue(this.customInboundAdapter, "messagingTemplate.sendTimeout")); + + assertSame(this.stompSessionManager, + TestUtils.getPropertyValue(this.defaultOutboundAdapterHandler, "stompSessionManager")); + headerMapper = TestUtils.getPropertyValue(this.defaultOutboundAdapterHandler, "headerMapper"); + assertNotNull(headerMapper); + assertNotSame(this.headerMapper, headerMapper); + assertNull(TestUtils.getPropertyValue(this.defaultOutboundAdapterHandler, "destinationExpression")); + assertSame(this.defaultOutboundAdapterHandler, + TestUtils.getPropertyValue(this.defaultOutboundAdapter, "handler")); + assertSame(this.defaultOutboundAdapterChannel, + TestUtils.getPropertyValue(this.defaultOutboundAdapter, "inputChannel")); + assertTrue(TestUtils.getPropertyValue(this.defaultOutboundAdapter, "autoStartup", Boolean.class)); + + assertSame(this.stompSessionManager, + TestUtils.getPropertyValue(this.customOutboundAdapterHandler, "stompSessionManager")); + assertSame(this.headerMapper, TestUtils.getPropertyValue(this.customOutboundAdapterHandler, "headerMapper")); + assertEquals("baz", + TestUtils.getPropertyValue(this.customOutboundAdapterHandler, "destinationExpression.literalValue")); + assertSame(this.customOutboundAdapterHandler, + TestUtils.getPropertyValue(this.customOutboundAdapter, "handler")); + assertSame(this.outboundChannel, TestUtils.getPropertyValue(this.customOutboundAdapter, "inputChannel")); + assertFalse(TestUtils.getPropertyValue(this.customOutboundAdapter, "autoStartup", Boolean.class)); + assertEquals(100, TestUtils.getPropertyValue(this.customOutboundAdapter, "phase")); + + @SuppressWarnings("unchecked") + MultiValueMap lifecycles = (MultiValueMap) + TestUtils.getPropertyValue(this.roleController, "lifecycles", MultiValueMap.class); + assertTrue(lifecycles.containsKey("bar")); + List bars = lifecycles.get("bar"); + bars.contains(this.customInboundAdapter); + assertTrue(lifecycles.containsKey("foo")); + List foos = lifecycles.get("bar"); + bars.contains(this.customOutboundAdapter); + } + +} diff --git a/spring-integration-stomp/src/test/resources/log4j.properties b/spring-integration-stomp/src/test/resources/log4j.properties index 5ac3c22110..b9fd67eef9 100644 --- a/spring-integration-stomp/src/test/resources/log4j.properties +++ b/spring-integration-stomp/src/test/resources/log4j.properties @@ -4,5 +4,6 @@ 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 +log4j.category.org.springframework.messaging=DEBUG +log4j.category.org.springframework.integration=DEBUG +#log4j.category.org.springframework.integration.stomp=WARN diff --git a/src/reference/asciidoc/index.adoc b/src/reference/asciidoc/index.adoc index dead119ea2..d690e8cac5 100644 --- a/src/reference/asciidoc/index.adoc +++ b/src/reference/asciidoc/index.adoc @@ -88,6 +88,8 @@ include::./rmi.adoc[] include::./sftp.adoc[] +include::./stomp.adoc[] + include::./stream.adoc[] include::./syslog.adoc[] diff --git a/src/reference/asciidoc/stomp.adoc b/src/reference/asciidoc/stomp.adoc new file mode 100644 index 0000000000..a1f0594ac0 --- /dev/null +++ b/src/reference/asciidoc/stomp.adoc @@ -0,0 +1,333 @@ +[[stomp]] +== STOMP Support + +[[stomp-introduction]] +=== Introduction + +Spring Integration _version 4.2_ introduced _STOMP Client_ support. +It is based on the architecture, infrastructure and API from the Spring Framework's _messaging_ module, _stomp_ package. +Many of Spring STOMP components (e.g. `StompSession` or `StompClientSupport`) +are used within Spring Integration. +For more information, please, refer to the http://docs.spring.io/spring/docs/current/spring-framework-reference/html/websocket.html#websocket-stomp-client[Spring Framework STOMP Support] +chapter in the Spring Framework reference manual. + +[[stomp-overview]] +=== Overview + +To configure STOMP (Simple [or Streaming] Text Orientated Messaging Protocol) let's start with the _STOMP Client_ object. +The Spring Framework provides these implementations: + +* `WebSocketStompClient` - built on the Spring WebSocket API with support for standard JSR-356 WebSocket, Jetty 9, +as well as SockJS for HTTP-based WebSocket emulation with SockJS Client. + +* `Reactor2TcpStompClient` - built on `NettyTcpClient` from the `reactor-net` project. + +Any other `StompClientSupport` implementation can be provided. +See the JavaDocs of those classes for more information. + +The `StompClientSupport` class is designed as a _factory_ to produce a `StompSession` for the provided +`StompSessionHandler` and all the remaining work is done through the _callbacks_ to that `StompSessionHandler` +and `StompSession` abstraction. +With the Spring Integration _adapter_ abstraction, we +need to provide some managed shared object to represent our application as a STOMP client with its unique session. +For this purpose, Spring Integration provides the `StompSessionManager` abstraction to manage the _single_ +`StompSession` between any provided `StompSessionHandler`. +This allows the use of _inbound_ or _outbound_ channel adapters (or both) for the particular STOMP Broker. +See `StompSessionManager` (and its implementations) JavaDocs for more information. + +[[stomp-inbound-adapter]] +=== STOMP Inbound Channel Adapter + +The `StompInboundChannelAdapter` is a one-stop `MessageProducer` component to subscribe our Spring Integration +application to the provided STOMP destinations and receive messages from them, converted from the STOMP +frames using the provided `MessageConverter` on the connected `StompSession`. +The destinations (and therefore STOMP subscriptions) can be changed at runtime using appropriate `@ManagedOperation` s +on the `StompInboundChannelAdapter`. + +For more configuration options see <> and the `StompInboundChannelAdapter` JavaDocs. + +[[stomp-outbound-adapter]] +=== STOMP Outbound Channel Adapter + +The `StompMessageHandler` is the `MessageHandler` for the `` +to send the outgoing `Message` s to the STOMP `destination` (pre-configured or determined at runtime via a SpEL expression) STOMP through the `StompSession`, provided by the shared `StompSessionManager`. + +For more configuration option see <> and the `StompMessageHandler` JavaDocs. + +[[stomp-headers]] +=== STOMP Headers Mapping + +The STOMP protocol provides _headers_ as part of frame; the entire structure of the STOMP frame +has this format: + +.... +COMMAND +header1:value1 +header2:value2 + +Body^@ +.... + +Spring Framework provides `StompHeaders`, to represent these headers. +See the JavaDocs for more details. +STOMP frames are converted to/from `Message` and these headers are mapped to/from `MessageHeaders`. +Spring Integration provides a default `HeaderMapper` implementation for the STOMP adapters. +The implementation is `StompHeaderMapper` which provides `fromHeaders()` and `toHeaders()` operations for the +_inbound_ and _outbound_ adapters respectively. + +As with many other Spring Integration modules, the `IntegrationStompHeaders` class has been +introduced to map standard STOMP headers to `MessageHeaders` with `stomp_` as the header name prefix. +In addition, all `MessageHeaders` with that prefix are mapped to the `StompHeaders` when sending to a destination. + +For more information, see the JavaDocs of those classes and the `mapped-headers` attribute description in the +<>. + +[[stomp-events]] +=== STOMP Integration Events + +Many STOMP operations are asynchronous, including error handling. +For example, STOMP has a `RECEIPT` server frame that is returned when a client frame has requested one by addding +the `RECEIPT` header. +To provide access to these asynchronous events, Spring Integration emits `StompIntegrationEvent` s which can be +obtained by implementing an `ApplicationListener` or using an event inbound channel adapter. + +Specifically, a `StompExceptionEvent` is emitted from the `AbstractStompSessionManager`, when a +`stompSessionListenableFuture` receives `onFailure()` in case of failure to connect to STOMP Broker. +Another example is the `StompMessageHandler` which processes +`ERROR` STOMP frames, which are server responses to improper, unaccepted, messages sent by this `StompMessageHandler`. + +The `StompReceiptEvent` s are emitted from the `StompMessageHandler` as a part of `StompSession.Receiptable` +callbacks in the asynchronous answers for the sent messages to the `StompSession`. +The `StompReceiptEvent` can be positive and negative depending on whether or not the `RECEIPT` frame was received +from the server within the `receiptTimeLimit` period, which can be configured on the `StompClientSupport` instance. +Defaults to `15 * 1000`. + +NOTE: The `StompSession.Receiptable` callbacks are added only if the `RECEIPT` STOMP header of the message to send +is not `null`. +Automatic `RECEIPT` header generation can be enabled on the `StompSession` through its `autoReceipt` option and +on the `StompSessionManager` respectively. + +See the next paragraph for more information how to configure Spring Integration to accept those `ApplicationEvent` s. + +[[stomp-java-config]] +=== STOMP Adapters Java Configuration + +A comprehensive Java & Annotation Configuration for STOMP Adapters may look like this: + +[source,java] +---- +@Configuration +@EnableIntegration +public class StompConfiguration { + + @Bean + public Reactor2TcpStompClient stompClient() { + Reactor2TcpStompClient stompClient = new Reactor2TcpStompClient("127.0.0.1", 61613); + stompClient.setMessageConverter(new PassThruMessageConverter()); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.afterPropertiesSet(); + stompClient.setTaskScheduler(taskScheduler); + stompClient.setReceiptTimeLimit(5000); + return stompClient; + } + + @Bean + public StompSessionManager stompSessionManager() { + Reactor2TcpStompSessionManager stompSessionManager = new Reactor2TcpStompSessionManager(stompClient()); + stompSessionManager.setAutoReceipt(true); + return stompSessionManager; + } + + @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; + } + + @Bean + public PollableChannel stompEvents() { + return new QueueChannel(); + } + + @Bean + public ApplicationListener stompEventListener() { + ApplicationEventListeningMessageProducer producer = new ApplicationEventListeningMessageProducer(); + producer.setEventTypes(StompIntegrationEvent.class); + producer.setOutputChannel(stompEvents()); + return producer; + } + +} +---- + +[[stomp-namespace]] +=== STOMP Namespace Support + +Spring Integration _STOMP_ namespace implements the _inbound_ and _outbound_ channel adapter components described below. +To include it in your configuration, simply provide the following namespace declaration in your application context +configuration file: + +[source,xml] +---- + + + ... + +---- + +** + +[source,xml] +---- + + channel="" <2> + stomp-session-manager="" <3> + header-mapper="" <4> + mapped-headers="" <5> + destination="" <6> + destination-expression="" <7> + auto-startup="" <8> + phase=""/> <9> +---- + + + +<1> The component bean name. +The `MessageHandler` is registered with the bean alias `id + '.handler'`. +If the `channel` attribute isn't provided, a `DirectChannel` is created and registered with the application context +with this `id` attribute as the bean name. +In this case, the endpoint is registered with the bean name `id + '.adapter'`. + + +<2> Identifies the channel attached to this adapter. +_Optional_ - if `id` is present - see `id`. + + +<3> Reference to a `StompSessionManager` bean, which encapsulates the low-level connection and `StompSession` +handling operations. +_Required_. + + +<4> Reference to a bean implementing `HeaderMapper` that maps Spring Integration MessageHeaders to/from +STOMP frame headers. +This is mutually exclusive with `mapped-headers`. +Defaults to `StompHeaderMapper`. + + +<5> Comma-separated list of names of STOMP Headers to be mapped to the STOMP frame headers. +This can only be provided if the `header-mapper` reference is not set. +The values in this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo"). +A special token `STOMP_OUTBOUND_HEADERS` represents all the standard STOMP headers +(content-length, receipt, heart-beat etc); they are included by default. +If you wish to add your own headers, you must also include this token if you wish the standard headers to also be +mapped or provide your own `HeaderMapper` implementation using `header-mapper`. + + +<6> Name of the destination to which STOMP Messages will be sent. +Mutually exclusive with the `destination-expression`. + + +<7> A SpEL expression to be evaluated at runtime against each Spring Integration `Message` as the root object. +Mutually exclusive with the `destination`. + + +<8> Boolean value indicating whether this endpoint should start automatically. +Default to `true`. + + +<9> The lifecycle phase within which this endpoint should start and stop. +The lower the value the earlier this endpoint will start and the later it will stop. +The default is `Integer.MIN_VALUE`. +Values can be negative. +See `SmartLifeCycle`. + +** + +[source,xml] +---- + + channel="" <2> + error-channel="" <3> + stomp-session-manager="" <4> + header-mapper="" <5> + mapped-headers="" <6> + destinations="" <7> + send-timeout="" <8> + payload-type="" <9> + auto-startup="" <10> + phase=""/> <11> +---- + + + +<1> The component bean name. +If the `channel` attribute isn't provided, a `DirectChannel` is created and registered with the application context +with this `id` attribute as the bean name. +In this case, the endpoint is registered with the bean name `id + '.adapter'`. + + +<2> Identifies the channel attached to this adapter. + + +<3> The `MessageChannel` bean reference to which the `ErrorMessages` should be sent. + + +<4> See the same option on the ``. + + +<5> Comma-separated list of names of STOMP Headers to be mapped from the STOMP frame headers. +This can only be provided if the `header-mapper` reference is not set. +The values in this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo"). +A special token `STOMP_INBOUND_HEADERS` represents all the standard STOMP headers +(content-length, receipt, heart-beat etc); they are included by default. +If you wish to add your own headers, you must also include this token if you wish the standard headers to also be +mapped or provide your own `HeaderMapper` implementation using `header-mapper`. + + +<6> See the same option on the ``. + + +<7> Comma-separated list of STOMP destination names to subscribe. +The list of destinations (and therefore subscriptions) can be modified at runtime +through the `addDestination() and `removeDestination()` `@ManagedOperation` s. + + +<8> Maximum amount of time in milliseconds to wait when sending a message to the channel if the channel may block. +For example, a `QueueChannel` can block until space is available if its maximum capacity has been reached. + + +<9> Fully qualified name of the java type for the target `payload` to convert from the incoming STOMP Frame. +Default to `String.class`. + + +<10> See the same option on the ``. + + +<11> See the same option on the ``. diff --git a/src/reference/asciidoc/web-sockets.adoc b/src/reference/asciidoc/web-sockets.adoc index 549f3207e7..0180761b32 100644 --- a/src/reference/asciidoc/web-sockets.adoc +++ b/src/reference/asciidoc/web-sockets.adoc @@ -88,7 +88,7 @@ If `useBroker = false` and received message is of `SimpMessageType.CONNECT` type NOTE: Spring's WebSocket Support allows the configuration of only one Broker Relay, hence we don't require an `AbstractBrokerMessageHandler` reference, it is detected in the Application Context. -For more configuration option see <>. +For more configuration options see <>. [[web-socket-outbound-adapter]] === WebSocket Outbound Channel Adapter @@ -100,7 +100,7 @@ On the client side, the `WebSocketSession` `id` message header isn't required, b To use the STOMP sub-protocol, this adapter should be configured with a `StompSubProtocolHandler`. Then you can send any STOMP message type to this adapter, using `StompHeaderAccessor.create(StompCommand...)` and a `MessageBuilder`, or just using a `HeaderEnricher` (see <>). -For more configuration option see below. +For more configuration options see below. [[web-sockets-namespace]] === WebSockets Namespace Support @@ -310,9 +310,10 @@ By default `Jackson2SockJsMessageCodec` is used requiring the Jackson library to <1> The component bean name. -If the `channel` attribute isn't provided, a `DirectChannel` is created and registered with the application context with this `id` attribute as the bean name. +If the `channel` attribute isn't provided, a `DirectChannel` is created and registered with the application context +with this `id` attribute as the bean name. In this case, the endpoint is registered with the bean name `id + '.adapter'`. -And the `MessageHandler` is registered with the bean alias `id +'.adapter'`. +And the `MessageHandler` is registered with the bean alias `id + '.handler'`. <2> Identifies the channel attached to this adapter. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index f347eb7420..e1bb8f120b 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -62,6 +62,12 @@ occurs. See <> for more information. +[[x4.2-stomp]] +==== STOMP Support + +STOMP support has been added to the framework as _inbound_ and _outbound_ channel adapters pair. +See <> for more information. + [[x4.2-general]] === General Changes