//______________________________________________________________________________
//   |       |
//   | knipp |           Knipp Medien und Kommunikation GmbH
//    -------
//
//
//     Knipp RRI Toolkit
//
//
//     Copyright (C) 2005 by Knipp Medien und Kommunikation GmbH
//
//     This library is free software; you can redistribute it and/or
//     modify it under the terms of the GNU Lesser General Public License 
//     as published by the Free Software Foundation; either version 2.1 
//     of the License, or any later version.
// 
//     This library is distributed in the hope that it will be useful, 
//     but WITHOUT ANY WARRANTY; without even the implied warranty of
//     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 
//     GNU Lesser General Public License for more details.
//
//     You should have received a copy of the GNU Lesser General
//     Public License along with this library; if not, write to the 
//     Free Software Foundation, Inc., 59 Temple Place, Suite 330, 
//     Boston, MA 02111-1307 USA 
//
//
//     initial author:    Klaus Malorny
//
//     version:           1.4 (15127), 2011/06/03 14:52:49
//                        $Revision: 14936 $ 
//                        $Date: 2010-11-05 15:57:40 +0100 (Fri, 05 Nov 2010) $
//                        $Author: Klaus.Malorny $
//

package de.knipp.rritoolkit;

import java.io.IOException;

import java.util.Queue;
import java.util.LinkedList;
import java.util.Iterator;

import de.knipp.rritoolkit.request.Request;
import de.knipp.rritoolkit.request.LogoutRequest;
import de.knipp.rritoolkit.request.LoginRequest;

import de.knipp.rritoolkit.response.Response;

import static de.knipp.rritoolkit.response.ErrorType.test;
import static de.knipp.rritoolkit.response.ErrorType.ErrUserOrPasswordWrong;
import static de.knipp.rritoolkit.response.ErrorType.FatalTooManyLogins;
import static de.knipp.rritoolkit.response.ErrorType.FatalNotLoggedIn;
import static de.knipp.rritoolkit.response.ErrorType.Other;





/**
 *  class that manages a pool of channels to the registry
 *
 *  @version  $Revision: 14936 $
 *  @author   Klaus Malorny
 */

public class PooledPort implements Port
{
  /** the codec we are using */
  
  private Codec codec;
  
  
  /** the channel factory */
  
  private ChannelFactory factory;
  
  
  /** the number of channels to allocate */
  
  private int allocateChannels;
  
  
  /** minimum number of channels to leave in the pool */
  
  private int minChannels;
  
  
  /** the maximum number of channels to allocate */
  
  private int maxChannels;

  
  /** user name for the login */
  
  private String userName;
  
  
  /** the password for the login */
  
  private String password;
  
  
  /** the object on which internal synchronisation is coordinated */
  
  private final Object syncObject;
  
  
  /** the number of currently allocated channels */
  
  private int channelCount;
  
  
  /** the pool of available channels */
  
  private Queue<RegistryChannel> available;
  
  
  /** the queue of waiting threads */
  
  private Queue<WaitEntry> waiting;
  
  
  /** the pool mananger */
  
  private PoolManager manager;

  
  /** the thread the manager is running on */
  
  private Thread managerThread;
  
  
  /** the shutdown flag */
  
  private boolean shutdown;
  
  
  
  
  /**
   *  constructor
   *
   *  @param theFactory               the channel factory
   *  @param theCodec                 the codec to use; if {@code null}
   *                                  a default codec is used.
   *  @param theUserName              the user name
   *  @param thePassword              the password
   *  @param allocate                 the number of connections that shall
   *                                  be allocated without request (should be
   *                                  less or equal to min)
   *  @param min                      the minium number of channels to keep in
   *                                  the pool
   *  @param max                      the maximum number of channels that
   *                                  shall be allocated
   */
  
