/*
* ====================================================================
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you 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.
* ====================================================================
*
* This software consists of voluntary contributions made by many
* individuals on behalf of the Apache Software Foundation. For more
* information on the Apache Software Foundation, please see
* MultiThreadedHttpConnectionManager
* in HttpClient 3.x, see there for original authors. It implements the same
* algorithm for connection re-use and connection-per-host enforcement:
*
synchronized
methods.
*
* @since 4.0
*
* @deprecated (4.2) use {@link ch.boye.httpclientandroidlib.pool.AbstractConnPool}
*/
@Deprecated
public class ConnPoolByRoute extends AbstractConnPool {
public HttpClientAndroidLog log = new HttpClientAndroidLog(getClass());
private final Lock poolLock;
/** Connection operator for this pool */
protected final ClientConnectionOperator operator;
/** Connections per route lookup */
protected final ConnPerRoute connPerRoute;
/** References to issued connections */
protected final Setnull
*
* @return a waiting thread representation
*/
protected WaitingThread newWaitingThread(final Condition cond,
final RouteSpecificPool rospl) {
return new WaitingThread(cond, rospl);
}
private void closeConnection(final BasicPoolEntry entry) {
final OperatedClientConnection conn = entry.getConnection();
if (conn != null) {
try {
conn.close();
} catch (final IOException ex) {
log.debug("I/O error closing connection", ex);
}
}
}
/**
* Get a route-specific pool of available connections.
*
* @param route the route
* @param create whether to create the pool if it doesn't exist
*
* @return the pool for the argument route,
* never null
if create
is true
*/
protected RouteSpecificPool getRoutePool(final HttpRoute route,
final boolean create) {
RouteSpecificPool rospl = null;
poolLock.lock();
try {
rospl = routeToPool.get(route);
if ((rospl == null) && create) {
// no pool for this route yet (or anymore)
rospl = newRouteSpecificPool(route);
routeToPool.put(route, rospl);
}
} finally {
poolLock.unlock();
}
return rospl;
}
public int getConnectionsInPool(final HttpRoute route) {
poolLock.lock();
try {
// don't allow a pool to be created here!
final RouteSpecificPool rospl = getRoutePool(route, false);
return (rospl != null) ? rospl.getEntryCount() : 0;
} finally {
poolLock.unlock();
}
}
public int getConnectionsInPool() {
poolLock.lock();
try {
return numConnections;
} finally {
poolLock.unlock();
}
}
@Override
public PoolEntryRequest requestPoolEntry(
final HttpRoute route,
final Object state) {
final WaitingThreadAborter aborter = new WaitingThreadAborter();
return new PoolEntryRequest() {
public void abortRequest() {
poolLock.lock();
try {
aborter.abort();
} finally {
poolLock.unlock();
}
}
public BasicPoolEntry getPoolEntry(
final long timeout,
final TimeUnit tunit)
throws InterruptedException, ConnectionPoolTimeoutException {
return getEntryBlocking(route, state, timeout, tunit, aborter);
}
};
}
/**
* Obtains a pool entry with a connection within the given timeout.
* If a {@link WaitingThread} is used to block, {@link WaitingThreadAborter#setWaitingThread(WaitingThread)}
* must be called before blocking, to allow the thread to be interrupted.
*
* @param route the route for which to get the connection
* @param timeout the timeout, 0 or negative for no timeout
* @param tunit the unit for the timeout
,
* may be null
only if there is no timeout
* @param aborter an object which can abort a {@link WaitingThread}.
*
* @return pool entry holding a connection for the route
*
* @throws ConnectionPoolTimeoutException
* if the timeout expired
* @throws InterruptedException
* if the calling thread was interrupted
*/
protected BasicPoolEntry getEntryBlocking(
final HttpRoute route, final Object state,
final long timeout, final TimeUnit tunit,
final WaitingThreadAborter aborter)
throws ConnectionPoolTimeoutException, InterruptedException {
Date deadline = null;
if (timeout > 0) {
deadline = new Date
(System.currentTimeMillis() + tunit.toMillis(timeout));
}
BasicPoolEntry entry = null;
poolLock.lock();
try {
RouteSpecificPool rospl = getRoutePool(route, true);
WaitingThread waitingThread = null;
while (entry == null) {
Asserts.check(!shutdown, "Connection pool shut down");
if (log.isDebugEnabled()) {
log.debug("[" + route + "] total kept alive: " + freeConnections.size() +
", total issued: " + leasedConnections.size() +
", total allocated: " + numConnections + " out of " + maxTotalConnections);
}
// the cases to check for:
// - have a free connection for that route
// - allowed to create a free connection for that route
// - can delete and replace a free connection for another route
// - need to wait for one of the things above to come true
entry = getFreeEntry(rospl, state);
if (entry != null) {
break;
}
final boolean hasCapacity = rospl.getCapacity() > 0;
if (log.isDebugEnabled()) {
log.debug("Available capacity: " + rospl.getCapacity()
+ " out of " + rospl.getMaxEntries()
+ " [" + route + "][" + state + "]");
}
if (hasCapacity && numConnections < maxTotalConnections) {
entry = createEntry(rospl, operator);
} else if (hasCapacity && !freeConnections.isEmpty()) {
deleteLeastUsedEntry();
// if least used entry's route was the same as rospl,
// rospl is now out of date : we preemptively refresh
rospl = getRoutePool(route, true);
entry = createEntry(rospl, operator);
} else {
if (log.isDebugEnabled()) {
log.debug("Need to wait for connection" +
" [" + route + "][" + state + "]");
}
if (waitingThread == null) {
waitingThread =
newWaitingThread(poolLock.newCondition(), rospl);
aborter.setWaitingThread(waitingThread);
}
boolean success = false;
try {
rospl.queueThread(waitingThread);
waitingThreads.add(waitingThread);
success = waitingThread.await(deadline);
} finally {
// In case of 'success', we were woken up by the
// connection pool and should now have a connection
// waiting for us, or else we're shutting down.
// Just continue in the loop, both cases are checked.
rospl.removeThread(waitingThread);
waitingThreads.remove(waitingThread);
}
// check for spurious wakeup vs. timeout
if (!success && (deadline != null) &&
(deadline.getTime() <= System.currentTimeMillis())) {
throw new ConnectionPoolTimeoutException
("Timeout waiting for connection from pool");
}
}
} // while no entry
} finally {
poolLock.unlock();
}
return entry;
}
@Override
public void freeEntry(final BasicPoolEntry entry, final boolean reusable, final long validDuration, final TimeUnit timeUnit) {
final HttpRoute route = entry.getPlannedRoute();
if (log.isDebugEnabled()) {
log.debug("Releasing connection" +
" [" + route + "][" + entry.getState() + "]");
}
poolLock.lock();
try {
if (shutdown) {
// the pool is shut down, release the
// connection's resources and get out of here
closeConnection(entry);
return;
}
// no longer issued, we keep a hard reference now
leasedConnections.remove(entry);
final RouteSpecificPool rospl = getRoutePool(route, true);
if (reusable && rospl.getCapacity() >= 0) {
if (log.isDebugEnabled()) {
final String s;
if (validDuration > 0) {
s = "for " + validDuration + " " + timeUnit;
} else {
s = "indefinitely";
}
log.debug("Pooling connection" +
" [" + route + "][" + entry.getState() + "]; keep alive " + s);
}
rospl.freeEntry(entry);
entry.updateExpiry(validDuration, timeUnit);
freeConnections.add(entry);
} else {
closeConnection(entry);
rospl.dropEntry();
numConnections--;
}
notifyWaitingThread(rospl);
} finally {
poolLock.unlock();
}
}
/**
* If available, get a free pool entry for a route.
*
* @param rospl the route-specific pool from which to get an entry
*
* @return an available pool entry for the given route, or
* null
if none is available
*/
protected BasicPoolEntry getFreeEntry(final RouteSpecificPool rospl, final Object state) {
BasicPoolEntry entry = null;
poolLock.lock();
try {
boolean done = false;
while(!done) {
entry = rospl.allocEntry(state);
if (entry != null) {
if (log.isDebugEnabled()) {
log.debug("Getting free connection"
+ " [" + rospl.getRoute() + "][" + state + "]");
}
freeConnections.remove(entry);
if (entry.isExpired(System.currentTimeMillis())) {
// If the free entry isn't valid anymore, get rid of it
// and loop to find another one that might be valid.
if (log.isDebugEnabled()) {
log.debug("Closing expired free connection"
+ " [" + rospl.getRoute() + "][" + state + "]");
}
closeConnection(entry);
// We use dropEntry instead of deleteEntry because the entry
// is no longer "free" (we just allocated it), and deleteEntry
// can only be used to delete free entries.
rospl.dropEntry();
numConnections--;
} else {
leasedConnections.add(entry);
done = true;
}
} else {
done = true;
if (log.isDebugEnabled()) {
log.debug("No free connections"
+ " [" + rospl.getRoute() + "][" + state + "]");
}
}
}
} finally {
poolLock.unlock();
}
return entry;
}
/**
* Creates a new pool entry.
* This method assumes that the new connection will be handed
* out immediately.
*
* @param rospl the route-specific pool for which to create the entry
* @param op the operator for creating a connection
*
* @return the new pool entry for a new connection
*/
protected BasicPoolEntry createEntry(final RouteSpecificPool rospl,
final ClientConnectionOperator op) {
if (log.isDebugEnabled()) {
log.debug("Creating new connection [" + rospl.getRoute() + "]");
}
// the entry will create the connection when needed
final BasicPoolEntry entry = new BasicPoolEntry(op, rospl.getRoute(), connTTL, connTTLTimeUnit);
poolLock.lock();
try {
rospl.createdEntry(entry);
numConnections++;
leasedConnections.add(entry);
} finally {
poolLock.unlock();
}
return entry;
}
/**
* Deletes a given pool entry.
* This closes the pooled connection and removes all references,
* so that it can be GCed.
*
* Note: Does not remove the entry from the freeConnections list. * It is assumed that the caller has already handled this step.
* * * @param entry the pool entry for the connection to delete */ protected void deleteEntry(final BasicPoolEntry entry) { final HttpRoute route = entry.getPlannedRoute(); if (log.isDebugEnabled()) { log.debug("Deleting connection" + " [" + route + "][" + entry.getState() + "]"); } poolLock.lock(); try { closeConnection(entry); final RouteSpecificPool rospl = getRoutePool(route, true); rospl.deleteEntry(entry); numConnections--; if (rospl.isUnused()) { routeToPool.remove(route); } } finally { poolLock.unlock(); } } /** * Delete an old, free pool entry to make room for a new one. * Used to replace pool entries with ones for a different route. */ protected void deleteLeastUsedEntry() { poolLock.lock(); try { final BasicPoolEntry entry = freeConnections.remove(); if (entry != null) { deleteEntry(entry); } else if (log.isDebugEnabled()) { log.debug("No free connection to delete"); } } finally { poolLock.unlock(); } } @Override protected void handleLostEntry(final HttpRoute route) { poolLock.lock(); try { final RouteSpecificPool rospl = getRoutePool(route, true); rospl.dropEntry(); if (rospl.isUnused()) { routeToPool.remove(route); } numConnections--; notifyWaitingThread(rospl); } finally { poolLock.unlock(); } } /** * Notifies a waiting thread that a connection is available. * This will wake a thread waiting in the specific route pool, * if there is one. * Otherwise, a thread in the connection pool will be notified. * * @param rospl the pool in which to notify, ornull
*/
protected void notifyWaitingThread(final RouteSpecificPool rospl) {
//@@@ while this strategy provides for best connection re-use,
//@@@ is it fair? only do this if the connection is open?
// Find the thread we are going to notify. We want to ensure that
// each waiting thread is only interrupted once, so we will remove
// it from all wait queues before interrupting.
WaitingThread waitingThread = null;
poolLock.lock();
try {
if ((rospl != null) && rospl.hasThread()) {
if (log.isDebugEnabled()) {
log.debug("Notifying thread waiting on pool" +
" [" + rospl.getRoute() + "]");
}
waitingThread = rospl.nextThread();
} else if (!waitingThreads.isEmpty()) {
if (log.isDebugEnabled()) {
log.debug("Notifying thread waiting on any pool");
}
waitingThread = waitingThreads.remove();
} else if (log.isDebugEnabled()) {
log.debug("Notifying no-one, there are no waiting threads");
}
if (waitingThread != null) {
waitingThread.wakeup();
}
} finally {
poolLock.unlock();
}
}
@Override
public void deleteClosedConnections() {
poolLock.lock();
try {
final Iteratoridletime
*/
@Override
public void closeIdleConnections(final long idletime, final TimeUnit tunit) {
Args.notNull(tunit, "Time unit");
final long t = idletime > 0 ? idletime : 0;
if (log.isDebugEnabled()) {
log.debug("Closing connections idle longer than " + t + " " + tunit);
}
// the latest time for which connections will be closed
final long deadline = System.currentTimeMillis() - tunit.toMillis(t);
poolLock.lock();
try {
final Iterator