/*
    * $HeadURL: $
    * $Date: $
    * $Revision: $
    * $Author: $
    * 
    * Copyright (c) 2005 MindTree Consulting Ltd. 
    * 
    * This file is part of Insight.
   * 
   * Insight is free software: you can redistribute it and/or modify it under the 
   * terms of the GNU General Public License as published by the Free Software 
   * Foundation, either version 3 of the License, or (at your option) any later 
   * version.
   * 
   * Insight 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 General 
   * Public License for more details.
   * 
   * You should have received a copy of the GNU General Public License along with 
   * Insight.  If not, see <http://www.gnu.org/licenses/>.
   */
  package com.mindtree.techworks.insight.receiver;
  
  import java.io.BufferedReader;
  import java.io.IOException;
  import java.io.Serializable;
  import java.io.StringReader;
  import java.net.DatagramPacket;
  import java.net.DatagramSocket;
  import java.net.SocketException;
  import java.net.SocketTimeoutException;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.HashMap;
  import java.util.Iterator;
  import java.util.LinkedList;
  import java.util.List;
  import java.util.Map;
  
  import com.mindtree.techworks.insight.model.ReceiverFormat;
  import com.mindtree.techworks.insight.remoteprotocol.RemoteProtocolException;
  import com.mindtree.techworks.insight.remoteprotocol.spi.DataPacket;
  import com.mindtree.techworks.insight.remoteprotocol.spi.LogMessageDataContent;
  import com.mindtree.techworks.insight.remoteprotocol.spi.MessageType;
  import com.mindtree.techworks.insight.remoteprotocol.spi.NullMessageDataContent;
  import com.mindtree.techworks.insight.remoteprotocol.spi.PacketDataContent;
  import com.mindtree.techworks.insight.remoteprotocol.spi.PacketHeader;
  import com.mindtree.techworks.insight.remoteprotocol.spi.StartupMessageDataContent;
  import com.mindtree.techworks.insight.spi.LogEvent;
  import com.mindtree.techworks.insight.spi.LogNamespace;
  
  
  /**
   * This is a concrete implementation of the
   * <code>com.mindtree.techworks.insight.receiver.AbstractReceiver</code> class. It
   * supports listening on an UDP port and receiving log events in the form of
   * DataPackets as defined in
   * <code>com.mindtree.techworks.insight.remoteprotocol.spi.DataPacket</code>.
   * <p>
   * This class supports receiving from multiple namespaces at the same time, and
   * also from multiple sources at the same time.
   * </p>
   * 
   * @see com.mindtree.techworks.insight.receiver.AbstractReceiver
   * @see com.mindtree.techworks.insight.remoteprotocol.spi.DataPacket
   * @author <a href="mailto:bindul_bhowmik@mindtree.com">Bindul Bhowmik</a>
   * @version $Revision: 27 $ $Date: 2007-12-16 04:58:03 -0700 (Sun, 16 Dec 2007) $
   */
  public class RemoteProtocolStreamReceiver extends AbstractReceiver {
  	
  	/**
  	 * The default timeout set on the datagram socket
  	 */
  	private static final int SOCKET_TIMEOUT = 2000;
  
  	// -------------------------------------------------------------------------
  	// Instance variables
  	// -------------------------------------------------------------------------
  
  	/**
  	 * The queue containing all packets that are read.
  	 */
  	protected LinkedList packetQueue = new LinkedList ();
  
  	/**
  	 * The port on which the receiver is listening
  	 */
  	protected int portListeningOn;
  
  	/**
  	 * Map of known sources from which this receiver is receiving data.
  	 */
  	private Map knownSources = new HashMap (4);
  
  	/**
  	 * Map of unknown sources from which we are getting data.
  	 */
 	private Map unknownSources = new HashMap (2);
 	
 	/**
 	 * Has a list of all loaded namespaces.
 	 */
 	private List loadedNamespaces = new ArrayList(2);
 
 	/**
 	 * The current receiver format being used to interpret log messages.
 	 */
 	private ReceiverFormat[] receiverFormat;
 
 	/**
 	 * Namespace used for messages for which no log namespace has been defined.
 	 */
 	private LogNamespace unknownNamespace;
 
 	/**
 	 * The datagram socket used by this class to listen to a port
 	 */
 	private DatagramSocket socket;
 
 	// -------------------------------------------------------------------------
 	// Constructor
 	// -------------------------------------------------------------------------
 
 	/**
 	 * Creates an instance of the Receiver
 	 * 
 	 * @param receiverFormat
 	 *            The Receiver format used for the receiver.
 	 * @param portListeningOn
 	 *            The port on which to listen.
 	 * @throws ReceiverInitializationException
 	 *             If the Receiver could not be initialized properly.
 	 */
 	public RemoteProtocolStreamReceiver (ReceiverFormat[] receiverFormat,
 			int portListeningOn) throws ReceiverInitializationException {
 
 		this.receiverFormat = receiverFormat;
 		this.portListeningOn = portListeningOn;
 		this.unknownNamespace = new LogNamespace ("unknown", "unknown",
 				receiverFormat, "unknown");
 		this.initialize (null, true);
 	}
 
 	// -------------------------------------------------------------------------
 	// Methods overridden from com.mindtree.techworks.insight.receiver.AbstractReceiver
 	// -------------------------------------------------------------------------
 
 	/**
 	 * @see com.mindtree.techworks.insight.receiver.AbstractReceiver#getNamespaces()
 	 */
 	public LogNamespace[] getNamespaces () {
 
 		if (loadedNamespaces.size() > 0) {
 			LogNamespace[] namespaces = new LogNamespace[loadedNamespaces.size()];
 			return (LogNamespace []) loadedNamespaces.toArray(namespaces);
 		}
 		return new LogNamespace[] {unknownNamespace};
 	}
 
 	/**
 	 * Always returns <code>true</code> since the RemoteProtocolStreamReceiver
 	 * is by nature tailing.
 	 * 
 	 * @see com.mindtree.techworks.insight.receiver.AbstractReceiver#isTailing()
 	 */
 	public boolean isTailing () {
 		return true;
 	}
 
 	/**
 	 * Initializes this instance. Both the parameters are ignored.
 	 * 
 	 * @see com.mindtree.techworks.insight.receiver.AbstractReceiver#initialize(com.mindtree.techworks.insight.spi.LogNamespace,
 	 *      boolean)
 	 */
 	protected void initialize (LogNamespace namespace, boolean isTailing)
 			throws ReceiverInitializationException {
 		try {
 			socket = new DatagramSocket (this.portListeningOn);
 			socket.setSoTimeout(SOCKET_TIMEOUT);
 		} catch (SocketException e) {
 			throw new ReceiverInitializationException (
 					"Could not open socket.", e);
 		}		
 	}
 
 	/**
 	 * @see com.mindtree.techworks.insight.receiver.AbstractReceiver#deInitialize()
 	 */
 	protected void deInitialize () {
 		this.socket.close();
 	}
 
 	/**
 	 * @see com.mindtree.techworks.insight.receiver.AbstractReceiver#getNextEvents()
 	 */
 	protected LogEvent [] getNextEvents () {
 
 		populatePacketQueue();
 		
 		// Check the current size of the packet queue and start processing.
 		// There might be packets received during the period, but they will be
 		// processed on the next call to this method.
 		int packetCountToProcess = packetQueue.size ();
 		ArrayList processedEvents = new ArrayList();
 		for (int i = 0; i < packetCountToProcess; i++ ) {
 			DataPacket currentPacket = null;
 			synchronized (packetQueue) {
 				currentPacket = (DataPacket) packetQueue.removeFirst ();
 			}
 			switch (currentPacket.getPacketHeader ().getMessageType ()
 					.getType ()) {
 				case MessageType.STARTUP_MESSAGE_TYPE:
 					processStartupPacket (currentPacket);
 					break;
 				case MessageType.LOG_DATA_MESSAGE_TYPE:
 					processedEvents.addAll (processLogPacket (currentPacket));
 					break;
 				case MessageType.SHUTDOWN_MESSAGE_TYPE:
 					Short sourceId = new Short (currentPacket
 							.getPacketHeader ().getSourceIdentifier ());
 					knownSources.remove (sourceId);
 					break;
 				default:
 			// do nothing!
 			}
 		}
 
 		if (null != processedEvents) {
 			LogEvent [] loggingEvents = new LogEvent [processedEvents.size ()];
 			return (LogEvent []) processedEvents.toArray (loggingEvents);
 		}
 		return null;
 	}
 
 
 	/**
 	 * @see com.mindtree.techworks.insight.receiver.AbstractReceiver#hasMoreEvents()
 	 */
 	protected boolean hasMoreEvents () {
 
 		if ( !this.packetQueue.isEmpty ()) {
 			for (Iterator packetItr = packetQueue.iterator (); packetItr
 					.hasNext ();) {
 				if (MessageType.LOG_DATA_MESSAGE
 						.equals (((DataPacket) packetItr.next ())
 								.getPacketHeader ().getMessageType ())) {
 					return true;
 				}
 			}
 			return false;
 		}
 		return false;
 	}
 
 	// -------------------------------------------------------------------------
 	// Private utility methods
 	// -------------------------------------------------------------------------
 
 	/**
 	 * This method processes the startup message from a receiver.
 	 * 
 	 * @param currentPacket
 	 *            The current packet, must be a startup packet.
 	 */
 	private void processStartupPacket (DataPacket currentPacket) {
 
 		// Check the packet type
 		if ( !MessageType.STARTUP_MESSAGE.equals (currentPacket
 				.getPacketDataContent ().getMessageType ())) {
 			return;
 		}
 
 		// Get the source details object or create it
 		Short sourceId = new Short (currentPacket.getPacketHeader ()
 				.getSourceIdentifier ());
 		SourceDetails sourceDetails = null;
 		if ( !knownSources.containsKey (sourceId)) {
 			// Check in unknown sources
 			if (unknownSources.containsKey (sourceId)) {
 				sourceDetails = (SourceDetails) unknownSources.get (sourceId);
 				unknownSources.remove (sourceId);
 			} else {
 				sourceDetails = new SourceDetails (sourceId.shortValue ());
 			}
 		} else {
 			sourceDetails = (SourceDetails) knownSources.get (sourceId);
 		}
 
 		// Check if namespace is set
 		if ( !sourceDetails.isLogNamespaceSet ()) {
 			StartupMessageDataContent messageData = (StartupMessageDataContent) currentPacket
 					.getPacketDataContent ();
 			String namespace = messageData.getNamespace ();
 			String nodeId = namespace.split ("/")[2];
 
 			LogNamespace logNamespace = new LogNamespace (namespace, null,
 					receiverFormat, nodeId);
 			sourceDetails.setLogNamespace (logNamespace);
 
 			LogInterpreter logInterpreter = new LogInterpreter (logNamespace);
 			sourceDetails.setLogInterpreter (logInterpreter);
 
 			knownSources.put (sourceId, sourceDetails);
 		}
 
 	}
 
 	/**
 	 * Processes a log message packet
 	 * 
 	 * @param currentPacket
 	 *            The log message packet to process
 	 * @return The lsit of messages interpretted.
 	 */
 	private ArrayList processLogPacket (DataPacket currentPacket) {
 
 		// Check the packet type
 		if ( !MessageType.LOG_DATA_MESSAGE.equals (currentPacket
 				.getPacketDataContent ().getMessageType ())) {
 			return new ArrayList (0);
 		}
 
 		// Get the source details object
 		Short sourceId = new Short (currentPacket.getPacketHeader ()
 				.getSourceIdentifier ());
 		SourceDetails sourceDetails = null;
 		if ( !knownSources.containsKey (sourceId)) {
 			// Check unknown sources
 			if (unknownSources.containsKey (sourceId)) {
 				sourceDetails = (SourceDetails) unknownSources.get (sourceId);
 			} else {
 				sourceDetails = new SourceDetails (sourceId.shortValue ());
 				sourceDetails.setLogNamespace (unknownNamespace);
 				sourceDetails.setLogInterpreter (new LogInterpreter (unknownNamespace));
 				sourceDetails.setLogNamespaceSet (false);
 
 				unknownSources.put (sourceId, sourceDetails);
 			}
 		} else {
 			sourceDetails = (SourceDetails) knownSources.get (sourceId);
 		}
 
 		// Get the LogInterpreter and the message
 		LogInterpreter logInterpreter = sourceDetails.getLogInterpreter ();
 		LogMessageDataContent logMessageDataContent = (LogMessageDataContent) currentPacket
 				.getPacketDataContent ();
 		ArrayList logMessages = new ArrayList ();
 		String logData = null;
 		
 		try {
 			logData = logMessageDataContent.getLogMessage ();
 		} catch (RemoteProtocolException e) {
 			// TODO Log this here
 			logData = "";
 		}
 		
 		// Add the namespace to the loaded namespaces if not already existing
 		if (!loadedNamespaces.contains(sourceDetails.getLogNamespace())) {
 			loadedNamespaces.add(sourceDetails.getLogNamespace());
 		}
 
 		BufferedReader logReader = new BufferedReader (new StringReader (
 				logData));
 		try {
 			while (logReader.ready ()) {
 				String line = logReader.readLine ();
 				if (null != line) {
 					LogEvent [] logEvents = logInterpreter
 							.parseLogMessage(line);
 					logMessages.addAll (Arrays.asList (logEvents));
 				} else {
 					break;
 				}
 			}
 		} catch (IOException ioe) {
 			// Just ignore it and send the messages we have interpretted!
 		}
 
 		return logMessages;
 	}
 
 		/**
 		 * @see java.lang.Thread#run()
 		 */
 		private void populatePacketQueue () {
 			DatagramPacket packet = new DatagramPacket (
 					new byte [DataPacket.MAX_PACKET_SIZE],
 					DataPacket.MAX_PACKET_SIZE);
 			try {
 				socket.receive (packet);
 
 				try {
 					PacketHeader header = new PacketHeader (packet
 							.getData ());
 					// TODO Change this implementation to get the packet
 					// TODO See java.nio
 					byte [] messageData = new byte [packet.getLength ()];
 					System.arraycopy (packet.getData (), 0, messageData, 0,
 										packet.getLength ());
 
 					PacketDataContent content = null;
 
 					switch (header.getMessageType ().getType ()) {
 						case MessageType.STARTUP_MESSAGE_TYPE:
 							content = new StartupMessageDataContent (
 									messageData, header);
 							break;
 						case MessageType.LOG_DATA_MESSAGE_TYPE:
 							content = new LogMessageDataContent (
 									messageData, header);
 							break;
 						case MessageType.SHUTDOWN_MESSAGE_TYPE:
 							content = new NullMessageDataContent ();
 							content
 									.setMessageType (MessageType.SHUTDOWN_MESSAGE);
 							break;
 						default:
 					// do nothing
 					}
 
 					if (null != content) {
 						DataPacket dataPacket = new DataPacket (content,
 								header);
 						synchronized (packetQueue) {
 							packetQueue.addLast (dataPacket);
 						}
 					}
 				} catch (RemoteProtocolException e) {
 					// Do nothing
 
 				}					
 			} catch (SocketTimeoutException ste) {
 				// Try again.
 			} catch (IOException ioex) {
 				// Do nothing!
 				ioex.printStackTrace ();
 			} catch (Exception ie) {
 				// Someone wants us to stop
 				ie.printStackTrace();
 			}
 		}
 
 	/**
 	 * Inner class to hold the different objects in relation to a particular
 	 * source including the <code>LogNamespace</code>,
 	 * <code>LogInterpreter</code> (including the <code>ReceiverFormat</code>)
 	 * and if a startup message has been received for this source.
 	 * 
 	 * @see Serializable
 	 * @author <a href="mailto:bindul_bhowmik@mindtree.com">Bindul Bhowmik</a>
 	 * @version $Revision: 27 $ $Date: 2007-12-16 04:58:03 -0700 (Sun, 16 Dec 2007) $
 	 */
 	protected class SourceDetails implements Serializable {
 
 		// ---------------------------------------------------------------------
 		// Class variables
 		// ---------------------------------------------------------------------
 
 		/**
 		 * The serial version UID for the class
 		 */
 		private static final long serialVersionUID = -4363523350633442160L;
 
 		// ---------------------------------------------------------------------
 		// Instance variables
 		// ---------------------------------------------------------------------
 
 		/**
 		 * The namespace object associated with the source.
 		 */
 		private LogNamespace logNamespace;
 
 		/**
 		 * The source id of the identifier
 		 */
 		private short sourceId;
 
 		/**
 		 * Indicates if a startup message has been received for this source.
 		 */
 		private boolean isLogNamespaceSet = false;
 
 		/**
 		 * The log interpreter used for this source.
 		 */
 		private LogInterpreter logInterpreter;
 
 		// ---------------------------------------------------------------------
 		// Constructors
 		// ---------------------------------------------------------------------
 
 		/**
 		 * Creates an instance of this source details.
 		 * 
 		 * @param sourceId
 		 *            The identifier of the source.
 		 */
 		public SourceDetails (short sourceId) {
 
 			this.sourceId = sourceId;
 		}
 
 		// ---------------------------------------------------------------------
 		// Accessors
 		// ---------------------------------------------------------------------
 
 		/**
 		 * Returns the isLogNamespaceSet
 		 * 
 		 * @return Returns the isLogNamespaceSet.
 		 */
 		public boolean isLogNamespaceSet () {
 
 			return isLogNamespaceSet;
 		}
 
 
 		/**
 		 * Sets the isLogNamespaceSet
 		 * 
 		 * @param isLogNamespaceSet
 		 *            The isLogNamespaceSet to set.
 		 */
 		public void setLogNamespaceSet (boolean isLogNamespaceSet) {
 
 			this.isLogNamespaceSet = isLogNamespaceSet;
 		}
 
 
 		/**
 		 * Returns the logNamespace
 		 * 
 		 * @return Returns the logNamespace.
 		 */
 		public LogNamespace getLogNamespace () {
 
 			return logNamespace;
 		}
 
 
 		/**
 		 * Sets the logNamespace
 		 * 
 		 * @param logNamespace
 		 *            The logNamespace to set.
 		 */
 		public void setLogNamespace (LogNamespace logNamespace) {
 
 			this.logNamespace = logNamespace;
 		}
 
 
 		/**
 		 * Returns the sourceId
 		 * 
 		 * @return Returns the sourceId.
 		 */
 		public short getSourceId () {
 
 			return sourceId;
 		}
 
 
 		/**
 		 * Returns the logInterpreter
 		 * 
 		 * @return Returns the logInterpreter.
 		 */
 		public LogInterpreter getLogInterpreter () {
 
 			return logInterpreter;
 		}
 
 
 		/**
 		 * Sets the logInterpreter
 		 * 
 		 * @param logInterpreter
 		 *            The logInterpreter to set.
 		 */
 		public void setLogInterpreter (LogInterpreter logInterpreter) {
 
 			this.logInterpreter = logInterpreter;
 		}
 	}
 }