//______________________________________________________________________________
//   |       |
//   | 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.tcp;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.EOFException;

import java.net.Socket;

import java.util.SortedSet;
import java.util.TreeSet;

import de.knipp.rritoolkit.RegistryChannel;


/**
 *  implementation of the {@code RegistryChannel} interface for
 *  SSL TCP connections
 *
 *  @version  $Revision: 14936 $
 *  @author   Klaus Malorny
 */

public class TcpChannel implements RegistryChannel, Comparable<TcpChannel>
{
  /** the parameters */
  
  private TcpParams params;
  
  
  /** the socket */
  
  private Socket socket;
  
  
  /** the input stream */
  
  private final DataInputStream inStream;
  
  
  /** the output stream */
  
  private final DataOutputStream outStream;
  
  
  /** flag whether connection has been closed */
  
  private boolean closed;
  
  
  /** the idle timeout */

  private long idleTimeout;
  
  
  /** the time the connection shall be closed */
  
  private long idleCloseTime;
  
  
  /** an instance ID */
  
  private int instanceID;
  
  
  /** the number of connections ever allocated */
  
  private static int connectionCount;
  
  
  /** set of connections that are to be closed */
  
  private final static SortedSet<TcpChannel> pendingClose;
  
  
  /** the thread that closes timed-out channels */
  
  private static IdleMonitor monitor;
  
  
  /** the minimum time to remain until the channel is reported as closed */
  
  private static final long MIN_REMAIN_TIME = 2000L;
  
  
  /**
   *  class constructor
   */
  
  static
  {
    pendingClose = new TreeSet<TcpChannel> ();
  }
  
  
  
  /**
   *  constructor
   *
   *  @param theParams                the parameters
   *  @param theSocket                the socket
   *  @throws IOException             if the connection could not be established
   */
  
  public TcpChannel (TcpParams theParams, Socket theSocket)
    throws IOException
  {
    params = theParams;
    socket = theSocket;
    inStream = new DataInputStream (socket.getInputStream ());
    outStream = new DataOutputStream (socket.getOutputStream ());
    closed = false;
    
    synchronized (getClass ())
    {
      instanceID = ++connectionCount;
    }
    
    idleTimeout = params.getIdleTimeout ();
    
    if (idleTimeout > 0)
    { 
      checkMonitor ();
      setCloseTime (System.currentTimeMillis () + idleTimeout);
    }
  }
  
  
  
  /**
   *  check whether the idle monitor has been started. If not, it is started
   */
  
  private static synchronized void checkMonitor ()
  {
    if (monitor == null)
    {
      monitor = new IdleMonitor ();
      Thread t = new Thread (monitor, "TCP Channel Idle Monitor");
      t.setDaemon (true);
      t.start ();
    }
  }
  
  
  /**
   *  update the close time
   *
   *  @param closeTime               the new close time, {@code 0} to
   *                                 disable the automatic closing
   */
  
  private void setCloseTime (long closeTime)
  {
    synchronized (pendingClose)
    {
      boolean hasCloseTime = closeTime != 0;
      boolean hadCloseTime = idleCloseTime != 0;
      
      if (!hasCloseTime && !hadCloseTime) return;
      
      if (hadCloseTime) pendingClose.remove (this);
      
      idleCloseTime = closeTime;
      
      if (hasCloseTime)
      {
        pendingClose.add (this);
        
        if (pendingClose.first () == this)
          pendingClose.notify ();  // wake up
      }
    }
  }
  
  
  
  //--- implementation of Comparable -------------------------------------------
  

  /**
   *  compare this instance with another one. The comparison is based on the
   *  {@code idleCloseTime}. If two times are accidentially equal,
   *  the internal instance ID is used for the comparison to make sure that
   *  two instances are only reported as equal if they are identical. Otherwise
   *  the one instance would kick out the other instance in the set.
   *
   *  @param obj                    the other object
   *  @return                       the result ({@code -1} if this
   *                                instance has an earlier 
   *                                {@code idleCloseTime})
   */