  public PooledPort (ChannelFactory theFactory, Codec theCodec, 
    String theUserName, String thePassword, int allocate, int min, int max)
  {
    factory = theFactory;
    
    if (theCodec == null)
      codec = new Codec (DefaultResponseFactory.getFactory (), true);
     else
      codec = theCodec;
    
    userName = theUserName;
    password = thePassword;
    allocateChannels = allocate;
    minChannels = min;
    maxChannels = max;
    
    channelCount = 0;
    available = new LinkedList<RegistryChannel> ();
    waiting = new LinkedList<WaitEntry> ();
    syncObject = new Object ();
    
    manager = new PoolManager ();
    managerThread = new Thread (manager, "RRIToolkit Pool Manager");
    managerThread.setDaemon (true);
    managerThread.start ();
  }
  
  
  
  /**
   *  discard a channel
   *
   *  @param channel                  the channel to discard
   */
  
  private void discardChannel (RegistryChannel channel)
  {
    synchronized (syncObject)
    {
      // If there are waiting threads and the limit has been reached previously,
      // wakeup the pool manager so he can allocate a new channel
      
      boolean limitWasReached = channelCount == maxChannels;
      channelCount--;

      if (limitWasReached && !waiting.isEmpty ()) syncObject.notify ();
    }

  }
  
  
  
  /**
   *  return a channel to the pool or to a waiting thread
   *
   *  @param channel                  the channel to return
   */
  
  private void returnChannel (RegistryChannel channel)
  {
    if (channel.isClosed ())
    {
      discardChannel (channel);
      return;
    }
    
    // channel seems to be o.k. give it either the next waiting thread or
    // return it to the pool
    
    synchronized (syncObject)
    {
      if (waiting.isEmpty ())
      {
        available.add (channel);
        syncObject.notify ();   // checks for the pool size
      
      } else
      {
        WaitEntry we = waiting.remove ();
        
        synchronized (we)
        {
          we.wakeUp (channel);
        }
      }
    }
  }

  
  /**
   *  acquire a channel. If no channel is available, the thread enters a wait
   *  until a channel is returned or a new channel is allocated
   *
   *  @return                the channel
   *  @throws IOException    if the channel could not be allocated
   */
  
  private RegistryChannel acquireChannel ()
    throws IOException
  {
    WaitEntry we;
    
    synchronized (syncObject)
    {
      if (shutdown)
        throw new IOException ("port has been shut down");
      
      while (true)
      {
        if (available.isEmpty ()) break;
        RegistryChannel channel = available.remove ();
        
        if (channel.isClosed ())
        {
          discardChannel (channel);
          continue;
        }
        
        return channel;
      }
      
      // no channels available, wait
      
      we = new WaitEntry ();
      waiting.add (we);
      syncObject.notify ();
    }

    return we.waitForChannel ();
  }
  
  
  
  
  /**
   *  execute a request on the given channel
   *
   *  @param channel          the channel
   *  @param request          the request
   *  @return                 the response received
   *  @throws IOException     if either the transmission or the reception failed
   */
  
  private Response execute (RegistryChannel channel, Request request)
    throws IOException
  {
    try
    {
      byte[] data = codec.encodeRequest (request);

      try
      {
        channel.send (data, codec.xmlUsed ());
        
      } catch (IOException e)
      {
        throw new WriteException ("failed to send request to registry", e);
      }
      
      data = channel.receive ();

      return codec.decodeResponse (data, request);
      
    } catch (EncodeException e)
    {
      throw new IOException ("failed to encode the request", e);
      
    } catch (ResponseParseException e)
    {
      throw new IOException ("failed to decode the response", e);
    }
  }
  
  
  
  /**
   *  terminate the port and close all open channels. It is the duty
   *  of the caller to make sure that no other thread is currently a 
   *  channels at the point of the call.
   */
  
  public void shutdown ()
  {
    synchronized (syncObject)
    {
      if (shutdown) return;
      shutdown = true;
      syncObject.notify ();
    }
    
    // wait until manager terminates
    
    try
    {
      managerThread.join ();
      
    } catch (InterruptedException e)
    {
      // ignored
    }
  }
  
  
  //--- implementation of Port -------------------------------------------------
  
  
  /**
   *  "execute" a request, i.e. send it to the registry and wait for the
   *  response
   *
   *  @param request          the request
   *  @return                 the response received
   *  @throws IOException     if either the transmission or the reception failed
   */

