Merge remote-tracking branch 'upstream/master' into 4.0.0-WIP

Conflicts:
	spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java
	spring-integration-core/src/test/java/org/springframework/integration/router/config/RouterWithMappingTests.java
	spring-integration-groovy/src/main/java/org/springframework/integration/groovy/GroovyScriptExecutingMessageProcessor.java
	spring-integration-http/src/test/java/org/springframework/integration/http/inbound/HttpRequestHandlingMessagingGatewayWithPathMappingTests.java
	spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractTxTimeoutMessageStoreTests.java
	spring-integration-jmx/src/main/java/org/springframework/integration/jmx/OperationInvokingMessageHandler.java

Resolved.

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java

@@ -17,17 +17,22 @@
 package org.springframework.integration.ip.tcp.connection;
 
 import java.net.Socket;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 /**
  * Abstract class for client connection factories; client connection factories
  * establish outgoing connections.
  * @author Gary Russell
+ * @author Artem Bilan
  * @since 2.0
  *
  */
 public abstract class AbstractClientConnectionFactory extends AbstractConnectionFactory {
 
-	private TcpConnectionSupport theConnection;
+	private final ReadWriteLock theConnectionLock = new ReentrantReadWriteLock();
+
+	private volatile TcpConnectionSupport theConnection;
 
 	/**
 	 * Constructs a factory that will established connections to the host and port.
@@ -45,18 +50,70 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection
 	 */
 	public TcpConnectionSupport getConnection() throws Exception {
 		this.checkActive();
-		if (this.isSingleUse()) {
-			return obtainConnection();
-		} else {
-			synchronized(this) {
-				TcpConnectionSupport connection = obtainConnection();
-				this.setTheConnection(connection);
+		return this.obtainConnection();
+	}
+
+	protected TcpConnectionSupport obtainConnection() throws Exception {
+		if (!this.isSingleUse()) {
+			TcpConnectionSupport connection = this.obtainSharedConnection();
+			if (connection != null) {
 				return connection;
 			}
 		}
+		return this.obtainNewConnection();
+	}
+
+	protected final TcpConnectionSupport obtainSharedConnection() throws InterruptedException {
+		this.theConnectionLock.readLock().lockInterruptibly();
+		try {
+			TcpConnectionSupport theConnection = this.getTheConnection();
+			if (theConnection != null && theConnection.isOpen()) {
+				return theConnection;
+			}
+		}
+		finally {
+			this.theConnectionLock.readLock().unlock();
+		}
+
+		return null;
+	}
+
+	protected final TcpConnectionSupport obtainNewConnection() throws Exception {
+		boolean singleUse = this.isSingleUse();
+		if (!singleUse) {
+			this.theConnectionLock.writeLock().lockInterruptibly();
+		}
+		try {
+			TcpConnectionSupport connection;
+			if (!singleUse) {
+				// Another write lock holder might have created a new one by now.
+				connection = this.obtainSharedConnection();
+				if (connection != null) {
+					return connection;
+				}
+			}
+
+			if (logger.isDebugEnabled()) {
+				logger.debug("Opening new socket connection to " + this.getHost() + ":" + this.getPort());
+			}
+
+			connection = this.buildNewConnection();
+			if (!singleUse) {
+				this.setTheConnection(connection);
+			}
+			connection.publishConnectionOpenEvent();
+			return connection;
+		}
+		finally {
+			if (!singleUse) {
+				this.theConnectionLock.writeLock().unlock();
+			}
+		}
 	}
 
-	protected abstract TcpConnectionSupport obtainConnection() throws Exception;
+	protected TcpConnectionSupport buildNewConnection() throws Exception {
+		throw new UnsupportedOperationException("Factories that don't override this class' obtainConnection() must implement this method");
+	}
 
 	/**
 	 * Transfers attributes such as (de)serializers, singleUse etc to a new connection.

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java

@@ -127,7 +127,6 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 		if (connectionFactoryName != null) {
 			this.connectionFactoryName = connectionFactoryName;
 		}
-		this.publishConnectionOpenEvent();
 		if (logger.isDebugEnabled()) {
 			logger.debug("New connection " + this.getConnectionId());
 		}

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java

@@ -44,20 +44,8 @@ public class TcpNetClientConnectionFactory extends
 		super(host, port);
 	}
 
-	/**
-	 * @throws IOException
-	 * @throws SocketException
-	 * @throws Exception
-	 */
 	@Override
-	protected TcpConnectionSupport obtainConnection() throws Exception {
-		TcpConnectionSupport theConnection = this.getTheConnection();
-		if (theConnection != null && theConnection.isOpen()) {
-			return theConnection;
-		}
-		if (logger.isDebugEnabled()) {
-			logger.debug("Opening new socket connection to " + this.getHost() + ":" + this.getPort());
-		}
+	protected TcpConnectionSupport buildNewConnection() throws IOException, SocketException, Exception {
 		Socket socket = createSocket(this.getHost(), this.getPort());
 		setSocketAttributes(socket);
 		TcpConnectionSupport connection = new TcpNetConnection(socket, false, this.isLookupHost(),

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java

@@ -64,7 +64,8 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
 		try {
 			if (this.getLocalAddress() == null) {
 				theServerSocket = createServerSocket(this.getPort(), this.getBacklog(), null);
-			} else {
+			}
+			else {
 				InetAddress whichNic = InetAddress.getByName(this.getLocalAddress());
 				theServerSocket = createServerSocket(this.getPort(), this.getBacklog(), whichNic);
 			}
@@ -80,7 +81,8 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
 				 */
 				try {
 					socket = serverSocket.accept();
-				} catch (SocketTimeoutException ste) {
+				}
+				catch (SocketTimeoutException ste) {
 					if (logger.isDebugEnabled()) {
 						logger.debug("Timed out on accept; continuing");
 					}
@@ -104,9 +106,11 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
 					this.initializeConnection(connection, socket);
 					this.getTaskExecutor().execute(connection);
 					this.harvestClosedConnections();
+					connection.publishConnectionOpenEvent();
 				}
 			}
-		} catch (Exception e) {
+		}
+		catch (Exception e) {
 			// don't log an error if we had a good socket once and now it's closed
 			if (e instanceof SocketException && theServerSocket != null) {
 				logger.warn("Server Socket closed");

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java

@@ -18,7 +18,6 @@ package org.springframework.integration.ip.tcp.connection;
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
-import java.net.SocketException;
 import java.nio.ByteBuffer;
 import java.nio.channels.CancelledKeyException;
 import java.nio.channels.ClosedChannelException;
@@ -61,13 +60,9 @@ public class TcpNioClientConnectionFactory extends
 		super(host, port);
 	}
 
-	/**
-	 * @throws Exception
-	 * @throws IOException
-	 * @throws SocketException
-	 */
 	@Override
-	protected TcpConnectionSupport obtainConnection() throws Exception {
+	protected void checkActive() throws IOException {
+		super.checkActive();
 		int n = 0;
 		while (this.selector == null) {
 			try {
@@ -76,16 +71,13 @@ public class TcpNioClientConnectionFactory extends
 				Thread.currentThread().interrupt();
 			}
 			if (n++ > 600) {
-				throw new Exception("Factory failed to start");
+				throw new IOException("Factory failed to start");
 			}
 		}
-		TcpConnectionSupport theConnection = this.getTheConnection();
-		if (theConnection != null && theConnection.isOpen()) {
-			return theConnection;
-		}
-		if (logger.isDebugEnabled()) {
-			logger.debug("Opening new socket channel connection to " + this.getHost() + ":" + this.getPort());
-		}
+	}
+
+	@Override
+	protected TcpConnectionSupport buildNewConnection() throws Exception {
 		SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress(this.getHost(), this.getPort()));
 		setSocketAttributes(socketChannel.socket());
 		TcpNioConnection connection = this.tcpNioConnectionSupport.createNewConnection(

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java

@@ -168,6 +168,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 				connection.setLastRead(now);
 				this.channelMap.put(channel, connection);
 				channel.register(selector, SelectionKey.OP_READ, connection);
+				connection.publishConnectionOpenEvent();
 			}
 			catch (Exception e) {
 				logger.error("Exception accepting new connection", e);

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests-context.xml

@@ -13,9 +13,9 @@
 
 	<bean id="tcpIpUtils" class="org.springframework.integration.test.util.SocketUtils" />
 
-	<int-ip:tcp-connection-factory id="server"
+	<int-ip:tcp-connection-factory id="serverNet"
 		type="server"
-		using-nio="true"
+		using-nio="false"
 		single-use="true"
 		port="#{tcpIpUtils.findAvailableServerSocket(10000)}"
 		task-executor="exec"
@@ -23,18 +23,44 @@
 		so-timeout="20000"
 	/>
 
-	<int-ip:tcp-connection-factory id="client"
+	<int-ip:tcp-connection-factory id="clientNet"
 		type="client"
 		host="localhost"
-		port="#{server.port}"
+		port="#{serverNet.port}"
 		single-use="true"
 		lookup-host="false"
 		so-timeout="100000"
 	/>
 
-	<int-ip:tcp-inbound-gateway id="looper"
+	<int-ip:tcp-connection-factory id="serverNio"
+		type="server"
+		using-nio="true"
+		single-use="true"
+		port="#{tcpIpUtils.findAvailableServerSocket(20000)}"
+		task-executor="exec"
+		lookup-host="false"
+		so-timeout="20000"
+	/>
+
+	<int-ip:tcp-connection-factory id="clientNio"
+		type="client"
+		host="localhost"
+		using-nio="true"
+		port="#{serverNio.port}"
+		single-use="true"
+		lookup-host="false"
+		so-timeout="100000"
+	/>
+
+	<int-ip:tcp-inbound-gateway id="gwNet"
 		request-channel="serverSideChannel"
-		connection-factory="server"
+		connection-factory="serverNet"
+		reply-timeout="1"
+		/>
+
+	<int-ip:tcp-inbound-gateway id="gwNio"
+		request-channel="serverSideChannel"
+		connection-factory="serverNio"
 		reply-timeout="1"
 		/>
 

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java

@@ -65,10 +65,16 @@ public class ConnectionToConnectionTests {
 	AbstractApplicationContext ctx;
 
 	@Autowired
-	private AbstractClientConnectionFactory client;
+	private AbstractClientConnectionFactory clientNet;
 
 	@Autowired
-	private AbstractServerConnectionFactory server;
+	private AbstractServerConnectionFactory serverNet;
+
+	@Autowired
+	private AbstractClientConnectionFactory clientNio;
+
+	@Autowired
+	private AbstractServerConnectionFactory serverNio;
 
 	@Autowired
 	private QueueChannel serverSideChannel;
@@ -90,9 +96,19 @@ public class ConnectionToConnectionTests {
 		ctx.close();
 	}
 
-	@SuppressWarnings("unchecked")
 	@Test
-	public void testConnect() throws Exception {
+	public void testConnectNet() throws Exception {
+		testConnectGuts(this.clientNet, this.serverNet, "gwNet", true);
+	}
+
+	@Test
+	public void testConnectNio() throws Exception {
+		testConnectGuts(this.clientNio, this.serverNio, "gwNio", false);
+	}
+
+	@SuppressWarnings("unchecked")
+	private void testConnectGuts(AbstractClientConnectionFactory client, AbstractServerConnectionFactory server,
+			String gatewayName, boolean expectExceptionOnClose) throws Exception {
 		TestingUtilities.waitListening(server, null);
 		client.start();
 		for (int i = 0; i < 100; i++) {
@@ -102,7 +118,7 @@ public class ConnectionToConnectionTests {
 			assertNotNull(message);
 			MessageHistory history = MessageHistory.read(message);
 			//org.springframework.integration.test.util.TestUtils
-			Properties componentHistoryRecord = TestUtils.locateComponentInHistory(history, "looper", 0);
+			Properties componentHistoryRecord = TestUtils.locateComponentInHistory(history, gatewayName, 0);
 			assertNotNull(componentHistoryRecord);
 			assertTrue(componentHistoryRecord.get("type").equals("ip:tcp-inbound-gateway"));
 			assertNotNull(message);
@@ -116,7 +132,7 @@ public class ConnectionToConnectionTests {
 		Message<TcpConnectionEvent> eventMessage;
 		while ((eventMessage = (Message<TcpConnectionEvent>) events.receive(1000)) != null) {
 			TcpConnectionEvent event = eventMessage.getPayload();
-			if ("client".equals(event.getConnectionFactoryName())) {
+			if (event.getConnectionFactoryName().startsWith("client")) {
 				if (event instanceof TcpConnectionOpenEvent) {
 					clientOpens++;
 				}
@@ -127,7 +143,7 @@ public class ConnectionToConnectionTests {
 					clientExceptions++;
 				}
 			}
-			else if ("server".equals(event.getConnectionFactoryName())) {
+			else if (event.getConnectionFactoryName().startsWith("server")) {
 				if (event instanceof TcpConnectionOpenEvent) {
 					serverOpens++;
 				}
@@ -138,7 +154,9 @@ public class ConnectionToConnectionTests {
 		}
 		assertEquals(100, clientOpens);
 		assertEquals(100, clientCloses);
-		assertEquals(100, clientExceptions);
+		if (expectExceptionOnClose) {
+			assertEquals(100, clientExceptions);
+		}
 		assertEquals(100, serverOpens);
 		assertEquals(100, serverCloses);
 	}
@@ -146,16 +164,16 @@ public class ConnectionToConnectionTests {
 	@Test
 	public void testConnectRaw() throws Exception {
 		ByteArrayRawSerializer serializer = new ByteArrayRawSerializer();
-		client.setSerializer(serializer);
-		server.setDeserializer(serializer);
-		client.start();
-		TcpConnection connection = client.getConnection();
+		clientNet.setSerializer(serializer);
+		serverNet.setDeserializer(serializer);
+		clientNet.start();
+		TcpConnection connection = clientNet.getConnection();
 		connection.send(MessageBuilder.withPayload("Test").build());
 		Message<?> message = serverSideChannel.receive(10000);
 		assertNotNull(message);
 		MessageHistory history = MessageHistory.read(message);
 		//org.springframework.integration.test.util.TestUtils
-		Properties componentHistoryRecord = TestUtils.locateComponentInHistory(history, "looper", 0);
+		Properties componentHistoryRecord = TestUtils.locateComponentInHistory(history, "gwNet", 0);
 		assertNotNull(componentHistoryRecord);
 		assertTrue(componentHistoryRecord.get("type").equals("ip:tcp-inbound-gateway"));
 		assertNotNull(message);
@@ -164,16 +182,16 @@ public class ConnectionToConnectionTests {
 
 	@Test
 	public void testLookup() throws Exception {
-		client.start();
-		TcpConnection connection = client.getConnection();
+		clientNet.start();
+		TcpConnection connection = clientNet.getConnection();
 		assertFalse(connection.getConnectionId().contains("localhost"));
 		connection.close();
-		client.setLookupHost(true);
-		connection = client.getConnection();
+		clientNet.setLookupHost(true);
+		connection = clientNet.getConnection();
 		assertTrue(connection.getConnectionId().contains("localhost"));
 		connection.close();
-		client.setLookupHost(false);
-		connection = client.getConnection();
+		clientNet.setLookupHost(false);
+		connection = clientNet.getConnection();
 		assertFalse(connection.getConnectionId().contains("localhost"));
 		connection.close();
 	}

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java

@@ -52,10 +52,10 @@ public class ConnectionEventTests {
 				theEvent.add((TcpConnectionEvent) event);
 			}
 		}, "foo");
-		assertTrue(theEvent.size() > 0);
-		assertNotNull(theEvent.get(0));
-		assertTrue(theEvent.get(0) instanceof TcpConnectionOpenEvent);
-		assertTrue(theEvent.get(0).toString().endsWith("[factory=foo, connectionId=" + conn.getConnectionId() + "] **OPENED**"));
+		/*
+		 *  Open is not published by the connection itself; the factory publishes it after initialization.
+		 *  See ConnectionToConnectionTests.
+		 */
 		@SuppressWarnings("unchecked")
 		Serializer<Object> serializer = mock(Serializer.class);
 		RuntimeException toBeThrown = new RuntimeException("foo");
@@ -67,17 +67,17 @@ public class ConnectionEventTests {
 			fail("Expected exception");
 		}
 		catch (Exception e) {}
-		assertTrue(theEvent.size() > 1);
-		assertNotNull(theEvent.get(1));
-		assertTrue(theEvent.get(1) instanceof TcpConnectionExceptionEvent);
-		assertTrue(theEvent.get(1).toString().endsWith("[factory=foo, connectionId=" + conn.getConnectionId() + "]"));
-		assertTrue(theEvent.get(1).toString().contains("cause=java.lang.RuntimeException: foo]"));
-		TcpConnectionExceptionEvent event = (TcpConnectionExceptionEvent) theEvent.get(1);
+		assertTrue(theEvent.size() > 0);
+		assertNotNull(theEvent.get(0));
+		assertTrue(theEvent.get(0) instanceof TcpConnectionExceptionEvent);
+		assertTrue(theEvent.get(0).toString().endsWith("[factory=foo, connectionId=" + conn.getConnectionId() + "]"));
+		assertTrue(theEvent.get(0).toString().contains("cause=java.lang.RuntimeException: foo]"));
+		TcpConnectionExceptionEvent event = (TcpConnectionExceptionEvent) theEvent.get(0);
 		assertNotNull(event.getCause());
 		assertSame(toBeThrown, event.getCause());
-		assertTrue(theEvent.size() > 2);
-		assertNotNull(theEvent.get(2));
-		assertTrue(theEvent.get(2).toString().endsWith("[factory=foo, connectionId=" + conn.getConnectionId() + "] **CLOSED**"));
+		assertTrue(theEvent.size() > 1);
+		assertNotNull(theEvent.get(1));
+		assertTrue(theEvent.get(1).toString().endsWith("[factory=foo, connectionId=" + conn.getConnectionId() + "] **CLOSED**"));
 	}
 
 }