/**
 * 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.
 */

package org.apache.cassandra.net;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.locks.*;

import org.apache.log4j.Logger;

class TcpConnectionManager
{
    private Lock lock_ = new ReentrantLock();
    private List<TcpConnection> allConnections_;
    private EndPoint localEp_;
    private EndPoint remoteEp_;
    private int initialSize_;
    private int growthFactor_;
    private int maxSize_;
    private long lastTimeUsed_;
    private boolean isShut_;

    private int inUse_;

    TcpConnectionManager(int initialSize, int growthFactor, int maxSize, EndPoint localEp, EndPoint remoteEp)
    {
        initialSize_ = initialSize;
        growthFactor_ = growthFactor;
        maxSize_ = maxSize;
        localEp_ = localEp;
        remoteEp_ = remoteEp;
        isShut_ = false;
        lastTimeUsed_ = System.currentTimeMillis();
        allConnections_ = new ArrayList<TcpConnection>();
    }

    TcpConnection getConnection() throws IOException
    {
        lock_.lock();
        try
        {
            if (allConnections_.isEmpty())
            {
                TcpConnection conn = new TcpConnection(this, localEp_, remoteEp_);
                addToPool(conn);
                conn.inUse_ = true;
                incUsed();
                return conn;
            }

            TcpConnection least = getLeastLoaded();

            if ( (least != null && least.pending() == 0) || allConnections_.size() == maxSize_) {
                least.inUse_ = true;
                incUsed();
                return least;
            }

            TcpConnection connection = new TcpConnection(this, localEp_, remoteEp_);
            if ( connection != null && !contains(connection) )
            {
                addToPool(connection);
                connection.inUse_ = true;
                incUsed();
                return connection;
            }
            else
            {
                if ( connection != null )
                {
                    connection.closeSocket();
                }
                return getLeastLoaded();
            }
        }
        finally
        {
            lock_.unlock();
        }
    }

    protected TcpConnection getLeastLoaded()
    {
        TcpConnection connection = null;
        lock_.lock();
        try
        {
            Collections.sort(allConnections_);
            connection = (allConnections_.size() > 0 ) ? allConnections_.get(0) : null;
        }
        finally
        {
            lock_.unlock();
        }
        return connection;
    }

    void removeConnection(TcpConnection connection)
    {
        lock_.lock();
        try
        {
            allConnections_.remove(connection);
        }
        finally
        {
            lock_.unlock();
        }
    }

    void incUsed()
    {
        inUse_++;
    }

    void decUsed()
    {
        inUse_--;
    }

    int getConnectionsInUse()
    {
        return inUse_;
    }

    void addToPool(TcpConnection connection)
    {
        lock_.lock();
        try
        {
            if ( contains(connection) )
                return;

            if ( allConnections_.size() < maxSize_ )
            {
                allConnections_.add(connection);
            }
            else
            {
                connection.closeSocket();
            }
        }
        finally
        {
            lock_.unlock();
        }
    }

    void shutdown()
    {
        lock_.lock();
        try
        {
            while ( allConnections_.size() > 0 )
            {
                TcpConnection connection = allConnections_.remove(0);
                connection.closeSocket();
            }
        }
        finally
        {
            lock_.unlock();
        }
        isShut_ = true;
    }

    int getPoolSize()
    {
        lock_.lock();
        try
        {
            return allConnections_.size();
        }
        finally
        {
            lock_.unlock();
        }
    }

    EndPoint getLocalEndPoint()
    {
        return localEp_;
    }

    EndPoint getRemoteEndPoint()
    {
        return remoteEp_;
    }

    int getPendingWrites()
    {
        int total = 0;
        lock_.lock();
        try
        {
            for ( TcpConnection connection : allConnections_ )
            {
                total += connection.pending();
            }
        }
        finally
        {
            lock_.unlock();
        }
        return total;
    }

    boolean contains(TcpConnection connection)
    {
        lock_.lock();
        try
        {
            return allConnections_.contains(connection);
        }
        finally
        {
            lock_.unlock();
        }
    }
}

@@ -2186,17 +2186,16 @@
     if (
- 
 (least !
@@ -2264,16 +2264,28 @@
 axSize_)
+%0A           
  %7B%0A     
@@ -2481,33 +2481,32 @@
             if (
- 
 connection != nu
@@ -2524,33 +2524,32 @@
 ains(connection)
- 
 )%0A            %7B%0A
@@ -2756,17 +2756,16 @@
     if (
- 
 connecti
@@ -2774,17 +2774,16 @@
  != null
- 
 )%0A      
@@ -3229,17 +3229,16 @@
 ze() %3E 0
- 
 ) ? allC
@@ -3880,17 +3880,16 @@
     if (
- 
 contains
@@ -3900,17 +3900,16 @@
 nection)
- 
 )%0A      
@@ -3943,17 +3943,16 @@
     if (
- 
 allConne
@@ -3976,17 +3976,16 @@
 maxSize_
- 
 )%0A      
@@ -4312,17 +4312,16 @@
  while (
- 
 allConne
@@ -4338,17 +4338,16 @@
 ze() %3E 0
- 
 )%0A      
@@ -5040,17 +5040,16 @@
    for (
- 
 TcpConne
@@ -5082,17 +5082,16 @@
 ections_
- 
 )%0A      

reformat whitespace. patch by jbellis for CASSANDRA-488