path: root/mobile/android/thirdparty/ch/boye/httpclientandroidlib/impl/conn/tsccm/ConnPoolByRoute.java

/*
 * ====================================================================
 * 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
 * <http://www.apache.org/>.
 *
 */
package ch.boye.httpclientandroidlib.impl.conn.tsccm;

import java.io.IOException;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

import ch.boye.httpclientandroidlib.androidextra.HttpClientAndroidLog;
/* LogFactory removed by HttpClient for Android script. */
import ch.boye.httpclientandroidlib.conn.ClientConnectionOperator;
import ch.boye.httpclientandroidlib.conn.ConnectionPoolTimeoutException;
import ch.boye.httpclientandroidlib.conn.OperatedClientConnection;
import ch.boye.httpclientandroidlib.conn.params.ConnManagerParams;
import ch.boye.httpclientandroidlib.conn.params.ConnPerRoute;
import ch.boye.httpclientandroidlib.conn.routing.HttpRoute;
import ch.boye.httpclientandroidlib.params.HttpParams;
import ch.boye.httpclientandroidlib.util.Args;
import ch.boye.httpclientandroidlib.util.Asserts;

/**
 * A connection pool that maintains connections by route.
 * This class is derived from <code>MultiThreadedHttpConnectionManager</code>
 * in HttpClient 3.x, see there for original authors. It implements the same
 * algorithm for connection re-use and connection-per-host enforcement:
 * <ul>
 * <li>connections are re-used only for the exact same route</li>
 * <li>connection limits are enforced per route rather than per host</li>
 * </ul>
 * Note that access to the pool data structures is synchronized via the
 * {@link AbstractConnPool#poolLock poolLock} in the base class,
 * not via <code>synchronized</code> 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 Set<BasicPoolEntry> leasedConnections;

    /** The list of free connections */
    protected final Queue<BasicPoolEntry> freeConnections;

    /** The list of WaitingThreads waiting for a connection */
    protected final Queue<WaitingThread> waitingThreads;

    /** Map of route-specific pools */
    protected final Map<HttpRoute, RouteSpecificPool> routeToPool;

    private final long connTTL;

    private final TimeUnit connTTLTimeUnit;

    protected volatile boolean shutdown;

    protected volatile int maxTotalConnections;

    protected volatile int numConnections;

    /**
     * Creates a new connection pool, managed by route.
     *
     * @since 4.1
     */
    public ConnPoolByRoute(
            final ClientConnectionOperator operator,
            final ConnPerRoute connPerRoute,
            final int maxTotalConnections) {
        this(operator, connPerRoute, maxTotalConnections, -1, TimeUnit.MILLISECONDS);
    }

    /**
     * @since 4.1
     */
    public ConnPoolByRoute(
            final ClientConnectionOperator operator,
            final ConnPerRoute connPerRoute,
            final int maxTotalConnections,
            final long connTTL,
            final TimeUnit connTTLTimeUnit) {
        super();
        Args.notNull(operator, "Connection operator");
        Args.notNull(connPerRoute, "Connections per route");
        this.poolLock = super.poolLock;
        this.leasedConnections = super.leasedConnections;
        this.operator = operator;
        this.connPerRoute = connPerRoute;
        this.maxTotalConnections = maxTotalConnections;
        this.freeConnections = createFreeConnQueue();
        this.waitingThreads  = createWaitingThreadQueue();
        this.routeToPool     = createRouteToPoolMap();
        this.connTTL = connTTL;
        this.connTTLTimeUnit = connTTLTimeUnit;
    }

    protected Lock getLock() {
        return this.poolLock;
    }

    /**
     * Creates a new connection pool, managed by route.
     *
     * @deprecated (4.1)  use {@link ConnPoolByRoute#ConnPoolByRoute(ClientConnectionOperator, ConnPerRoute, int)}
     */
    @Deprecated
    public ConnPoolByRoute(final ClientConnectionOperator operator, final HttpParams params) {
        this(operator,
                ConnManagerParams.getMaxConnectionsPerRoute(params),
                ConnManagerParams.getMaxTotalConnections(params));
    }

    /**
     * Creates the queue for {@link #freeConnections}.
     * Called once by the constructor.
     *
     * @return  a queue
     */
    protected Queue<BasicPoolEntry> createFreeConnQueue() {
        return new LinkedList<BasicPoolEntry>();
    }

    /**
     * Creates the queue for {@link #waitingThreads}.
     * Called once by the constructor.
     *
     * @return  a queue
     */
    protected Queue<WaitingThread> createWaitingThreadQueue() {
        return new LinkedList<WaitingThread>();
    }

    /**
     * Creates the map for {@link #routeToPool}.
     * Called once by the constructor.
     *
     * @return  a map
     */
    protected Map<HttpRoute, RouteSpecificPool> createRouteToPoolMap() {
        return new HashMap<HttpRoute, RouteSpecificPool>();
    }


    /**
     * Creates a new route-specific pool.
     * Called by {@link #getRoutePool} when necessary.
     *
     * @param route     the route
     *
     * @return  the new pool
     */
    protected RouteSpecificPool newRouteSpecificPool(final HttpRoute route) {
        return new RouteSpecificPool(route, this.connPerRoute);
    }


    /**
     * Creates a new waiting thread.
     * Called by {@link #getRoutePool} when necessary.
     *
     * @param cond      the condition to wait for
     * @param rospl     the route specific pool, or <code>null</code>
     *
     * @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 <code>null</code> if <code>create</code> is <code>true</code>
     */
    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 <code>timeout</code>,
     *                  may be <code>null</code> 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
     *          <code>null</code> 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.
     *
     * <p><b>Note:</b> Does not remove the entry from the freeConnections list.
     * It is assumed that the caller has already handled this step.</p>
     * <!-- @@@ is that a good idea? or rather fix it? -->
     *
     * @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, or <code>null</code>
     */
    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 Iterator<BasicPoolEntry>  iter = freeConnections.iterator();
            while (iter.hasNext()) {
                final BasicPoolEntry entry = iter.next();
                if (!entry.getConnection().isOpen()) {
                    iter.remove();
                    deleteEntry(entry);
                }
            }
        } finally {
            poolLock.unlock();
        }
    }

    /**
     * Closes idle connections.
     *
     * @param idletime  the time the connections should have been idle
     *                  in order to be closed now
     * @param tunit     the unit for the <code>idletime</code>
     */
    @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<BasicPoolEntry>  iter = freeConnections.iterator();
            while (iter.hasNext()) {
                final BasicPoolEntry entry = iter.next();
                if (entry.getUpdated() <= deadline) {
                    if (log.isDebugEnabled()) {
                        log.debug("Closing connection last used @ " + new Date(entry.getUpdated()));
                    }
                    iter.remove();
                    deleteEntry(entry);
                }
            }
        } finally {
            poolLock.unlock();
        }
    }

    @Override
    public void closeExpiredConnections() {
        log.debug("Closing expired connections");
        final long now = System.currentTimeMillis();

        poolLock.lock();
        try {
            final Iterator<BasicPoolEntry>  iter = freeConnections.iterator();
            while (iter.hasNext()) {
                final BasicPoolEntry entry = iter.next();
                if (entry.isExpired(now)) {
                    if (log.isDebugEnabled()) {
                        log.debug("Closing connection expired @ " + new Date(entry.getExpiry()));
                    }
                    iter.remove();
                    deleteEntry(entry);
                }
            }
        } finally {
            poolLock.unlock();
        }
    }

    @Override
    public void shutdown() {
        poolLock.lock();
        try {
            if (shutdown) {
                return;
            }
            shutdown = true;

            // close all connections that are issued to an application
            final Iterator<BasicPoolEntry> iter1 = leasedConnections.iterator();
            while (iter1.hasNext()) {
                final BasicPoolEntry entry = iter1.next();
                iter1.remove();
                closeConnection(entry);
            }

            // close all free connections
            final Iterator<BasicPoolEntry> iter2 = freeConnections.iterator();
            while (iter2.hasNext()) {
                final BasicPoolEntry entry = iter2.next();
                iter2.remove();

                if (log.isDebugEnabled()) {
                    log.debug("Closing connection"
                            + " [" + entry.getPlannedRoute() + "][" + entry.getState() + "]");
                }
                closeConnection(entry);
            }

            // wake up all waiting threads
            final Iterator<WaitingThread> iwth = waitingThreads.iterator();
            while (iwth.hasNext()) {
                final WaitingThread waiter = iwth.next();
                iwth.remove();
                waiter.wakeup();
            }

            routeToPool.clear();

        } finally {
            poolLock.unlock();
        }
    }

    /**
     * since 4.1
     */
    public void setMaxTotalConnections(final int max) {
        poolLock.lock();
        try {
            maxTotalConnections = max;
        } finally {
            poolLock.unlock();
        }
    }


    /**
     * since 4.1
     */
    public int getMaxTotalConnections() {
        return maxTotalConnections;
    }

}