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

INT-3527: WebSocket and 'o-c-a' Improvements

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

* `WebSocketInboundChannelAdapter` now handles `CONNECT` STOMP message and sends `CONNECT_ACK` message to the `WebSocketSession` immediately
* `ExpressionMessageProducerSupport` implementations now checks the result of `expression` and if it is a `Message<?>` it is sent to channel without creating a new one
which previously wrapped that `Message<?>` as the `payload`.
* Add `<script>` support to the `<outbound-channel-adapter>`
* Upgrade to the SF 4.1.1
* Add appropriate notes to the Docs
This commit is contained in:
Artem Bilan
2014-10-13 17:26:45 +03:00
committed by Gary Russell
parent d5f1bb6516
commit 252c53db4c
13 changed files with 138 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -118,7 +118,7 @@ subprojects { subproject ->
springSecurityVersion = '3.2.5.RELEASE'
springSocialTwitterVersion = '1.1.0.RELEASE'
springRetryVersion = '1.1.1.RELEASE'
springVersion = project.hasProperty('springVersion') ? project.springVersion : '4.1.0.RELEASE'
springVersion = project.hasProperty('springVersion') ? project.springVersion : '4.1.1.RELEASE'
springWsVersion = '2.2.0.RELEASE'
xmlUnitVersion = '1.5'
xstreamVersion = '1.4.7'

29
spring-integration-core/src/main/java/org/springframework/integration/config/xml/DefaultOutboundChannelAdapterParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2014 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.
@@ -18,6 +18,7 @@ package org.springframework.integration.config.xml;
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.support.RootBeanDefinition;
@@ -27,6 +28,7 @@ import org.springframework.integration.config.ExpressionFactoryBean;
import org.springframework.integration.handler.ExpressionEvaluatingMessageHandler;
import org.springframework.integration.handler.MethodInvokingMessageHandler;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
/**
* Parser for the &lt;outbound-channel-adapter/&gt; element.
@@ -39,20 +41,32 @@ public class DefaultOutboundChannelAdapterParser extends AbstractOutboundChannel
@Override
protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
BeanComponentDefinition innerConsumerDefinition = IntegrationNamespaceUtils.parseInnerHandlerDefinition(element, parserContext);
Object source = parserContext.extractSource(element);
BeanComponentDefinition innerConsumerDefinition =
IntegrationNamespaceUtils.parseInnerHandlerDefinition(element, parserContext);
String consumerRef = element.getAttribute(IntegrationNamespaceUtils.REF_ATTRIBUTE);
String methodName = element.getAttribute(IntegrationNamespaceUtils.METHOD_ATTRIBUTE);
String consumerExpressionString = element.getAttribute(IntegrationNamespaceUtils.EXPRESSION_ATTRIBUTE);
Element scriptElement = DomUtils.getChildElementByTagName(element, "script");
boolean isInnerConsumer = innerConsumerDefinition != null;
boolean isRef = StringUtils.hasText(consumerRef);
boolean isExpression = StringUtils.hasText(consumerExpressionString);
boolean hasMethod = StringUtils.hasText(methodName);
boolean hasScript = scriptElement != null;
if (!(isInnerConsumer ^ (isRef ^ isExpression))) {
if (!isInnerConsumer & !isRef & !isExpression & !hasScript) {
parserContext.getReaderContext().error(
"Exactly one of the 'ref', 'expression' or inner bean is required.", element);
"Exactly one of the 'ref', 'expression', <script> or inner bean is required.", source);
}
if (hasScript) {
if (isRef | isExpression) {
parserContext.getReaderContext().error(
"Neither 'ref' nor 'expression' are permitted when an inner script element is configured.",
source);
}
}
if (hasMethod & isExpression) {
@@ -68,6 +82,13 @@ public class DefaultOutboundChannelAdapterParser extends AbstractOutboundChannel
expressionDef.getConstructorArgumentValues().addGenericArgumentValue(consumerExpressionString);
consumerBuilder.addConstructorArgValue(expressionDef);
}
else if (hasScript) {
consumerBuilder = BeanDefinitionBuilder.genericBeanDefinition(MethodInvokingMessageHandler.class);
BeanDefinition scriptBeanDefinition = parserContext.getDelegate().parseCustomElement(scriptElement,
consumerBuilder.getBeanDefinition());
consumerBuilder.addConstructorArgValue(scriptBeanDefinition);
consumerBuilder.addConstructorArgValue("processMessage");
}
else {
consumerBuilder = BeanDefinitionBuilder.genericBeanDefinition(MethodInvokingMessageHandler.class);
if (isRef) {

12
spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd
View File

@@ -1152,11 +1152,11 @@
</xsd:complexType>
<xsd:complexType name="outboundChannelAdapterType">
<xsd:all>
<xsd:choice minOccurs="0" maxOccurs="3">
<xsd:element name="poller" type="basePollerType" minOccurs="0" maxOccurs="1" />
<xsd:element ref="beans:bean" minOccurs="0" maxOccurs="1" />
<xsd:element name="request-handler-advice-chain" type="handlerAdviceChainType" minOccurs="0" maxOccurs="1" />
</xsd:all>
<xsd:any namespace="##other" processContents="strict" minOccurs="0" maxOccurs="1" />
</xsd:choice>
<xsd:attributeGroup ref="channelAdapterAttributes" />
<xsd:attribute name="order">
<xsd:annotation>
@@ -1171,9 +1171,9 @@
</xsd:complexType>
<xsd:complexType name="outboundChannelAdapterTypeChain">
<xsd:all>
<xsd:element ref="beans:bean" minOccurs="0" maxOccurs="1" />
</xsd:all>
<xsd:choice minOccurs="0">
<xsd:any namespace="##other" processContents="strict" minOccurs="0" maxOccurs="1" />
</xsd:choice>
<xsd:attribute name="id" type="xsd:string" />
</xsd:complexType>

13
spring-integration-event/src/main/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducer.java
View File

@@ -63,9 +63,7 @@ public class ApplicationEventListeningMessageProducer extends ExpressionMessageP
* In addition, this method re-registers the current instance as a {@link ApplicationListener}
* with the {@link ApplicationEventMulticaster} which clears the listener cache. The cache will be
* refreshed on the next appropriate {@link ApplicationEvent}.
*
* @param eventTypes The event types.
*
* @see ApplicationEventMulticaster#addApplicationListener
* @see #supportsEventType
*/
@@ -104,8 +102,15 @@ public class ApplicationEventListeningMessageProducer extends ExpressionMessageP
this.sendMessage((Message<?>) event.getSource());
}
else {
Object payload = this.evaluatePayloadExpression(event);
this.sendMessage(this.getMessageBuilderFactory().withPayload(payload).build());
Message<?> message = null;
Object result = this.evaluatePayloadExpression(event);
if (result instanceof Message) {
message = (Message<?>) result;
}
else {
message = this.getMessageBuilderFactory().withPayload(result).build();
}
this.sendMessage(message);
}
}
}

