From 5b395a8c5c458be79bc8d607d7a4462d5841cddc Mon Sep 17 00:00:00 2001
From: Gary Russell <grussell@pivotal.io>
Date: Wed, 27 May 2015 18:30:05 -0400
Subject: [PATCH] INT-3722: Additional Fix: TCP OG with Caching CF

JIRA: https://jira.spring.io/browse/INT-3722

While the initial fix solved the reported problem, it caused a memory leak when
the gateway completed its work before the reader thread "closed" the connection,
releasing it to the pool. This left the connection in the deferred close state.

Handle the condition where the gateway signals it has completed its work
before the connection is released to the pool.
(cherry picked from commit 3339313)
---
 .../CachingClientConnectionFactory.java       | 29 +++++++++++++------
 .../CachingClientConnectionFactoryTests.java  |  1 +
 2 files changed, 21 insertions(+), 9 deletions(-)

diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java
index ad44eb8d09..d46612b020 100644
--- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java
+++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java
@@ -16,7 +16,9 @@
 
 package org.springframework.integration.ip.tcp.connection;
 
+import java.util.HashSet;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executor;
 
@@ -48,6 +50,8 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
 	private final Map<String, CachedConnection> deferredClosures =
 			new ConcurrentHashMap<String, CachedConnection>();
 
+	private final Set<String> okToRelease = new HashSet<String>();
+
 	private volatile boolean deferClose;
 
 	/**
@@ -149,9 +153,14 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
 
 	@Override
 	public void closeDeferred(String connectionId) {
-		CachedConnection deferred = this.deferredClosures.remove(connectionId);
-		if (deferred != null) {
-			deferred.doClose();
+		synchronized(this.okToRelease) {
+			CachedConnection deferred = this.deferredClosures.remove(connectionId);
+			if (deferred != null) {
+				deferred.doClose();
+			}
+			else {
+				this.okToRelease.add(connectionId);
+			}
 		}
 	}
 
@@ -165,12 +174,14 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
 		}
 
 		@Override
-		public synchronized void close() {
-			if (deferClose && !this.released) {
-				deferredClosures.put(getConnectionId(), this);
-			}
-			else {
-				doClose();
+		public void close() {
+			synchronized(okToRelease) {
+				if (deferClose && !this.released && !okToRelease.remove(getConnectionId())) {
+					deferredClosures.put(getConnectionId(), this);
+				}
+				else {
+					doClose();
+				}
 			}
 		}
 
diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java
index 390f27048f..62dea6bf9e 100644
--- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java
+++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java
@@ -444,6 +444,7 @@ public class CachingClientConnectionFactoryTests {
 	}
 
 	@Test
+//	@Repeat(1000) // INT-3722
 	public void gatewayIntegrationTest() throws Exception {
 		final List<String> connectionIds = new ArrayList<String>();
 		final AtomicBoolean okToRun = new AtomicBoolean(true);