Merge pull request #397 from garyrussell/INT-1871
* INT-1871: INT-1871 TCP CachingClientConnectionFactory
This commit is contained in:
Oleg Zhurakousky
@@ -0,0 +1,74 @@
|
||||
/*
|
||||
* Copyright 2002-2012 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.util;
|
||||
|
||||
/**
|
||||
* Represents a pool of items.
|
||||
* @author Gary Russell
|
||||
* @since 2.2
|
||||
*
|
||||
*/
|
||||
public interface Pool<T> {
|
||||
|
||||
/**
|
||||
* Obtains an item from the pool.
|
||||
* @return the item.
|
||||
*/
|
||||
T getItem();
|
||||
|
||||
/**
|
||||
* Releases an item back into the pool. This must be an item that
|
||||
* was previously retrieved using {@link #getItem()}.
|
||||
* @param t the item.
|
||||
* @throws IllegalArgumentException when a "foreign" object
|
||||
* is released.
|
||||
*/
|
||||
void releaseItem(T t);
|
||||
|
||||
/**
|
||||
* Removes all idle items from the pool.
|
||||
*/
|
||||
void removeAllIdleItems();
|
||||
|
||||
/**
|
||||
* Returns the current size (limit) of the pool.
|
||||
* @return the size.
|
||||
*/
|
||||
int getPoolSize();
|
||||
|
||||
/**
|
||||
* Returns the number of items that have been allocated
|
||||
* but are not currently in use.
|
||||
* @return The number of items.
|
||||
*/
|
||||
int getIdleCount();
|
||||
|
||||
/**
|
||||
* Returns the number of allocated items that are currently
|
||||
* checked out of the pool.
|
||||
* @return The number of items.
|
||||
*/
|
||||
int getActiveCount();
|
||||
|
||||
/**
|
||||
* Returns the current count of allocated items (in use and
|
||||
* idle). May be less than the pool size, and reflects the
|
||||
* high water mark of pool usage.
|
||||
* @return the number of items.
|
||||
*/
|
||||
int getAllocatedCount();
|
||||
|
||||
}
|
||||
@@ -0,0 +1,263 @@
|
||||
/*
|
||||
* Copyright 2002-2012 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.util;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.HashSet;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
import java.util.concurrent.LinkedBlockingQueue;
|
||||
import java.util.concurrent.Semaphore;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import org.springframework.integration.MessagingException;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* Implementation of {@link Pool} supporting dynamic resizing and a variable
|
||||
* timeout when attempting to obtain an item from the pool. Pool grows on
|
||||
* demand up to the limit.
|
||||
* @author Gary Russell
|
||||
* @since 2.2
|
||||
*
|
||||
*/
|
||||
public class SimplePool<T> implements Pool<T> {
|
||||
|
||||
protected final Log logger = LogFactory.getLog(this.getClass());
|
||||
|
||||
private final Semaphore permits = new Semaphore(0);
|
||||
|
||||
private final AtomicInteger poolSize = new AtomicInteger();
|
||||
|
||||
private final AtomicInteger targetPoolSize = new AtomicInteger();
|
||||
|
||||
private long waitTimeout = Long.MAX_VALUE;
|
||||
|
||||
private final BlockingQueue<T> available = new LinkedBlockingQueue<T>();
|
||||
|
||||
private final Set<T> allocated = Collections.synchronizedSet(new HashSet<T>());
|
||||
|
||||
private final PoolItemCallback<T> callback;
|
||||
|
||||
/**
|
||||
* Creates a SimplePool with a specific limit.
|
||||
* @param poolSize The maximum number of items the pool supports.
|
||||
* @param callback A {@link PoolItemCallback} implementation called during various
|
||||
* pool operations.
|
||||
*/
|
||||
public SimplePool(int poolSize, PoolItemCallback<T> callback) {
|
||||
if (poolSize <= 0) {
|
||||
this.poolSize.set(Integer.MAX_VALUE);
|
||||
this.targetPoolSize.set(Integer.MAX_VALUE);
|
||||
this.permits.release(Integer.MAX_VALUE);
|
||||
}
|
||||
else {
|
||||
this.poolSize.set(poolSize);
|
||||
this.targetPoolSize.set(poolSize);
|
||||
this.permits.release(poolSize);
|
||||
}
|
||||
this.callback = callback;
|
||||
}
|
||||
|
||||
/**
|
||||
* Adjusts the current pool size. When reducing the pool size, attempts to
|
||||
* remove the delta from the pool. If there are not enough unused items in
|
||||
* the pool, the actual pool size will decrease to the specified size as in-use
|
||||
* items are returned.
|
||||
* @param poolSize The desired target pool size.
|
||||
*/
|
||||
public synchronized void setPoolSize(int poolSize) {
|
||||
int delta = poolSize - this.poolSize.get();
|
||||
this.targetPoolSize.addAndGet(delta);
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug(String.format("Target pool size changed by %d, now %d", delta, this.targetPoolSize.get()));
|
||||
}
|
||||
if (delta > 0) {
|
||||
this.poolSize.addAndGet(delta);
|
||||
this.permits.release(delta);
|
||||
}
|
||||
else while (delta < 0) {
|
||||
if (!this.permits.tryAcquire()) {
|
||||
break;
|
||||
}
|
||||
T item = this.available.poll();
|
||||
if (item == null) {
|
||||
this.permits.release();
|
||||
break;
|
||||
}
|
||||
doRemoveItem(item);
|
||||
this.poolSize.decrementAndGet();
|
||||
delta++;
|
||||
}
|
||||
if (delta < 0 && logger.isDebugEnabled()) {
|
||||
logger.debug(String.format("Pool is overcommitted by %d; items will be removed when returned", -delta));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the current size of the pool; may be greater than the target pool size
|
||||
* if it was recently reduced and too many items were in use to allow the new size
|
||||
* to be set.
|
||||
*/
|
||||
public int getPoolSize() {
|
||||
return this.poolSize.get();
|
||||
}
|
||||
|
||||
public int getIdleCount() {
|
||||
return this.available.size();
|
||||
}
|
||||
|
||||
public int getActiveCount() {
|
||||
return this.getAllocatedCount() - this.getIdleCount();
|
||||
}
|
||||
|
||||
public int getAllocatedCount() {
|
||||
return this.allocated.size();
|
||||
}
|
||||
|
||||
/**
|
||||
* Adjusts the wait timeout - the time for which getItem() will wait if no idle
|
||||
* entries are available. <br/>Default: infinity.
|
||||
* @param waitTimeout The wait timeout in milliseconds.
|
||||
*/
|
||||
public void setWaitTimeout(long waitTimeout) {
|
||||
this.waitTimeout = waitTimeout;
|
||||
}
|
||||
|
||||
/**
|
||||
* Obtains an item from the pool; waits up to waitTime milliseconds (default infinity).
|
||||
* @throws MessagingException if no items become available in time.
|
||||
*/
|
||||
public T getItem() {
|
||||
boolean permitted = false;
|
||||
try {
|
||||
try {
|
||||
permitted = this.permits.tryAcquire(this.waitTimeout, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
throw new MessagingException("Interrupted awaiting a pooled resource", e);
|
||||
}
|
||||
if (!permitted) {
|
||||
throw new IllegalStateException("Timed out while waiting to aquire a pool entry.");
|
||||
}
|
||||
T item = doGetItem();
|
||||
return item;
|
||||
}
|
||||
catch (Exception e) {
|
||||
if (permitted) {
|
||||
this.permits.release();
|
||||
}
|
||||
if (e instanceof MessagingException) {
|
||||
throw (MessagingException) e;
|
||||
}
|
||||
throw new MessagingException("Failed to obtain pooled item", e);
|
||||
}
|
||||
}
|
||||
|
||||
private T doGetItem() {
|
||||
T item = this.available.poll();
|
||||
if (item != null && logger.isDebugEnabled()) {
|
||||
logger.debug("Obtained " + item + " from pool.");
|
||||
}
|
||||
if (item == null) {
|
||||
item = this.callback.createForPool();
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("Obtained new " + item + ".");
|
||||
}
|
||||
allocated.add(item);
|
||||
}
|
||||
else if (this.callback.isStale(item)) {
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("Received a stale item, will attempt to get a new one.");
|
||||
}
|
||||
item = doGetItem();
|
||||
}
|
||||
return item;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns an item to the pool. Item may be null, in which case a subsequent getItem()
|
||||
* will return a new instance.
|
||||
*/
|
||||
public synchronized void releaseItem(T item) {
|
||||
Assert.isTrue(item == null || this.allocated.contains(item),
|
||||
"You can only release items that were obtained from the pool");
|
||||
if (this.poolSize.get() > targetPoolSize.get()) {
|
||||
poolSize.decrementAndGet();
|
||||
if (item != null) {
|
||||
doRemoveItem(item);
|
||||
}
|
||||
}
|
||||
else {
|
||||
if (logger.isDebugEnabled()){
|
||||
logger.debug("Releasing " + item + " back to the pool");
|
||||
}
|
||||
if (item != null) {
|
||||
available.add(item);
|
||||
}
|
||||
permits.release();
|
||||
}
|
||||
}
|
||||
|
||||
public synchronized void removeAllIdleItems() {
|
||||
T item;
|
||||
while ((item = this.available.poll()) != null) {
|
||||
doRemoveItem(item);
|
||||
}
|
||||
}
|
||||
|
||||
private void doRemoveItem(T item) {
|
||||
this.allocated.remove(item);
|
||||
this.callback.removedFromPool(item);
|
||||
}
|
||||
|
||||
/**
|
||||
* User of the pool provide an implementation of this interface; called during
|
||||
* various pool operations.
|
||||
*
|
||||
*/
|
||||
public static interface PoolItemCallback<T> {
|
||||
|
||||
/**
|
||||
* Called by the pool when a new instance is required to populate the pool. Only
|
||||
* called if no idle non-stale instances are available.
|
||||
* @return The item.
|
||||
*/
|
||||
T createForPool();
|
||||
|
||||
/**
|
||||
* Called by the pool when an idle item is retrieved from the pool. Indicates
|
||||
* whether that item is usable, or should be discarded. The pool takes no
|
||||
* further action on a stale item, discards it, and attempts to find or create
|
||||
* another item.
|
||||
* @param item The item.
|
||||
* @return true if the item should not be used.
|
||||
*/
|
||||
boolean isStale(T item);
|
||||
|
||||
/**
|
||||
* Called by the pool when an item is forcibly removed from the pool - for example
|
||||
* when the pool size is reduced. The implementation should perform any cleanup
|
||||
* necessary on the item, such as closing connections etc.
|
||||
* @param item The item.
|
||||
*/
|
||||
void removedFromPool(T item);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,152 @@
|
||||
/*
|
||||
* Copyright 2002-2012 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.util;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertFalse;
|
||||
import static org.junit.Assert.assertNotSame;
|
||||
import static org.junit.Assert.assertSame;
|
||||
import static org.junit.Assert.fail;
|
||||
|
||||
import java.util.HashSet;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
import org.junit.Test;
|
||||
import org.springframework.integration.MessagingException;
|
||||
|
||||
/**
|
||||
* @author Gary Russell
|
||||
* @since 2.2
|
||||
*
|
||||
*/
|
||||
public class SimplePoolTests {
|
||||
|
||||
@Test
|
||||
public void testReuseAndStale() {
|
||||
final Set<String> strings = new HashSet<String>();
|
||||
final AtomicBoolean stale = new AtomicBoolean();
|
||||
SimplePool<String> pool = stringPool(2, strings, stale);
|
||||
String s1 = pool.getItem();
|
||||
String s2 = pool.getItem();
|
||||
assertNotSame(s1, s2);
|
||||
pool.releaseItem(s1);
|
||||
String s3 = pool.getItem();
|
||||
assertSame(s1, s3);
|
||||
stale.set(true);
|
||||
pool.releaseItem(s3);
|
||||
s3 = pool.getItem();
|
||||
assertNotSame(s1, s3);
|
||||
assertFalse(strings.remove(s1));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testOverCommitandResize() {
|
||||
final Set<String> strings = new HashSet<String>();
|
||||
final AtomicBoolean stale = new AtomicBoolean();
|
||||
SimplePool<String> pool = stringPool(2, strings, stale);
|
||||
String s1 = pool.getItem();
|
||||
assertEquals(0, pool.getIdleCount());
|
||||
assertEquals(1, pool.getActiveCount());
|
||||
assertEquals(1, pool.getAllocatedCount());
|
||||
pool.releaseItem(s1);
|
||||
assertEquals(1, pool.getIdleCount());
|
||||
assertEquals(0, pool.getActiveCount());
|
||||
assertEquals(1, pool.getAllocatedCount());
|
||||
s1 = pool.getItem();
|
||||
assertEquals(0, pool.getIdleCount());
|
||||
assertEquals(1, pool.getActiveCount());
|
||||
assertEquals(1, pool.getAllocatedCount());
|
||||
String s2 = pool.getItem();
|
||||
assertNotSame(s1, s2);
|
||||
pool.setWaitTimeout(1);
|
||||
assertEquals(0, pool.getIdleCount());
|
||||
assertEquals(2, pool.getActiveCount());
|
||||
assertEquals(2, pool.getAllocatedCount());
|
||||
try {
|
||||
pool.getItem();
|
||||
fail("Expected exception");
|
||||
} catch (MessagingException e) {}
|
||||
|
||||
// resize up
|
||||
pool.setPoolSize(4);
|
||||
|
||||
assertEquals(0, pool.getIdleCount());
|
||||
assertEquals(2, pool.getActiveCount());
|
||||
assertEquals(2, pool.getAllocatedCount());
|
||||
String s3 = pool.getItem();
|
||||
String s4 = pool.getItem();
|
||||
assertEquals(0, pool.getIdleCount());
|
||||
assertEquals(4, pool.getActiveCount());
|
||||
assertEquals(4, pool.getAllocatedCount());
|
||||
pool.releaseItem(s4);
|
||||
assertEquals(1, pool.getIdleCount());
|
||||
assertEquals(3, pool.getActiveCount());
|
||||
assertEquals(4, pool.getAllocatedCount());
|
||||
|
||||
// resize down
|
||||
pool.setPoolSize(2);
|
||||
|
||||
assertEquals(0, pool.getIdleCount());
|
||||
assertEquals(3, pool.getActiveCount());
|
||||
assertEquals(3, pool.getPoolSize());
|
||||
assertEquals(3, pool.getAllocatedCount());
|
||||
pool.releaseItem(s3);
|
||||
assertEquals(0, pool.getIdleCount());
|
||||
assertEquals(2, pool.getActiveCount());
|
||||
assertEquals(2, pool.getPoolSize());
|
||||
assertEquals(2, pool.getAllocatedCount());
|
||||
assertEquals(2, strings.size());
|
||||
pool.releaseItem(s2);
|
||||
pool.releaseItem(s1);
|
||||
assertEquals(2, pool.getIdleCount());
|
||||
assertEquals(0, pool.getActiveCount());
|
||||
assertEquals(2, pool.getPoolSize());
|
||||
assertEquals(2, strings.size());
|
||||
assertEquals(2, pool.getAllocatedCount());
|
||||
}
|
||||
|
||||
@Test(expected=IllegalArgumentException.class)
|
||||
public void testForeignObject() {
|
||||
final Set<String> strings = new HashSet<String>();
|
||||
final AtomicBoolean stale = new AtomicBoolean();
|
||||
SimplePool<String> pool = stringPool(2, strings, stale);
|
||||
pool.getItem();
|
||||
pool.releaseItem("Hello, world!");
|
||||
}
|
||||
|
||||
private SimplePool<String> stringPool(int size, final Set<String> strings,
|
||||
final AtomicBoolean stale) {
|
||||
SimplePool<String> pool = new SimplePool<String>(size, new SimplePool.PoolItemCallback<String>() {
|
||||
private int i;
|
||||
public String createForPool() {
|
||||
String string = new String("String" + i++);
|
||||
strings.add(string);
|
||||
return string;
|
||||
}
|
||||
public boolean isStale(String item) {
|
||||
if (stale.get()) {
|
||||
strings.remove(item);
|
||||
}
|
||||
return stale.get();
|
||||
}
|
||||
public void removedFromPool(String item) {
|
||||
strings.remove(item);
|
||||
}
|
||||
});
|
||||
return pool;
|
||||
}
|
||||
}
|
||||
@@ -19,12 +19,11 @@ package org.springframework.integration.file.remote.session;
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.io.OutputStream;
|
||||
import java.util.concurrent.LinkedBlockingQueue;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import org.springframework.beans.factory.DisposableBean;
|
||||
import org.springframework.integration.util.UpperBound;
|
||||
import org.springframework.integration.util.SimplePool;
|
||||
|
||||
/**
|
||||
* A {@link SessionFactory} implementation that caches Sessions for reuse without
|
||||
@@ -41,15 +40,9 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBe
|
||||
|
||||
private static final Log logger = LogFactory.getLog(CachingSessionFactory.class);
|
||||
|
||||
|
||||
private volatile long sessionWaitTimeout = Integer.MAX_VALUE;
|
||||
|
||||
private final LinkedBlockingQueue<Session<F>> queue = new LinkedBlockingQueue<Session<F>>();
|
||||
|
||||
private final SessionFactory<F> sessionFactory;
|
||||
|
||||
private final UpperBound sessionPermits;
|
||||
|
||||
private final SimplePool<Session<F>> pool;
|
||||
|
||||
public CachingSessionFactory(SessionFactory<F> sessionFactory) {
|
||||
this(sessionFactory, 0);
|
||||
@@ -57,7 +50,19 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBe
|
||||
|
||||
public CachingSessionFactory(SessionFactory<F> sessionFactory, int sessionCacheSize) {
|
||||
this.sessionFactory = sessionFactory;
|
||||
this.sessionPermits = new UpperBound(sessionCacheSize);
|
||||
this.pool = new SimplePool<Session<F>>(sessionCacheSize, new SimplePool.PoolItemCallback<Session<F>>() {
|
||||
public Session<F> createForPool() {
|
||||
return CachingSessionFactory.this.sessionFactory.getSession();
|
||||
}
|
||||
|
||||
public boolean isStale(Session<F> session) {
|
||||
return !session.isOpen();
|
||||
}
|
||||
|
||||
public void removedFromPool(Session<F> session) {
|
||||
session.close();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -67,52 +72,19 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBe
|
||||
* @throws {@link IllegalStateException} if the wait expires prior to a Session becoming available.
|
||||
*/
|
||||
public void setSessionWaitTimeout(long sessionWaitTimeout) {
|
||||
this.sessionWaitTimeout = sessionWaitTimeout;
|
||||
this.pool.setWaitTimeout(sessionWaitTimeout);
|
||||
}
|
||||
|
||||
public Session<F> getSession() {
|
||||
boolean permitted = this.sessionPermits.tryAcquire(this.sessionWaitTimeout);
|
||||
if (!permitted) {
|
||||
throw new IllegalStateException("Timed out while waiting to aquire a Session.");
|
||||
}
|
||||
Session<F> session = this.doGetSession();
|
||||
return new CachedSession(session);
|
||||
public void setPoolSize(int poolSize) {
|
||||
this.pool.setPoolSize(poolSize);
|
||||
}
|
||||
|
||||
public Session<F> getSession() {
|
||||
return new CachedSession(this.pool.getItem());
|
||||
}
|
||||
|
||||
public void destroy() {
|
||||
if (this.queue != null) {
|
||||
for (Session<F> session : this.queue) {
|
||||
this.closeSession(session);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private Session<F> doGetSession() {
|
||||
Session<F> session = this.queue.poll();
|
||||
if (session != null && !session.isOpen()) {
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("Received a stale Session, will attempt to get a new one.");
|
||||
}
|
||||
return this.doGetSession();
|
||||
}
|
||||
else if (session == null){
|
||||
session = this.sessionFactory.getSession();
|
||||
}
|
||||
return session;
|
||||
}
|
||||
|
||||
private void closeSession(Session<F> session) {
|
||||
try {
|
||||
if (session != null) {
|
||||
session.close();
|
||||
}
|
||||
}
|
||||
catch (Throwable e) {
|
||||
if (logger.isWarnEnabled()) {
|
||||
// log and ignore
|
||||
logger.warn("Exception was thrown while destroying Session. ", e);
|
||||
}
|
||||
}
|
||||
this.pool.removeAllIdleItems();
|
||||
}
|
||||
|
||||
|
||||
@@ -120,16 +92,25 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBe
|
||||
|
||||
private final Session<F> targetSession;
|
||||
|
||||
private boolean released;
|
||||
|
||||
private CachedSession(Session<F> targetSession) {
|
||||
this.targetSession = targetSession;
|
||||
}
|
||||
|
||||
public void close() {
|
||||
if (logger.isDebugEnabled()){
|
||||
logger.debug("Releasing Session back to the pool.");
|
||||
public synchronized void close() {
|
||||
if (released) {
|
||||
if (logger.isDebugEnabled()){
|
||||
logger.debug("Session already released.");
|
||||
}
|
||||
}
|
||||
else {
|
||||
if (logger.isDebugEnabled()){
|
||||
logger.debug("Releasing Session back to the pool.");
|
||||
}
|
||||
pool.releaseItem(targetSession);
|
||||
released = true;
|
||||
}
|
||||
queue.add(targetSession);
|
||||
sessionPermits.release();
|
||||
}
|
||||
|
||||
public boolean remove(String path) throws IOException{
|
||||
|
||||
@@ -27,6 +27,7 @@ import org.apache.commons.net.ftp.FTPClient;
|
||||
import org.junit.Ignore;
|
||||
import org.junit.Test;
|
||||
import org.mockito.Mockito;
|
||||
import org.springframework.integration.MessagingException;
|
||||
import org.springframework.integration.file.remote.session.CachingSessionFactory;
|
||||
import org.springframework.integration.file.remote.session.Session;
|
||||
import org.springframework.integration.file.remote.session.SessionFactory;
|
||||
@@ -133,7 +134,7 @@ public class SessionFactoryTests {
|
||||
Mockito.verify(sessionFactory, Mockito.times(2)).getSession();
|
||||
}
|
||||
|
||||
@Test (expected=IllegalStateException.class) // timeout expire
|
||||
@Test (expected=MessagingException.class) // timeout expire
|
||||
public void testSessionWaitExpire() throws Exception{
|
||||
SessionFactory sessionFactory = Mockito.mock(SessionFactory.class);
|
||||
Session session = Mockito.mock(Session.class);
|
||||
|
||||
@@ -0,0 +1,323 @@
|
||||
/*
|
||||
* Copyright 2002-2012 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.ip.tcp.connection;
|
||||
|
||||
import java.util.concurrent.Executor;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import org.springframework.core.serializer.Deserializer;
|
||||
import org.springframework.core.serializer.Serializer;
|
||||
import org.springframework.integration.MessagingException;
|
||||
import org.springframework.integration.util.SimplePool;
|
||||
|
||||
/**
|
||||
* @author Gary Russell
|
||||
* @since 2.2
|
||||
*
|
||||
*/
|
||||
public class CachingClientConnectionFactory extends AbstractClientConnectionFactory {
|
||||
|
||||
private final Log logger = LogFactory.getLog(this.getClass());
|
||||
|
||||
private final AbstractClientConnectionFactory targetConnectionFactory;
|
||||
|
||||
private final SimplePool<TcpConnection> pool;
|
||||
|
||||
public CachingClientConnectionFactory(AbstractClientConnectionFactory target, int poolSize) {
|
||||
super("", 0);
|
||||
// override single-use to true to force "close" after use
|
||||
target.setSingleUse(true);
|
||||
this.targetConnectionFactory = target;
|
||||
pool = new SimplePool<TcpConnection>(poolSize, new SimplePool.PoolItemCallback<TcpConnection>() {
|
||||
|
||||
public TcpConnection createForPool() {
|
||||
try {
|
||||
return targetConnectionFactory.getConnection();
|
||||
} catch (Exception e) {
|
||||
throw new MessagingException("Failed to obtain connection", e);
|
||||
}
|
||||
}
|
||||
|
||||
public boolean isStale(TcpConnection connection) {
|
||||
return !connection.isOpen();
|
||||
}
|
||||
|
||||
public void removedFromPool(TcpConnection connection) {
|
||||
connection.close();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
public void setConnectionWaitTimeout(int connectionWaitTimeout) {
|
||||
this.pool.setWaitTimeout(connectionWaitTimeout);
|
||||
}
|
||||
|
||||
public synchronized void setPoolSize(int poolSize) {
|
||||
this.pool.setPoolSize(poolSize);
|
||||
}
|
||||
|
||||
public int getPoolSize() {
|
||||
return this.pool.getPoolSize();
|
||||
}
|
||||
|
||||
public int getIdleCount() {
|
||||
return this.pool.getIdleCount();
|
||||
}
|
||||
|
||||
public int getActiveCount() {
|
||||
return this.pool.getActiveCount();
|
||||
}
|
||||
|
||||
public int getAllocatedCount() {
|
||||
return this.pool.getAllocatedCount();
|
||||
}
|
||||
|
||||
public TcpConnection getOrMakeConnection() throws Exception {
|
||||
return new CachedConnection(this.pool.getItem());
|
||||
}
|
||||
|
||||
private class CachedConnection extends AbstractTcpConnectionInterceptor {
|
||||
|
||||
private volatile boolean released;
|
||||
|
||||
public CachedConnection(TcpConnection connection) {
|
||||
super.setTheConnection(connection);
|
||||
if (connection instanceof AbstractTcpConnection) {
|
||||
((AbstractTcpConnection) connection).registerListener(this);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void close() {
|
||||
/**
|
||||
* If the delegate is stopped, actually close
|
||||
* the connection.
|
||||
*/
|
||||
if (!isRunning()) {
|
||||
if (logger.isDebugEnabled()){
|
||||
logger.debug("Factory not running - closing " + this.getConnectionId());
|
||||
}
|
||||
pool.releaseItem(null); // just open up a permit
|
||||
super.close();
|
||||
}
|
||||
else if(this.released) {
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("Connection " + this.getConnectionId() + " has already been released");
|
||||
}
|
||||
}
|
||||
else {
|
||||
pool.releaseItem(this.getTheConnection());
|
||||
this.released = true;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getConnectionId() {
|
||||
return "Cached:" + super.getConnectionId();
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return this.getConnectionId();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
///////////////// DELEGATE METHODS ///////////////////////
|
||||
|
||||
public void run() {
|
||||
|
||||
}
|
||||
|
||||
public boolean isRunning() {
|
||||
return targetConnectionFactory.isRunning();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
targetConnectionFactory.close();
|
||||
}
|
||||
|
||||
public int hashCode() {
|
||||
return targetConnectionFactory.hashCode();
|
||||
}
|
||||
|
||||
public void setComponentName(String componentName) {
|
||||
targetConnectionFactory.setComponentName(componentName);
|
||||
}
|
||||
|
||||
public String getComponentType() {
|
||||
return targetConnectionFactory.getComponentType();
|
||||
}
|
||||
|
||||
public boolean equals(Object obj) {
|
||||
return targetConnectionFactory.equals(obj);
|
||||
}
|
||||
|
||||
public int getSoTimeout() {
|
||||
return targetConnectionFactory.getSoTimeout();
|
||||
}
|
||||
|
||||
public void setSoTimeout(int soTimeout) {
|
||||
targetConnectionFactory.setSoTimeout(soTimeout);
|
||||
}
|
||||
|
||||
public int getSoReceiveBufferSize() {
|
||||
return targetConnectionFactory.getSoReceiveBufferSize();
|
||||
}
|
||||
|
||||
public void setSoReceiveBufferSize(int soReceiveBufferSize) {
|
||||
targetConnectionFactory.setSoReceiveBufferSize(soReceiveBufferSize);
|
||||
}
|
||||
|
||||
public int getSoSendBufferSize() {
|
||||
return targetConnectionFactory.getSoSendBufferSize();
|
||||
}
|
||||
|
||||
public void setSoSendBufferSize(int soSendBufferSize) {
|
||||
targetConnectionFactory.setSoSendBufferSize(soSendBufferSize);
|
||||
}
|
||||
|
||||
public boolean isSoTcpNoDelay() {
|
||||
return targetConnectionFactory.isSoTcpNoDelay();
|
||||
}
|
||||
|
||||
public void setSoTcpNoDelay(boolean soTcpNoDelay) {
|
||||
targetConnectionFactory.setSoTcpNoDelay(soTcpNoDelay);
|
||||
}
|
||||
|
||||
public int getSoLinger() {
|
||||
return targetConnectionFactory.getSoLinger();
|
||||
}
|
||||
|
||||
public void setSoLinger(int soLinger) {
|
||||
targetConnectionFactory.setSoLinger(soLinger);
|
||||
}
|
||||
|
||||
public boolean isSoKeepAlive() {
|
||||
return targetConnectionFactory.isSoKeepAlive();
|
||||
}
|
||||
|
||||
public void setSoKeepAlive(boolean soKeepAlive) {
|
||||
targetConnectionFactory.setSoKeepAlive(soKeepAlive);
|
||||
}
|
||||
|
||||
public int getSoTrafficClass() {
|
||||
return targetConnectionFactory.getSoTrafficClass();
|
||||
}
|
||||
|
||||
public void setSoTrafficClass(int soTrafficClass) {
|
||||
targetConnectionFactory.setSoTrafficClass(soTrafficClass);
|
||||
}
|
||||
|
||||
public String getHost() {
|
||||
return targetConnectionFactory.getHost();
|
||||
}
|
||||
|
||||
public int getPort() {
|
||||
return targetConnectionFactory.getPort();
|
||||
}
|
||||
|
||||
public TcpListener getListener() {
|
||||
return targetConnectionFactory.getListener();
|
||||
}
|
||||
|
||||
public TcpSender getSender() {
|
||||
return targetConnectionFactory.getSender();
|
||||
}
|
||||
|
||||
public Serializer<?> getSerializer() {
|
||||
return targetConnectionFactory.getSerializer();
|
||||
}
|
||||
|
||||
public Deserializer<?> getDeserializer() {
|
||||
return targetConnectionFactory.getDeserializer();
|
||||
}
|
||||
|
||||
public TcpMessageMapper getMapper() {
|
||||
return targetConnectionFactory.getMapper();
|
||||
}
|
||||
|
||||
public void registerListener(TcpListener listener) {
|
||||
targetConnectionFactory.registerListener(listener);
|
||||
}
|
||||
|
||||
public void registerSender(TcpSender sender) {
|
||||
targetConnectionFactory.registerSender(sender);
|
||||
}
|
||||
|
||||
public void setTaskExecutor(Executor taskExecutor) {
|
||||
targetConnectionFactory.setTaskExecutor(taskExecutor);
|
||||
}
|
||||
|
||||
public void setDeserializer(Deserializer<?> deserializer) {
|
||||
targetConnectionFactory.setDeserializer(deserializer);
|
||||
}
|
||||
|
||||
public void setSerializer(Serializer<?> serializer) {
|
||||
targetConnectionFactory.setSerializer(serializer);
|
||||
}
|
||||
|
||||
public void setMapper(TcpMessageMapper mapper) {
|
||||
targetConnectionFactory.setMapper(mapper);
|
||||
}
|
||||
|
||||
public boolean isSingleUse() {
|
||||
return targetConnectionFactory.isSingleUse();
|
||||
}
|
||||
|
||||
public void setSingleUse(boolean singleUse) {
|
||||
targetConnectionFactory.setSingleUse(singleUse);
|
||||
}
|
||||
|
||||
public void setInterceptorFactoryChain(
|
||||
TcpConnectionInterceptorFactoryChain interceptorFactoryChain) {
|
||||
targetConnectionFactory
|
||||
.setInterceptorFactoryChain(interceptorFactoryChain);
|
||||
}
|
||||
|
||||
public void setLookupHost(boolean lookupHost) {
|
||||
targetConnectionFactory.setLookupHost(lookupHost);
|
||||
}
|
||||
|
||||
public boolean isLookupHost() {
|
||||
return targetConnectionFactory.isLookupHost();
|
||||
}
|
||||
|
||||
public void start() {
|
||||
this.setActive(true);
|
||||
targetConnectionFactory.start();
|
||||
}
|
||||
|
||||
public synchronized void stop() {
|
||||
targetConnectionFactory.stop();
|
||||
this.pool.removeAllIdleItems();
|
||||
}
|
||||
|
||||
public int getPhase() {
|
||||
return targetConnectionFactory.getPhase();
|
||||
}
|
||||
|
||||
public boolean isAutoStartup() {
|
||||
return targetConnectionFactory.isAutoStartup();
|
||||
}
|
||||
|
||||
public void stop(Runnable callback) {
|
||||
targetConnectionFactory.stop(callback);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,46 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<beans xmlns="http://www.springframework.org/schema/beans"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xmlns:int-ip="http://www.springframework.org/schema/integration/ip"
|
||||
xmlns:int="http://www.springframework.org/schema/integration"
|
||||
xsi:schemaLocation="http://www.springframework.org/schema/integration/ip http://www.springframework.org/schema/integration/ip/spring-integration-ip-2.1.xsd
|
||||
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.1.xsd
|
||||
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
|
||||
|
||||
|
||||
<int-ip:tcp-connection-factory
|
||||
id="scf"
|
||||
type="server"
|
||||
so-timeout="60000"
|
||||
port="9876"/>
|
||||
|
||||
<int-ip:tcp-inbound-channel-adapter
|
||||
connection-factory="scf"
|
||||
channel="inbound"/>
|
||||
|
||||
<int:channel id="inbound">
|
||||
<int:queue/>
|
||||
</int:channel>
|
||||
|
||||
<int-ip:tcp-connection-factory
|
||||
id="ccf"
|
||||
type="client"
|
||||
host="localhost"
|
||||
port="9876"
|
||||
so-timeout="60000"
|
||||
|
||||
/>
|
||||
|
||||
<bean id="caching.ccf" class="org.springframework.integration.ip.tcp.connection.CachingClientConnectionFactory">
|
||||
<constructor-arg ref="ccf" />
|
||||
<constructor-arg value="10" />
|
||||
<property name="connectionWaitTimeout" value="10000"/>
|
||||
</bean>
|
||||
|
||||
<int-ip:tcp-outbound-channel-adapter
|
||||
connection-factory="caching.ccf"
|
||||
channel="outbound"/>
|
||||
|
||||
<int:channel id="outbound" />
|
||||
|
||||
</beans>
|
||||
@@ -0,0 +1,293 @@
|
||||
/*
|
||||
* Copyright 2002-2012 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.ip.tcp.connection;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
import static org.junit.Assert.assertNotSame;
|
||||
import static org.junit.Assert.assertSame;
|
||||
import static org.junit.Assert.fail;
|
||||
import static org.mockito.Mockito.doAnswer;
|
||||
import static org.mockito.Mockito.doThrow;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.times;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
import java.util.concurrent.Semaphore;
|
||||
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.mockito.invocation.InvocationOnMock;
|
||||
import org.mockito.stubbing.Answer;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.integration.Message;
|
||||
import org.springframework.integration.MessagingException;
|
||||
import org.springframework.integration.core.PollableChannel;
|
||||
import org.springframework.integration.core.SubscribableChannel;
|
||||
import org.springframework.integration.ip.IpHeaders;
|
||||
import org.springframework.integration.message.GenericMessage;
|
||||
import org.springframework.integration.test.util.TestUtils;
|
||||
import org.springframework.test.annotation.ExpectedException;
|
||||
import org.springframework.test.context.ContextConfiguration;
|
||||
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
|
||||
|
||||
/**
|
||||
* @author Gary Russell
|
||||
* @since 2.2
|
||||
*
|
||||
*/
|
||||
@ContextConfiguration
|
||||
@RunWith(SpringJUnit4ClassRunner.class)
|
||||
public class CachingClientConnectionFactoryTests {
|
||||
|
||||
@Autowired
|
||||
SubscribableChannel outbound;
|
||||
|
||||
@Autowired
|
||||
PollableChannel inbound;
|
||||
|
||||
@Autowired
|
||||
AbstractServerConnectionFactory serverCf;
|
||||
|
||||
@Test
|
||||
public void testReuse() throws Exception {
|
||||
AbstractClientConnectionFactory factory = mock(AbstractClientConnectionFactory.class);
|
||||
when(factory.isRunning()).thenReturn(true);
|
||||
TcpConnection mockConn1 = makeMockConnection("conn1");
|
||||
TcpConnection mockConn2 = makeMockConnection("conn2");
|
||||
when(factory.getConnection()).thenReturn(mockConn1).thenReturn(mockConn2);
|
||||
CachingClientConnectionFactory cachingFactory = new CachingClientConnectionFactory(factory, 2);
|
||||
cachingFactory.start();
|
||||
TcpConnection conn1 = cachingFactory.getConnection();
|
||||
assertEquals("Cached:" + mockConn1.toString(), conn1.toString());
|
||||
conn1.close();
|
||||
conn1 = cachingFactory.getConnection();
|
||||
assertEquals("Cached:" + mockConn1.toString(), conn1.toString());
|
||||
TcpConnection conn2 = cachingFactory.getConnection();
|
||||
assertEquals("Cached:" + mockConn2.toString(), conn2.toString());
|
||||
conn1.close();
|
||||
conn2.close();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testReuseNoLimit() throws Exception {
|
||||
AbstractClientConnectionFactory factory = mock(AbstractClientConnectionFactory.class);
|
||||
when(factory.isRunning()).thenReturn(true);
|
||||
TcpConnection mockConn1 = makeMockConnection("conn1");
|
||||
TcpConnection mockConn2 = makeMockConnection("conn2");
|
||||
when(factory.getConnection()).thenReturn(mockConn1).thenReturn(mockConn2);
|
||||
CachingClientConnectionFactory cachingFactory = new CachingClientConnectionFactory(factory, 0);
|
||||
cachingFactory.start();
|
||||
TcpConnection conn1 = cachingFactory.getConnection();
|
||||
assertEquals("Cached:" + mockConn1.toString(), conn1.toString());
|
||||
conn1.close();
|
||||
conn1 = cachingFactory.getConnection();
|
||||
assertEquals("Cached:" + mockConn1.toString(), conn1.toString());
|
||||
TcpConnection conn2 = cachingFactory.getConnection();
|
||||
assertEquals("Cached:" + mockConn2.toString(), conn2.toString());
|
||||
conn1.close();
|
||||
conn2.close();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testReuseClosed() throws Exception {
|
||||
AbstractClientConnectionFactory factory = mock(AbstractClientConnectionFactory.class);
|
||||
when(factory.isRunning()).thenReturn(true);
|
||||
TcpConnection mockConn1 = makeMockConnection("conn1");
|
||||
TcpConnection mockConn2 = makeMockConnection("conn2");
|
||||
when(factory.getConnection()).thenReturn(mockConn1)
|
||||
.thenReturn(mockConn2).thenReturn(mockConn1)
|
||||
.thenReturn(mockConn2);
|
||||
CachingClientConnectionFactory cachingFactory = new CachingClientConnectionFactory(factory, 2);
|
||||
cachingFactory.start();
|
||||
TcpConnection conn1 = cachingFactory.getConnection();
|
||||
assertEquals("Cached:" + mockConn1.toString(), conn1.toString());
|
||||
conn1.close();
|
||||
conn1 = cachingFactory.getConnection();
|
||||
assertEquals("Cached:" + mockConn1.toString(), conn1.toString());
|
||||
TcpConnection conn2 = cachingFactory.getConnection();
|
||||
assertEquals("Cached:" + mockConn2.toString(), conn2.toString());
|
||||
conn1.close();
|
||||
conn2.close();
|
||||
when(mockConn1.isOpen()).thenReturn(false);
|
||||
TcpConnection conn2a = cachingFactory.getConnection();
|
||||
assertEquals("Cached:" + mockConn2.toString(), conn2a.toString());
|
||||
assertSame(TestUtils.getPropertyValue(conn2, "theConnection"),
|
||||
TestUtils.getPropertyValue(conn2a, "theConnection"));
|
||||
conn2a.close();
|
||||
}
|
||||
|
||||
@Test @ExpectedException(MessagingException.class)
|
||||
public void testLimit() throws Exception {
|
||||
AbstractClientConnectionFactory factory = mock(AbstractClientConnectionFactory.class);
|
||||
when(factory.isRunning()).thenReturn(true);
|
||||
TcpConnection mockConn1 = makeMockConnection("conn1");
|
||||
TcpConnection mockConn2 = makeMockConnection("conn2");
|
||||
when(factory.getConnection()).thenReturn(mockConn1).thenReturn(mockConn2);
|
||||
CachingClientConnectionFactory cachingFactory = new CachingClientConnectionFactory(factory, 2);
|
||||
cachingFactory.setConnectionWaitTimeout(10);
|
||||
cachingFactory.start();
|
||||
TcpConnection conn1 = cachingFactory.getConnection();
|
||||
assertEquals("Cached:" + mockConn1.toString(), conn1.toString());
|
||||
conn1.close();
|
||||
conn1 = cachingFactory.getConnection();
|
||||
assertEquals("Cached:" + mockConn1.toString(), conn1.toString());
|
||||
TcpConnection conn2 = cachingFactory.getConnection();
|
||||
assertEquals("Cached:" + mockConn2.toString(), conn2.toString());
|
||||
cachingFactory.getConnection();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testStop() throws Exception {
|
||||
AbstractClientConnectionFactory factory = mock(AbstractClientConnectionFactory.class);
|
||||
when(factory.isRunning()).thenReturn(true);
|
||||
TcpConnection mockConn1 = makeMockConnection("conn1");
|
||||
TcpConnection mockConn2 = makeMockConnection("conn2");
|
||||
int i = 3;
|
||||
when(factory.getConnection()).thenReturn(mockConn1)
|
||||
.thenReturn(mockConn2)
|
||||
.thenReturn(makeMockConnection("conn" + (i++)));
|
||||
CachingClientConnectionFactory cachingFactory = new CachingClientConnectionFactory(factory, 2);
|
||||
cachingFactory.start();
|
||||
TcpConnection conn1 = cachingFactory.getConnection();
|
||||
assertEquals("Cached:" + mockConn1.toString(), conn1.toString());
|
||||
conn1.close();
|
||||
conn1 = cachingFactory.getConnection();
|
||||
assertEquals("Cached:" + mockConn1.toString(), conn1.toString());
|
||||
TcpConnection conn2 = cachingFactory.getConnection();
|
||||
assertEquals("Cached:" + mockConn2.toString(), conn2.toString());
|
||||
cachingFactory.stop();
|
||||
Answer<Object> answer = new Answer<Object> () {
|
||||
public Object answer(InvocationOnMock invocation) throws Throwable {
|
||||
return null;
|
||||
}};
|
||||
doAnswer(answer).when(mockConn1).close();
|
||||
doAnswer(answer).when(mockConn2).close();
|
||||
when(factory.isRunning()).thenReturn(false);
|
||||
conn1.close();
|
||||
conn2.close();
|
||||
verify(mockConn1).close();
|
||||
verify(mockConn2).close();
|
||||
when(factory.isRunning()).thenReturn(true);
|
||||
TcpConnection conn3 = cachingFactory.getConnection();
|
||||
assertNotSame(TestUtils.getPropertyValue(conn1, "theConnection"),
|
||||
TestUtils.getPropertyValue(conn3, "theConnection"));
|
||||
assertNotSame(TestUtils.getPropertyValue(conn2, "theConnection"),
|
||||
TestUtils.getPropertyValue(conn3, "theConnection"));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testEnlargePool() throws Exception {
|
||||
AbstractClientConnectionFactory factory = mock(AbstractClientConnectionFactory.class);
|
||||
when(factory.isRunning()).thenReturn(true);
|
||||
TcpConnection mockConn = makeMockConnection("conn");
|
||||
when(factory.getConnection()).thenReturn(mockConn);
|
||||
CachingClientConnectionFactory cachingFactory = new CachingClientConnectionFactory(factory, 2);
|
||||
cachingFactory.start();
|
||||
TcpConnection conn1 = cachingFactory.getConnection();
|
||||
TcpConnection conn2 = cachingFactory.getConnection();
|
||||
assertNotSame(conn1, conn2);
|
||||
Semaphore semaphore = TestUtils.getPropertyValue(
|
||||
TestUtils.getPropertyValue(cachingFactory, "pool"), "permits", Semaphore.class);
|
||||
assertEquals(0, semaphore.availablePermits());
|
||||
cachingFactory.setPoolSize(4);
|
||||
TcpConnection conn3 = cachingFactory.getConnection();
|
||||
TcpConnection conn4 = cachingFactory.getConnection();
|
||||
assertEquals(0, semaphore.availablePermits());
|
||||
conn1.close();
|
||||
conn1.close();
|
||||
conn2.close();
|
||||
conn3.close();
|
||||
conn4.close();
|
||||
assertEquals(4, semaphore.availablePermits());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testReducePool() throws Exception {
|
||||
AbstractClientConnectionFactory factory = mock(AbstractClientConnectionFactory.class);
|
||||
when(factory.isRunning()).thenReturn(true);
|
||||
TcpConnection mockConn1 = makeMockConnection("conn", true);
|
||||
TcpConnection mockConn2 = makeMockConnection("conn", true);
|
||||
TcpConnection mockConn3 = makeMockConnection("conn", true);
|
||||
TcpConnection mockConn4 = makeMockConnection("conn", true);
|
||||
when(factory.getConnection()).thenReturn(mockConn1)
|
||||
.thenReturn(mockConn2).thenReturn(mockConn3)
|
||||
.thenReturn(mockConn4);
|
||||
CachingClientConnectionFactory cachingFactory = new CachingClientConnectionFactory(factory, 4);
|
||||
cachingFactory.start();
|
||||
TcpConnection conn1 = cachingFactory.getConnection();
|
||||
TcpConnection conn2 = cachingFactory.getConnection();
|
||||
TcpConnection conn3 = cachingFactory.getConnection();
|
||||
TcpConnection conn4 = cachingFactory.getConnection();
|
||||
Semaphore semaphore = TestUtils.getPropertyValue(
|
||||
TestUtils.getPropertyValue(cachingFactory, "pool"), "permits", Semaphore.class);
|
||||
assertEquals(0, semaphore.availablePermits());
|
||||
conn1.close();
|
||||
assertEquals(1, semaphore.availablePermits());
|
||||
cachingFactory.setPoolSize(2);
|
||||
assertEquals(0, semaphore.availablePermits());
|
||||
assertEquals(3, cachingFactory.getActiveCount());
|
||||
conn2.close();
|
||||
assertEquals(0, semaphore.availablePermits());
|
||||
assertEquals(2, cachingFactory.getActiveCount());
|
||||
conn3.close();
|
||||
assertEquals(1, cachingFactory.getActiveCount());
|
||||
assertEquals(1, cachingFactory.getIdleCount());
|
||||
conn4.close();
|
||||
assertEquals(2, semaphore.availablePermits());
|
||||
assertEquals(0, cachingFactory.getActiveCount());
|
||||
assertEquals(2, cachingFactory.getIdleCount());
|
||||
verify(mockConn1).close();
|
||||
verify(mockConn2).close();
|
||||
}
|
||||
|
||||
private TcpConnection makeMockConnection(String name) {
|
||||
return makeMockConnection(name, false);
|
||||
}
|
||||
|
||||
private TcpConnection makeMockConnection(String name, boolean closeOk) {
|
||||
TcpConnection mockConn1 = mock(TcpConnection.class);
|
||||
when(mockConn1.getConnectionId()).thenReturn(name);
|
||||
when(mockConn1.toString()).thenReturn(name);
|
||||
when(mockConn1.isOpen()).thenReturn(true);
|
||||
if (!closeOk) {
|
||||
doThrow(new RuntimeException("close() not expected")).when(mockConn1).close();
|
||||
}
|
||||
return mockConn1;
|
||||
}
|
||||
|
||||
@Test
|
||||
public void integrationTest() throws Exception {
|
||||
int n = 0;
|
||||
while (!serverCf.isListening()) {
|
||||
Thread.sleep(100);
|
||||
n++;
|
||||
if (n > 10000) {
|
||||
fail("Server didn't begin listening");
|
||||
}
|
||||
}
|
||||
outbound.send(new GenericMessage<String>("Hello, world!"));
|
||||
Message<?> m = inbound.receive(1000);
|
||||
assertNotNull(m);
|
||||
String connectionId = m.getHeaders().get(IpHeaders.CONNECTION_ID, String.class);
|
||||
outbound.send(new GenericMessage<String>("Hello, world!"));
|
||||
m = inbound.receive(1000);
|
||||
assertNotNull(m);
|
||||
assertEquals(connectionId, m.getHeaders().get(IpHeaders.CONNECTION_ID, String.class));
|
||||
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user