13
spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducer.java
View File

@@ -24,6 +24,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.endpoint.ExpressionMessageProducerSupport;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
import com.gemstone.gemfire.cache.CacheClosedException;
@@ -41,6 +42,7 @@ import com.gemstone.gemfire.cache.util.CacheListenerAdapter;
*
* @author Mark Fisher
* @author David Turanski
* @author Artem Bilan
* @since 2.1
*/
@SuppressWarnings({"rawtypes", "unchecked"})
@@ -131,8 +133,15 @@ public class CacheListeningMessageProducer extends ExpressionMessageProducerSupp
}
private void publish(Object payload) {
sendMessage(CacheListeningMessageProducer.this.getMessageBuilderFactory().withPayload(payload).build());
private void publish(Object object) {
Message<?> message = null;
if (object instanceof Message) {
message = (Message<?>) object;
}
else {
message = getMessageBuilderFactory().withPayload(object).build();
}
sendMessage(message);
}
}

30
spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/ContinuousQueryMessageProducer.java
View File

@@ -20,7 +20,6 @@ import java.util.Arrays;
import java.util.HashSet;
import java.util.Set;
import com.gemstone.gemfire.cache.query.CqEvent;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -31,6 +30,8 @@ import org.springframework.integration.endpoint.ExpressionMessageProducerSupport
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
import com.gemstone.gemfire.cache.query.CqEvent;
/**
* Responds to a Gemfire continuous query (set using the #query field) that is
* constantly evaluated against a cache
@@ -39,6 +40,7 @@ import org.springframework.util.Assert;
*
* @author Josh Long
* @author David Turanski
* @author Artem Bilan
* @since 2.1
*
*/
@@ -55,12 +57,11 @@ public class ContinuousQueryMessageProducer extends ExpressionMessageProducerSup
private boolean durable;
private volatile Set<CqEventType> supportedEventTypes = new HashSet<CqEventType>(Arrays.asList(CqEventType.CREATED,
CqEventType.UPDATED));
private volatile Set<CqEventType> supportedEventTypes =
new HashSet<CqEventType>(Arrays.asList(CqEventType.CREATED, CqEventType.UPDATED));
/**
*
* @param queryListenerContainer a {@link org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer}
* @param queryListenerContainer a {@link ContinuousQueryListenerContainer}
* @param query the query string
*/
public ContinuousQueryMessageProducer(ContinuousQueryListenerContainer queryListenerContainer, String query) {
@@ -71,7 +72,6 @@ public class ContinuousQueryMessageProducer extends ExpressionMessageProducerSup
}
/**
*
* @param queryName optional query name
*/
public void setQueryName(String queryName) {
@@ -79,7 +79,6 @@ public class ContinuousQueryMessageProducer extends ExpressionMessageProducerSup
}
/**
*
* @param durable true if the query is a durable subscription
*/
public void setDurable(boolean durable) {
@@ -110,10 +109,7 @@ public class ContinuousQueryMessageProducer extends ExpressionMessageProducerSup
/*
* (non-Javadoc)
*
* @see
* org.springframework.data.gemfire.listener.QueryListener#onEvent(com.gemstone
* .gemfire.cache.query.CqEvent)
* @see org.springframework.data.gemfire.listener.QueryListener#onEvent(com.gemstone.gemfire.cache.query.CqEvent)
*/
@Override
public void onEvent(CqEvent event) {
@@ -122,9 +118,15 @@ public class ContinuousQueryMessageProducer extends ExpressionMessageProducerSup
logger.debug(String.format("processing cq event key [%s] event [%s]", event.getQueryOperation()
.toString(), event.getKey()));
}
Message<?> cqEventMessage = this.getMessageBuilderFactory().withPayload(evaluatePayloadExpression(event))
.build();
sendMessage(cqEventMessage);
Message<?> message = null;
Object object = evaluatePayloadExpression(event);
if (object instanceof Message) {
message = (Message<?>) object;
}
else {
message = getMessageBuilderFactory().withPayload(object).build();
}
sendMessage(message);
}
}

