From 252c53db4c3d3647a1062b6c97b81d571ff360af Mon Sep 17 00:00:00 2001
From: Artem Bilan <abilan@pivotal.io>
Date: Mon, 13 Oct 2014 17:26:45 +0300
Subject: [PATCH] INT-3527: WebSocket and 'o-c-a' Improvements

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
---
 build.gradle                                  |  2 +-
 .../DefaultOutboundChannelAdapterParser.java  | 29 ++++++++++++--
 .../config/xml/spring-integration-4.1.xsd     | 12 +++---
 ...licationEventListeningMessageProducer.java | 13 +++++--
 .../CacheListeningMessageProducer.java        | 13 ++++++-
 .../ContinuousQueryMessageProducer.java       | 30 ++++++++-------
 .../GroovyServiceActivatorTests-context.xml   |  7 ++++
 .../config/GroovyServiceActivatorTests.java   | 20 ++++++++--
 .../websocket/ClientWebSocketContainer.java   |  2 +-
 .../IntegrationWebSocketContainer.java        |  2 +-
 .../WebSocketInboundChannelAdapter.java       | 38 ++++++++++++++-----
 src/reference/docbook/web-sockets.xml         |  6 +++
 src/reference/docbook/whats-new.xml           |  9 +++++
 13 files changed, 138 insertions(+), 45 deletions(-)

diff --git a/build.gradle b/build.gradle
index 0ef2345035..d48ea9640f 100644
--- a/build.gradle
+++ b/build.gradle
@@ -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'
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/DefaultOutboundChannelAdapterParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/DefaultOutboundChannelAdapterParser.java
index 2c3031c391..a3ae31f4ae 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/DefaultOutboundChannelAdapterParser.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/DefaultOutboundChannelAdapterParser.java
@@ -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) {
diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd
index d5a8761a0d..e203d0e80d 100644
--- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd
+++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd
@@ -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>
 
diff --git a/spring-integration-event/src/main/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducer.java b/spring-integration-event/src/main/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducer.java
index 76aef48ac7..8387e73ba4 100644
--- a/spring-integration-event/src/main/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducer.java
+++ b/spring-integration-event/src/main/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducer.java
@@ -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);
 			}
 		}
 	}
diff --git a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducer.java b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducer.java
index 336a4ed79f..a2daae8fa2 100644
--- a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducer.java
+++ b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducer.java
@@ -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);
 		}
 	}
 
diff --git a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/ContinuousQueryMessageProducer.java b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/ContinuousQueryMessageProducer.java
index 043d239da6..53ed8eaff5 100644
--- a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/ContinuousQueryMessageProducer.java
+++ b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/ContinuousQueryMessageProducer.java
@@ -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);
 		}
 	}
 
diff --git a/spring-integration-groovy/src/test/java/org/springframework/integration/groovy/config/GroovyServiceActivatorTests-context.xml b/spring-integration-groovy/src/test/java/org/springframework/integration/groovy/config/GroovyServiceActivatorTests-context.xml
index da3348f1e5..fdd49b3c43 100644
--- a/spring-integration-groovy/src/test/java/org/springframework/integration/groovy/config/GroovyServiceActivatorTests-context.xml
+++ b/spring-integration-groovy/src/test/java/org/springframework/integration/groovy/config/GroovyServiceActivatorTests-context.xml
@@ -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>
diff --git a/spring-integration-groovy/src/test/java/org/springframework/integration/groovy/config/GroovyServiceActivatorTests.java b/spring-integration-groovy/src/test/java/org/springframework/integration/groovy/config/GroovyServiceActivatorTests.java
index 5bdbc7199f..49d3b18d20 100644
--- a/spring-integration-groovy/src/test/java/org/springframework/integration/groovy/config/GroovyServiceActivatorTests.java
+++ b/spring-integration-groovy/src/test/java/org/springframework/integration/groovy/config/GroovyServiceActivatorTests.java
@@ -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) {
diff --git a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ClientWebSocketContainer.java b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ClientWebSocketContainer.java
index 8144a599c4..a7774f7741 100644
--- a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ClientWebSocketContainer.java
+++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ClientWebSocketContainer.java
@@ -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);
diff --git a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/IntegrationWebSocketContainer.java b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/IntegrationWebSocketContainer.java
index ff31a37e64..734d7efb36 100644
--- a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/IntegrationWebSocketContainer.java
+++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/IntegrationWebSocketContainer.java
@@ -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;
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 626509018e..1782a367e5 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
@@ -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;
diff --git a/src/reference/docbook/web-sockets.xml b/src/reference/docbook/web-sockets.xml
index 5afe4bb4cc..d2e2da1cf7 100644
--- a/src/reference/docbook/web-sockets.xml
+++ b/src/reference/docbook/web-sockets.xml
@@ -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
diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml
index da6a5f7c6a..70963a7c57 100644
--- a/src/reference/docbook/whats-new.xml
+++ b/src/reference/docbook/whats-new.xml
@@ -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>