package org.jacorb.orb.miop;

import java.io.IOException;
import java.io.InterruptedIOException;
import java.net.DatagramPacket;
import java.net.MulticastSocket;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import org.jacorb.config.Configuration;
import org.jacorb.config.ConfigurationException;
import org.jacorb.orb.CDRInputStream;
import org.omg.CORBA.COMM_FAILURE;
import org.omg.CORBA.MARSHAL;
import org.omg.CORBA.SystemException;
import org.omg.ETF.BufferHolder;
import org.omg.ETF.Profile;


/**
 * Listen to a specified group port and class D IP address for messages
 * addressed to this group.
 *
 * @author Alysson Neves Bessani
 * @author Nick Cross
 * @version 1.0
 * @see MIOPConnection MIOPListener
 */
public class ServerMIOPConnection extends MIOPConnection implements Runnable
{
   /** This socket */
   private MulticastSocket socket             = null;

   /** This thread */
   private Thread          groupListener      = null;

   /** the complete messages (of byte[] arrays] */
   private LinkedList<byte[]>      fullMessages       = new LinkedList<byte[]> ();

   /**
    * Incomplete messages. When a message is complete it is transfered to
    * the fullMessage list
    */
   private HashMap<String, FragmentedMessage>         incompleteMessages = null;

   /** Current read message */
   private byte[]          current            = null;

   /** read pos in the current message */
   private int             currentPos         = 0;

   /** socket and connection configuration parameters */
   private int socketTimeout;
   private int timeToLive;

   private int incompleteMessagesThreshold;

   private short packetMaxSize;

   /**
    * Creates a new server MIOP connection that listen to the specified group.
    */
   public ServerMIOPConnection ()
   {
      super ();

      groupListener = new Thread (this);
      groupListener.setDaemon (true);
   }


   /**
    * Connect the socket. Called by MIOPListener.
    *
    * @param profile
    * @param time_out unused, we use SO_TIMEOUT
    */
   @Override
   public void connect (Profile profile, long time_out)
   {
      if ( ! is_connected())
      {
         if (profile instanceof MIOPProfile)
         {
            this.profile = (MIOPProfile) profile;
         }
         else
         {
            throw new org.omg.CORBA.BAD_PARAM
            ( "attempt to connect an MIOP connection "
              + "to a non-MIOP profile: " + profile.getClass());
         }

         try
         {
            socket = new MulticastSocket (((MIOPProfile)profile).getUIPMCProfile ().the_port);

            socket.setSoTimeout (socketTimeout);
            socket.setTimeToLive (timeToLive);
            socket.joinGroup (((MIOPProfile)profile).getGroupInetAddress ());

            connection_info = socket.toString ();
         }
         catch (Exception e)
         {
            if (socket != null)
            {
               socket.close ();
            }
            throw new RuntimeException ("Can't create multicast socket: " + profile);
         }

         connected = true;

         groupListener.start ();
      }
   }


   /**
    * Tests if there is data available on this connection.
    *
    * @return true if there are data available.
    */
   @Override
   public boolean is_data_available ()
   {
      return (current != null && currentPos < current.length) || !fullMessages.isEmpty ();
   }


   /**
    * Wait until there is some complete message ready data to be read from
    * connection.
    *
    * @param timeout unused.
    * @return always true.
    */
   @Override
   public synchronized boolean wait_next_data (long timeout)
   {
      while (fullMessages.isEmpty ())
      {
         try
         {
            wait ();
         }
         catch (InterruptedException ie)
         {
         }
      }
      current = fullMessages.removeFirst ();
      currentPos = 0;
      return true;
   }


   /**
    * Read data from the connection.
    *
    *
    * @param buffer the buffer holder.
    * @param offset the buffer offset.
    * @param minLength the minimum length to be read.
    * @param maxLength the maximum length to be read.
    * @param timeout unused
    */
   @Override
   public synchronized int read (BufferHolder buffer, int offset, int minLength, int maxLength,
            long timeout)
   {

      if (current == null)
      {
         wait_next_data (0);
      }

      int writen = 0;

      do
      {
         int toRead = Math.min (current.length - currentPos, maxLength);

         System.arraycopy (current, currentPos, buffer.value, offset + writen, toRead);

         writen += toRead;
         currentPos += toRead;

         if (currentPos == current.length)
         {
            current = null;
         }
         if (current == null && writen < minLength)
         {
            wait_next_data (0);
         }
      }
      while (writen < minLength);

      return writen;
   }