7
spring-integration-groovy/src/test/java/org/springframework/integration/groovy/config/GroovyServiceActivatorTests-context.xml
View File

@@ -61,5 +61,12 @@
</groovy:script>
</service-activator>
<beans:bean id="invoked" class="java.util.concurrent.atomic.AtomicBoolean"/>
<outbound-channel-adapter id="outboundChannelAdapterWithGroovy">
<groovy:script>
invoked.set(true)
</groovy:script>
</outbound-channel-adapter>
</beans:beans>

20
spring-integration-groovy/src/test/java/org/springframework/integration/groovy/config/GroovyServiceActivatorTests.java
View File

@@ -22,14 +22,13 @@ import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import groovy.lang.GroovyObject;
import groovy.lang.MissingPropertyException;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import org.hamcrest.Matchers;
import org.junit.Test;
@@ -38,13 +37,13 @@ import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.handler.ReplyRequiredException;
import org.springframework.integration.scripting.ScriptVariableGenerator;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.messaging.support.GenericMessage;
@@ -52,6 +51,9 @@ import org.springframework.scripting.groovy.GroovyObjectCustomizer;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import groovy.lang.GroovyObject;
import groovy.lang.MissingPropertyException;
/**
* @author Mark Fisher
* @author Oleg Zhurakousky
@@ -81,6 +83,12 @@ public class GroovyServiceActivatorTests {
@Autowired
private MyGroovyCustomizer groovyCustomizer;
@Autowired
private AtomicBoolean invoked;
@Autowired
private MessageChannel outboundChannelAdapterWithGroovy;
@Test
public void referencedScriptAndCustomiser() throws Exception{
@@ -193,6 +201,12 @@ public class GroovyServiceActivatorTests {
new ClassPathXmlApplicationContext("GroovyServiceActivatorTests-fail-withgenerator-context.xml", this.getClass());
}
@Test
public void testGroovyScriptForOutboundChannelAdapter() {
this.outboundChannelAdapterWithGroovy.send(new GenericMessage<String>("foo"));
assertTrue(this.invoked.get());
}
public static class SampleScriptVariSource implements ScriptVariableGenerator{
public Map<String, Object> generateScriptVariables(Message<?> message) {

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

@@ -92,7 +92,7 @@ public final class ClientWebSocketContainer extends IntegrationWebSocketContaine
* @return the {@link #clientSession}, if established.
*/
@Override
public WebSocketSession getSession(String sessionId) throws Exception {
public WebSocketSession getSession(String sessionId) {
if (this.isRunning()) {
try {
this.connectionLatch.await(10, TimeUnit.SECONDS);

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

@@ -109,7 +109,7 @@ public abstract class IntegrationWebSocketContainer implements DisposableBean {
return Collections.unmodifiableMap(this.sessions);
}
public WebSocketSession getSession(String sessionId) throws Exception {
public WebSocketSession getSession(String sessionId) {
WebSocketSession session = this.sessions.get(sessionId);
Assert.notNull(session, "Session not found for id '" + sessionId + "'");
return session;

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

@@ -35,6 +35,7 @@ import org.springframework.integration.websocket.support.SubProtocolHandlerRegis
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.converter.ByteArrayMessageConverter;
import org.springframework.messaging.converter.CompositeMessageConverter;
@@ -47,6 +48,7 @@ import org.springframework.messaging.simp.SimpMessageType;
import org.springframework.messaging.simp.broker.AbstractBrokerMessageHandler;
import org.springframework.messaging.simp.broker.SimpleBrokerMessageHandler;
import org.springframework.messaging.simp.stomp.StompBrokerRelayMessageHandler;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.MimeTypeUtils;
@@ -60,6 +62,8 @@ import org.springframework.web.socket.WebSocketSession;
*/
public class WebSocketInboundChannelAdapter extends MessageProducerSupport implements WebSocketListener {
private static final byte[] EMPTY_PAYLOAD = new byte[0];
private final List<MessageConverter> defaultConverters = new ArrayList<MessageConverter>(3);
{
@@ -111,7 +115,12 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport imple
@Override
public void handleMessage(Message<?> message) throws MessagingException {
handleMessageAndSend(message);
try {
handleMessageAndSend(message);
}
catch (Exception e) {
throw new MessageHandlingException(message, e);
}
}
});
@@ -249,18 +258,29 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport imple
return this.active;
}
private void handleMessageAndSend(Message<?> message) {
Object payload = this.messageConverter.fromMessage(message,
this.payloadType.get());
private void handleMessageAndSend(Message<?> message) throws Exception {
SimpMessageHeaderAccessor headerAccessor = SimpMessageHeaderAccessor.wrap(message);
SimpMessageType messageType = headerAccessor.getMessageType();
if ((messageType == null || SimpMessageType.MESSAGE.equals(messageType))
if ((messageType == null || SimpMessageType.MESSAGE.equals(messageType)
|| (SimpMessageType.CONNECT.equals(messageType) && !this.useBroker))
&& !checkDestinationPrefix(headerAccessor.getDestination())) {
headerAccessor.removeHeader(SimpMessageHeaderAccessor.NATIVE_HEADERS);
sendMessage(getMessageBuilderFactory().withPayload(payload).copyHeaders(headerAccessor.toMap()).build());
if (SimpMessageType.CONNECT.equals(messageType)) {
String sessionId = headerAccessor.getSessionId();
SimpMessageHeaderAccessor connectAck = SimpMessageHeaderAccessor.create(SimpMessageType.CONNECT_ACK);
connectAck.setSessionId(sessionId);
connectAck.setHeader(SimpMessageHeaderAccessor.CONNECT_MESSAGE_HEADER, message);
Message<byte[]> ackMessage = MessageBuilder.createMessage(EMPTY_PAYLOAD, connectAck.getMessageHeaders());
WebSocketSession session = this.webSocketContainer.getSession(sessionId);
this.subProtocolHandlerRegistry.findProtocolHandler(session).handleMessageToClient(session, ackMessage);
}
else {
headerAccessor.removeHeader(SimpMessageHeaderAccessor.NATIVE_HEADERS);
Object payload = this.messageConverter.fromMessage(message, this.payloadType.get());
sendMessage(getMessageBuilderFactory().withPayload(payload).copyHeaders(headerAccessor.toMap()).build());
}
}
else {
if (this.brokerHandler != null) {
if (this.useBroker) {
this.brokerHandler.handleMessage(message);
}
else if (logger.isDebugEnabled()) {
@@ -273,7 +293,7 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport imple
}
private boolean checkDestinationPrefix(String destination) {
if (this.brokerHandler != null) {
if (this.useBroker) {
Collection<String> destinationPrefixes = this.brokerHandler.getDestinationPrefixes();
if ((destination == null) || CollectionUtils.isEmpty(destinationPrefixes)) {
return false;

6
src/reference/docbook/web-sockets.xml
View File

@@ -135,6 +135,12 @@ public IntegrationWebSocketContainer serverWebSocketContainer() {
destinations, are routed to the <classname>AbstractBrokerMessageHandler</classname>, instead of to the
<code>outputChannel</code> of the <classname>WebSocketInboundChannelAdapter</classname>.
</para>
<para>
If <code>useBroker = false</code> and received message is of <code>SimpMessageType.CONNECT</code> type,
the <classname>WebSocketInboundChannelAdapter</classname> sends <code>SimpMessageType.CONNECT_ACK</code>
message to the <interfacename>WebSocketSession</interfacename> immediately without sending it to the
channel.
</para>
<note>
Spring's WebSocket Support allows the configuration of only one Broker Relay, hence we don't require an
<classname>AbstractBrokerMessageHandler</classname> reference, it is detected in the

9
src/reference/docbook/whats-new.xml
View File

@@ -212,5 +212,14 @@
See <xref linkend="aggregator-config"/> for more information.
</para>
</section>
<section id="4.1-script-outbound-channel-adapter">
<title>Outbound Channel Adapter and Scripts</title>
<para>
The <code>&lt;int:outbound-channel-adapter/&gt;</code> now supports the <code>&lt;script/&gt;</code>
sub-element. The underlying script must have a <code>void</code>
return type or return <code>null</code>.
See <xref linkend="groovy"/> and <xref linkend="scripting"/>.
</para>
</section>
</section>
</chapter>