  @Override
  public int compareTo (TcpChannel obj)
  {
    if (obj == this) return 0;
    
    if (obj.idleCloseTime == idleCloseTime)
      return instanceID < obj.instanceID ? -1 : 1;
    
    return idleCloseTime < obj.idleCloseTime ? -1 : 1;
  }
  
  
  
  
  //--- implementation of RegistryChannel --------------------------------------

  
  /**
   *  send the given message to the registry. As the transport mechanism
   *  may need to know whether the data represents the XML or key-value form
   *  of the request, it must be specified. If the channel is already closed,
   *  an exception is thrown.
   *
   *  @param data                     the data representing the message
   *  @param isXml                    {@code true} if the message uses the
   *                                  XML encoding, {@code false} if the
   *                                  message uses key-value pairs.
   *  @throws IOException             if the transmission failed
   */

  @Override
  public void send (byte[] data, boolean isXml)
    throws IOException
  {
    synchronized (outStream)
    {
      if (idleTimeout != 0) setCloseTime (0);
      
      if (closed)
        throw new IOException ("channel already closed");

      try
      {
        outStream.writeInt (data.length);
        outStream.write (data);
        outStream.flush ();
        
      } finally
      {
        if (idleTimeout != 0)
          setCloseTime (System.currentTimeMillis () + idleTimeout);
      }
    }
  }
  
  
  
  /**
   *  wait for a response from the registry.
   *
   *  @return                         the message that has been received
   *  @throws IOException             if the connection has been closed either
   *                                  before or during the call.
   */

  @Override
  public byte[] receive ()
    throws IOException
  {
    synchronized (inStream)
    {
      if (idleTimeout != 0) setCloseTime (0);
      
      if (closed)
        throw new EOFException ("channel already closed");
      
      try
      {  
        int length = inStream.readInt ();
        int maxLength = params.getMaxReceiveSize ();
        if (length < 0 || maxLength > 0 && length > maxLength)
          throw new IOException ("message length too small or larger than " +
            "the configured length of " + maxLength + " bytes");

        byte[] data = new byte[length];
        inStream.readFully (data);
        return data;
        
      } finally
      {
        if (idleTimeout != 0)
          setCloseTime (System.currentTimeMillis () + idleTimeout);
      }
    }
  }
  
  
  
  /**
   *  return whether the channel has been closed
   *
   *  @return                     {@code true} if closed
   */

  @Override
  public boolean isClosed ()
  {
    synchronized (pendingClose)
    {
      if (closed) return true;

      if (socket.isClosed ())
      {
        closed = true;
        return true;
      }
      
      return idleTimeout != 0 && 
        idleCloseTime - System.currentTimeMillis () < MIN_REMAIN_TIME;
    }
  }
  
  
  
  /**
   *  close the channel. Closing an already closed channel should be
   *  ignored
   */

  @Override
  public void close ()
  {
    if (idleTimeout != 0) setCloseTime (0);
    
    if (!closed) 
    {
      closed = true;
      
      try
      {
        socket.close ();

      } catch (IOException e)
      {
        // ignored
      }
    }
  }
  
  
  
  
  
  /**
   *  class that monitors idle connections and closes them if need be
   *
   *  @version  $Revision: 14936 $
   *  @author   Klaus Malorny
   */
  
  private static class IdleMonitor implements Runnable
  {
    
    
    /**
     *  periodically check channels whether they have timed out
     */

    @Override
    public void run ()
    {
      while (true)
      {
        TcpChannel tc = null;
        
        synchronized (pendingClose)
        {
          if (pendingClose.isEmpty ())
          {
            // no channel to wait for -> wait indefinitely
            
            try
            {
              pendingClose.wait ();
              
            } catch (InterruptedException e)
            {
              return;
            }
          } else
          {
            TcpChannel temp = pendingClose.first ();
            long now = System.currentTimeMillis ();
            long diff = temp.idleCloseTime - now;
            
            if (diff <= 0)
            {
              // candidate to close
              
              tc = temp;
              tc.closed = true;
              pendingClose.remove (tc);
              
            } else
            {
              // wait until time to close
              try
              {
                pendingClose.wait (diff);
                
              } catch (InterruptedException e)
              {
                return;
              }
            }
          }
        }
        
        if (tc != null)
        {
          try
          {
            tc.socket.close ();
            
          } catch (IOException e)
          {
            // ignored
          }
        }
      }
    }
    
    
    
  }
}