   /**
    * Close this connection.
    */
   @Override
   public synchronized void close ()
   {
      if (!connected)
      {
         return;
      }

      // Finish with the multicast socket
      try
      {
         socket.leaveGroup (((MIOPProfile)profile).getGroupInetAddress ());
      }
      catch (IOException ex)
      {
         if (logger.isDebugEnabled())
         {
            logger.debug ("Exception when closing the socket", ex);
         }
      }
      try
      {
         socket.close ();

         //this will cause exceptions when trying to read from
         //the streams. Better than "nulling" them.
         if( in_stream != null )
         {
            in_stream.close();
         }
         if( out_stream != null )
         {
            out_stream.close();
         }
      }
      catch (IOException ex)
      {
         if (logger.isDebugEnabled())
         {
            logger.debug ("Exception when closing the socket", ex);
         }
      }
      connected = false;
   }


   /**
    * Run method. Inherited from runnable, this method is used to stay listening
    * to the socket for new messages.
    */
   @Override
   public void run ()
   {
      // create incomplete table if doesn't exist
      if (incompleteMessages == null)
      {
          incompleteMessages = new HashMap<String, FragmentedMessage> ();
      }

      // allocates a buffer
      byte[] buffer = new byte [packetMaxSize];

      while (connected)
      {
         try
         {
            // if the number of messages in incomplete table is greater than a
            // specified threshold we inspect this table to clear incomplete packets
            // collections.
            if (incompleteMessages.size () > incompleteMessagesThreshold)
            {
               dropIncompleteMessages ();
            }

            // creates a new datagram to be read
            DatagramPacket packet = new DatagramPacket (buffer, buffer.length);

            try
            {
               // wait for the datagram
               socket.receive (packet);
            }
            catch (SocketTimeoutException ste)
            {
               continue;
            }
            catch( InterruptedIOException e )
            {
                throw new org.omg.CORBA.TRANSIENT ("Interrupted I/O: " + e);
            }
            catch (IOException se)
            {
               throw to_COMM_FAILURE (se);
            }
            // the packet was received successfully.
            CDRInputStream in = new CDRInputStream (configuration.getORB (), packet.getData ());

            // Read the header
            //
            // Manually read in the stream rather than using the generated
            // PacketHeader_1_0Helper
            // as we may need to alter endian half way through.
            org.omg.MIOP.PacketHeader_1_0 header = new org.omg.MIOP.PacketHeader_1_0 ();
            header.magic = new char[4];
            in.read_char_array (header.magic, 0, 4);

            // Verify the message is MIOP
            if ( ! MulticastUtil.matchMIOPMagic (header.magic))
            {
               // if it isn't a MIOP message I can ignore it
               continue;
            }

            // We know it is MIOP from now on.
            header.hdr_version = in.read_octet ();
            header.flags = in.read_octet ();

            // Set endian for the stream
            in.setLittleEndian ((0x01 & header.flags) != 0);

            header.packet_length = in.read_ushort ();
            header.packet_number = in.read_ulong ();
            header.number_of_packets = in.read_ulong ();
            header.Id = org.omg.MIOP.UniqueIdHelper.read (in);

            int pos = in.get_pos ();
            // difference to next MulticastUtil.BOUNDARY (which is an 8 byte boundary)
            int header_padding = MulticastUtil.BOUNDARY - (pos % MulticastUtil.BOUNDARY);
            header_padding = (header_padding == MulticastUtil.BOUNDARY) ? 0 : header_padding;

            // skip header_padding bytes anyway, because if no body is
            // present, nobody will try to read it
            in.skip (header_padding);

            // read the GIOP data
            byte data[] = new byte[header.packet_length];
            if (in.available () < data.length)
            {
               throw new MARSHAL
               (
                  "Impossible length in MIOP header. Header denotes length of " +
                  header.packet_length +
                  " but only " +
                  in.available () +
                  " is available."
               );
            }
            in.read_octet_array (data, 0, header.packet_length);

            String messageId = new String (header.Id);
            FragmentedMessage message = incompleteMessages.get (messageId);

            // verify if it's the first message to arrive
            if (message == null)
            {
               // If this is the first fragment of the message create a  fragmented message
               message = new FragmentedMessage ();
               try
               {
                   message.configure (configuration);
               }
               catch (ConfigurationException e)
               {
                   logger.error("couldn't create a Fragmented message", e);
                   throw new IllegalArgumentException("wrong configuration: " + e);
               }
               incompleteMessages.put (messageId, message);
            }

            if (logger.isDebugEnabled ())
            {
               logger.debug ("Received message number " + (header.packet_number + 1) + " out of " + header.number_of_packets + " and adding fragment of size " + data.length);
            }
            message.addFragment (header, data);

            // verify if it's the last message to arrive
            if (message.isComplete ())
            {
               synchronized (this)
               {
                  incompleteMessages.remove (messageId);
                  fullMessages.addLast (message.buildMessage ());
                  notifyAll ();
               }
            }
         }
         catch (COMM_FAILURE e)
         {
            if (logger.isDebugEnabled())
            {
                logger.debug("Transport to " + connection_info +
                             ": stream closed " + e.getMessage() );
            }
            if (connected)
            {
               close();
            }
         }
         catch (SystemException e)
         {
            if (logger.isWarnEnabled ())
            {
               logger.warn ("ServerMIOPConnection caught exception.", e);
            }
         }
         catch (Throwable e)
         {
            if (logger.isErrorEnabled ())
            {
               logger.error ("ServerMIOPConnection caught exception.", e);
            }
         }
      }
   }