  @Override
  public Response execute (Request request)
    throws IOException
  {
    RegistryChannel channel = null;
    
    
    try
    {
      while (true)
      {
        channel = acquireChannel ();

        try
        {
          return execute (channel, request);

        } catch (WriteException e)
        {
          // write failed, most likely due to a timed-out connection, get
          // new one from pool

          discardChannel (channel);
          channel = null;
          continue;
        }
      }
      
    } finally
    {
      if (channel != null) returnChannel (channel);
    }
  }


  
  /**
   *  thread that coordinates the pool, i.e. allocates and deallocates
   *  channels depending on the needs
   *
   *  @version  $Revision: 14936 $
   *  @author   Klaus Malorny
   */
  
  private class PoolManager implements Runnable
  {
    /** the point in time the next release test has to be performed */
    
    private long nextRelease;
    
    
    /** the point in time when the next channel may created (0 = immediately) */
    
    private long nextAllocate;
   
    
    /** the release period of channels over the minimum */
    
    private static final long RELEASE_PERIOD = 20000;  // 20 sec
    

    
    /** maximum time to check pool for closed connections */
    
    private static final long CHECK_POOL_PERIOD = 60000;  // 60 sec
    
    
    /** the timeout time if the maximum number of connections 
        has been reached */
    
    private static final long CREATE_REPEAT_WAIT = 10000;  // 10 sec
    
    
    
    /**
     *  try to create a new channel and add it to the pool
     *
     *  @return           {@code true} if further preallocated channels
     *                    may be created
     */

    private boolean createChannel ()
    {
      try
      {
        RegistryChannel channel = factory.createChannel ();        
        
        LoginRequest req = new LoginRequest ();
        req.setUser (userName);
        req.setPassword (password);
        Response resp = execute (channel, req);
        
        if (!resp.isSuccess ())
        {
          switch (test (resp, FatalTooManyLogins, FatalNotLoggedIn,
            ErrUserOrPasswordWrong))
          {
            case FatalTooManyLogins:
              // report problem if we don't have a single channel available
              
              boolean noChannels;
              
              synchronized (syncObject)
              {
                noChannels = channelCount == 0;
              }
              
              if (noChannels)
                throw new IOException ("too many channels used at registry, " +
                  "but none have been allocated by this pool");
              
              nextAllocate = System.currentTimeMillis () + CREATE_REPEAT_WAIT;
              channel.close ();
              return true;
              
            case ErrUserOrPasswordWrong:
            case FatalNotLoggedIn:
              channel.close ();
              throw new IOException ("wrong user name or password");
              
            case Other:
            default:
              channel.close ();
              throw new IOException ("login failed");
          }
        }
         
        synchronized (syncObject)
        {
          channelCount++;
        }

        returnChannel (channel);  // put the new connection to the pool
        return true;
        
      } catch (IOException e)
      {
        // forward the exception to the next waiting thread
        
        WaitEntry we;
        
        synchronized (syncObject)
        {
          we = waiting.poll ();
        }
        
        if (we != null)
        {
          we.wakeUp (e);
          
        } else
        {
          // log somewhere
          e.printStackTrace (System.err);
        }

        return false;
      }
    }

    
    
    /**
     *  release the given channel
     *
     *  @param channel               the channel to release
     */
    
    private void releaseChannel (RegistryChannel channel)
    {
      try
      {
        // try to log out
        
        if (!channel.isClosed ())
        {
          execute (channel, new LogoutRequest ());
          channel.close ();
        }
      } catch (IOException e)
      {
        // log somewhere
        e.printStackTrace (System.err);
      }
    }
    
    
    
    /**
     *  perform the duties
     */