   /**
    * Remove all incomplete messages that can be deleted from the incomplete
    * messages table.
    */
   private final synchronized void dropIncompleteMessages ()
   {
      Iterator<String> ids = incompleteMessages.keySet ().iterator ();

      while (ids.hasNext ())
      {
         Object id = ids.next ();
         if (incompleteMessages.get (id).canBeDiscarded ())
         {
            incompleteMessages.remove (id);
         }
      }
   }


   /*
    * (non-Javadoc)
    *
    * @see org.jacorb.orb.etf.ConnectionBase#getTimeout()
    */
   @Override
   protected int getTimeout ()
   {
       try
       {
           return socket.getSoTimeout();
       }
       catch (SocketException se)
       {
           throw to_COMM_FAILURE (se);
       }
   }


   /*
    * (non-Javadoc)
    *
    * @see org.jacorb.orb.etf.ConnectionBase#setTimeout(int)
    */
   @Override
   protected void setTimeout (int timeout)
   {
       if (socket != null)
       {
           try
           {
               if (logger.isInfoEnabled())
               {
                   logger.info ("Socket timeout set to " + timeout + " ms");
               }
               socket.setSoTimeout(timeout);
           }
           catch( SocketException se )
           {
               if (logger.isInfoEnabled())
               {
                   logger.info("SocketException", se);
               }
           }
       }
   }


   @Override
   public void configure(Configuration config) throws ConfigurationException
   {
       super.configure(config);

       socketTimeout = config.getAttributeAsInteger("jacorb.miop.timeout",
               MulticastUtil.SO_TIMEOUT);
       timeToLive = config.getAttributeAsInteger("jacorb.miop.time_to_live",
               MulticastUtil.TIME_TO_LIVE);
       incompleteMessagesThreshold = config.getAttributeAsInteger("jacorb.miop.incomplete_messages_threshold",
               MulticastUtil.INCOMPLETE_MESSAGES_THRESHOULD);
       packetMaxSize = (short) config.getAttributeAsInteger ("jacorb.miop.packet_max_size",
               MulticastUtil.PACKET_MAX_SIZE);
   }
}