    @Override
    public void run ()
    {
      boolean preallocate = true;
      
      while (true)
      {
        // see what needs to be done

        boolean release;
        boolean allocate;
        RegistryChannel releaseChannel = null;
        
        synchronized (syncObject)
        {
          if (shutdown) break;
          
          // housekeeping. Check for channels in the pool that have been closed
          
          Iterator<RegistryChannel> i = available.iterator ();
          
          while (i.hasNext ())
          {
            if (i.next ().isClosed ())
            {
              i.remove ();
              channelCount--;
            }
          }

          // determine which tasks need to be performed
          
          // are too many channels open?
          
          release = channelCount > allocateChannels && 
            available.size () > minChannels;
          
          // are more channels required?
          
          allocate = !waiting.isEmpty () && channelCount < maxChannels ||
            preallocate && channelCount < allocateChannels;
          
          // release & allocate should not happen at the same time, but
          // test anyway
          
          if (allocate) release = false;
          
          long waitTo = 0;

          // test whether the actions can be performed immediately or
          // have to be delayed
          
          long now = System.currentTimeMillis ();
          
          if (allocate && nextAllocate > now)
            waitTo = nextAllocate;

          if (release && nextRelease > now)
            waitTo = nextRelease;
          
          // cancel actions on wait
          
          if (waitTo != 0)
            allocate = release = false;
          
          // nothing to do, determine time of next check

          if (waitTo == 0 && !allocate && !release)
          {
            waitTo = now + CHECK_POOL_PERIOD;
          }
          
          if (waitTo > 0)
          {
            try
            {
              syncObject.wait (waitTo - now);
              
            } catch (InterruptedException e)
            {
              // should not happen
            }
          
          } else if (release)
          {
            // prepare release outside of the synchronized block
            
            releaseChannel = available.remove ();
            channelCount--;
          }
        }  
        

        if (allocate)
        {
          preallocate = createChannel ();
          
        } else if (release)
        {
          releaseChannel (releaseChannel);
          nextRelease = System.currentTimeMillis () + RELEASE_PERIOD;
        }
      }
      
      // perform shutdown
      
      synchronized (syncObject)
      {
        // if any open channels, close them
        
        for (RegistryChannel channel: available)
          releaseChannel (channel);
        
        available.clear ();
        
        // if any waiting threads, notify them

        for (WaitEntry we: waiting)
          we.wakeUp (new IOException ("port has been shut down"));
        
        waiting.clear ();
      }
    }
  }
  
  
  
  

  /**
   *  class that describes a waiting thread
   *
   *  @version  $Revision: 14936 $
   *  @author   Klaus Malorny
   */
  
  private static class WaitEntry
  {
    /** the assigned channel */
    
    private RegistryChannel channel;
    
    
    /** the exception to throw */
    
    private IOException ioException;
    
    
    /** flag whether result has been set */

    private boolean dataSet;
    
    
    
    /**
     *  wait for a connection
     *
     *  @return                 the channel
     *  @throws IOException     if the channel could not be assigned
     */
    
    public RegistryChannel waitForChannel ()
      throws IOException
    {
      synchronized (this)
      {
        if (!dataSet)
        {
          try
          {
            wait ();

          } catch (InterruptedException e)
          {
            // should not happen
            throw new IOException ("wait interrupted");
          }
        }
      }
      
      if (ioException != null) 
      {
        throw new IOException ("failed to acquire channel", ioException);
      }
      
      return channel;
    }
    
    
    
    /**
     *  wake up the waiting thread, giving it the specified channel
     *
     *  @param theChannel            the channel
     */
    
    public void wakeUp (RegistryChannel theChannel)
    {
      synchronized (this)
      {
        channel = theChannel;
        dataSet = true;
        notify ();
      }
    }
    
    
    
    /**
     *  wake up the waiting thread, causing it to throw the given exception
     *
     *  @param exception             the exception to throw
     */
    
    public void wakeUp (IOException exception)
    {
      synchronized (this)
      {
        ioException = exception;
        dataSet = true;
        notify ();
      }
    }
  }



  /**
   *  exception that specifically marks I/O exceptions that happen during
   *  the sending of the EPP command
   *
   *  @version  $Revision: 14936 $
   *  @author   Klaus Malorny
   */
  
  private static class WriteException extends IOException
  {
    /**
     *  constructor
     *
     *  @param msg                   the message
     *  @param cause                 the causing exception
     */
    
    public WriteException (String msg, IOException cause)
    {
      super (msg);
      initCause (cause);
    }
    
    
    /**
     *  constructor
     *
     *  @param msg                   the message
     */
    
    public WriteException (String msg)
    {
      this (msg, null);
    }
    
  }


}


