/*
* Conditions Of Use 
* 
* This software was developed by employees of the National Institute of
* Standards and Technology (NIST), an agency of the Federal Government.
* Pursuant to title 15 Untied States Code Section 105, works of NIST
* employees are not subject to copyright protection in the United States
* and are considered to be in the public domain.  As a result, a formal
* license is not needed to use the software.
* 
* This software is provided by NIST as a service and is expressly
* provided "AS IS."  NIST MAKES NO WARRANTY OF ANY KIND, EXPRESS, IMPLIED
* OR STATUTORY, INCLUDING, WITHOUT LIMITATION, THE IMPLIED WARRANTY OF
* MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE, NON-INFRINGEMENT
* AND DATA ACCURACY.  NIST does not warrant or make any representations
* regarding the use of the software or the results thereof, including but
* not limited to the correctness, accuracy, reliability or usefulness of
* the software.
* 
* Permission to use this software is contingent upon your acceptance
* of the terms of this agreement
*  
* .
* 
*/
/******************************************************************************
 * Product of NIST/ITL Advanced Networking Technologies Division (ANTD).      *
 ******************************************************************************/
package gov.nist.javax.sip.stack;

import gov.nist.javax.sip.header.*;
import gov.nist.javax.sip.message.*;
import gov.nist.javax.sip.parser.*;
import gov.nist.core.*;
import java.net.*;
import java.io.*;
import java.text.ParseException;

import javax.sip.address.Hop;
import javax.sip.message.Response;

/* Ahmet Uyar
* <auyar@csit.fsu.edu>sent in a bug report for TCP operation of the
* JAIN sipStack. Niklas Uhrberg suggested that a mechanism be added to
* limit the number of simultaneous open connections. The TLS
* Adaptations were contributed by Daniel Martinez. Hagai Sela
* contributed a bug fix for symmetric nat. Jeroen van Bemmel
* added compensation for buggy clients ( Microsoft RTC clients ).
* Bug fixes by viswashanti.kadiyala@antepo.com, Joost Yervante Damand				
*/


/**
 * This is a stack abstraction for TCP connections. This abstracts a stream of parsed
 * messages. The SIP sipStack starts this from the main SIPStack class for each
 * connection that it accepts. It starts a message parser in its own thread and
 * talks to the message parser via a pipe. The message parser calls back via the
 * parseError or processMessage functions that are defined as part of the
 * SIPMessageListener interface.
 * 
 * @see gov.nist.javax.sip.parser.PipelinedMsgParser
 * 
 * 
 * @author M. Ranganathan <br/>
 * 
 * @version 1.2 $Revision: 1.36 $ $Date: 2006/07/25 19:47:00 $ 
 */
public class TCPMessageChannel extends MessageChannel implements
		SIPMessageListener, Runnable {

	private Socket mySock;

	private PipelinedMsgParser myParser;

	protected InputStream myClientInputStream; // just to pass to thread.

	protected OutputStream myClientOutputStream;

	protected String key;

	protected boolean isCached;

	protected boolean isRunning;

	private Thread mythread;

	protected SIPTransactionStack sipStack;

	protected String myAddress;

	protected int myPort;

	protected InetAddress peerAddress;

	protected int peerPort;

	protected String peerProtocol;

	// Incremented whenever a transaction gets assigned
	// to the message channel and decremented when
	// a transaction gets freed from the message channel.
	protected int useCount;

	private TCPMessageProcessor tcpMessageProcessor;

	protected TCPMessageChannel(SIPTransactionStack sipStack) {
		this.sipStack = sipStack;

	}

	/**
	 * Constructor - gets called from the SIPStack class with a socket on
	 * accepting a new client. All the processing of the message is done here
	 * with the sipStack being freed up to handle new connections. The sock
	 * input is the socket that is returned from the accept. Global data that is
	 * shared by all threads is accessible in the Server structure.
	 * 
	 * @param sock
	 *            Socket from which to read and write messages. The socket is
	 *            already connected (was created as a result of an accept).
	 * 
	 * @param sipStack
	 *            Ptr to SIP Stack
	 */

	protected TCPMessageChannel(Socket sock, SIPTransactionStack sipStack,
			TCPMessageProcessor msgProcessor) throws IOException {

		if (sipStack.isLoggingEnabled()) {
			sipStack.logWriter.logDebug("creating new TCPMessageChannel ");
			sipStack.logWriter.logStackTrace();
		}
		mySock = sock;
		peerAddress = mySock.getInetAddress();
		myAddress = msgProcessor.getIPAddress().getHostAddress();
		myClientInputStream = mySock.getInputStream();
		myClientOutputStream = mySock.getOutputStream();
		mythread = new Thread(this);
		mythread.setDaemon(true);
		mythread.setName("TCPMessageChannelThread");
		// Stash away a pointer to our sipStack structure.
		this.sipStack = sipStack;

		this.tcpMessageProcessor = msgProcessor;
		this.myPort = this.tcpMessageProcessor.getPort();
		// Bug report by Vishwashanti Raj Kadiayl
		super.messageProcessor = msgProcessor;
		// Can drop this after response is sent potentially.
		mythread.start();
	}

	/**
	 * Constructor - connects to the given inet address. Acknowledgement --
	 * Lamine Brahimi (IBM Zurich) sent in a bug fix for this method. A thread
	 * was being uncessarily created.
	 * 
	 * @param inetAddr
	 *            inet address to connect to.
	 * @param sipStack
	 *            is the sip sipStack from which we are created.
	 * @throws IOException
	 *             if we cannot connect.
	 */
	protected TCPMessageChannel(InetAddress inetAddr, int port,
			SIPTransactionStack sipStack, TCPMessageProcessor messageProcessor)
			throws IOException {
		if (sipStack.isLoggingEnabled()) {
			sipStack.logWriter.logDebug("creating new TCPMessageChannel ");
			sipStack.logWriter.logStackTrace();
		}
		this.peerAddress = inetAddr;
		this.peerPort = port;
		this.myPort = messageProcessor.getPort();
		this.peerProtocol = "TCP";
		this.sipStack = sipStack;
		this.tcpMessageProcessor = messageProcessor;
		this.myAddress = messageProcessor.getIPAddress().getHostAddress();
		// Bug report by Vishwashanti Raj Kadiayl
		this.key = MessageChannel.getKey(peerAddress, peerPort, "TCP");
		super.messageProcessor = messageProcessor;

	}

	/**
	 * Returns "true" as this is a reliable transport.
	 */
	public boolean isReliable() {
		return true;
	}

	/**
	 * Close the message channel.
	 */
	public void close() {
		try {
			if (mySock != null)
				mySock.close();
			if (sipStack.isLoggingEnabled())
				sipStack.logWriter.logDebug("Closing message Channel " + this);
		} catch (IOException ex) {
			if (sipStack.isLoggingEnabled())
				sipStack.logWriter.logDebug("Error closing socket " + ex);
		}
	}

	/**
	 * Get my SIP Stack.
	 * 
	 * @return The SIP Stack for this message channel.
	 */
	public SIPTransactionStack getSIPStack() {
		return sipStack;
	}

	/**
	 * get the transport string.
	 * 
	 * @return "tcp" in this case.
	 */
	public String getTransport() {
		return "TCP";
	}

	/**
	 * get the address of the client that sent the data to us.
	 * 
	 * @return Address of the client that sent us data that resulted in this
	 *         channel being created.
	 */
	public String getPeerAddress() {
		if (peerAddress != null) {
			return peerAddress.getHostAddress();
		} else
			return getHost();
	}

	protected InetAddress getPeerInetAddress() {
		return peerAddress;
	}

	public String getPeerProtocol() {
		return this.peerProtocol;
	}

	/**
	 * Send message to whoever is connected to us. Uses the topmost via address
	 * to send to.
	 * 
	 * @param msg
	 *            is the message to send.
	 * @param retry
	 */
	private void sendMessage(byte[] msg, boolean retry) throws IOException {
		Socket sock = this.sipStack.ioHandler.sendBytes(this.messageProcessor
				.getIPAddress(), this.peerAddress, this.peerPort,
				this.peerProtocol, msg, retry);
		// Created a new socket so close the old one and stick the new
		// one in its place but dont do this if it is a datagram socket.
		// (could have replied via udp but received via tcp!).
		if (sock != mySock && sock != null) {
			try {
				if (mySock != null)
					mySock.close();
			} catch (IOException ex) {
			}
			mySock = sock;
			this.myClientInputStream = mySock.getInputStream();
			this.myClientOutputStream = mySock.getOutputStream();
			Thread thread = new Thread(this);
			thread.setDaemon(true);
			thread.setName("TCPMessageChannelThread");
			thread.start();
		}

	}

	/**
	 * Return a formatted message to the client. We try to re-connect with the
	 * peer on the other end if possible.
	 * 
	 * @param sipMessage
	 *            Message to send.
	 * @throws IOException
	 *             If there is an error sending the message
	 */
	public void sendMessage(SIPMessage sipMessage) throws IOException {
		byte[] msg = sipMessage.encodeAsBytes();

		long time = System.currentTimeMillis();

		this.sendMessage(msg, sipMessage instanceof SIPRequest);

		if (this.sipStack.serverLog.needsLogging(ServerLog.TRACE_MESSAGES))
			logMessage(sipMessage, peerAddress, peerPort, time);
	}

	/**
	 * Send a message to a specified address.
	 * 
	 * @param message
	 *            Pre-formatted message to send.
	 * @param receiverAddress
	 *            Address to send it to.
	 * @param receiverPort
	 *            Receiver port.
	 * @throws IOException
	 *             If there is a problem connecting or sending.
	 */
	public void sendMessage(byte message[], InetAddress receiverAddress,
			int receiverPort, boolean retry) throws IOException {
		if (message == null || receiverAddress == null)
			throw new IllegalArgumentException("Null argument");
		Socket sock = this.sipStack.ioHandler.sendBytes(this.messageProcessor
				.getIPAddress(), receiverAddress, receiverPort, "TCP", message,
				retry);
		//
		// Created a new socket so close the old one and s
		// Check for null (bug fix sent in by Christophe)
		if (sock != mySock && sock != null) {
			try {
				if (mySock != null)
					mySock.close();
			} catch (IOException ex) {
				/* ignore */
			}
			mySock = sock;
			this.myClientInputStream = mySock.getInputStream();
			this.myClientOutputStream = mySock.getOutputStream();
			// start a new reader on this end of the pipe.
			Thread mythread = new Thread(this);
			mythread.setDaemon(true);
			mythread.setName("TCPMessageChannelThread");
			mythread.start();
		}

	}

	/**
	 * Exception processor for exceptions detected from the parser. (This is
	 * invoked by the parser when an error is detected).
	 * 
	 * @param sipMessage --
	 *            the message that incurred the error.
	 * @param ex --
	 *            parse exception detected by the parser.
	 * @param header --
	 *            header that caused the error.
	 * @throws ParseException
	 *             Thrown if we want to reject the message.
	 */
	public void handleException(ParseException ex, SIPMessage sipMessage,
			Class hdrClass, String header, String message)
			throws ParseException {
		if (sipStack.isLoggingEnabled())
			sipStack.logWriter.logException(ex);
		// Log the bad message for later reference.
		if ((hdrClass != null)
				&& (hdrClass.equals(From.class) || hdrClass.equals(To.class)
						|| hdrClass.equals(CSeq.class)
						|| hdrClass.equals(Via.class)
						|| hdrClass.equals(CallID.class)
						|| hdrClass.equals(RequestLine.class) || hdrClass
						.equals(StatusLine.class))) {
			if (sipStack.isLoggingEnabled()) {
				this.getSIPStack().getLogWriter().logDebug(
						"Encountered Bad Message \n" + sipMessage.toString());
			}
			throw ex;
		} else {
			sipMessage.addUnparsed(header);
		}
	}

	/**
	 * Gets invoked by the parser as a callback on successful message parsing
	 * (i.e. no parser errors).
	 * 
	 * @param sipMessage
	 *            Mesage to process (this calls the application for processing
	 *            the message).
	 */
	public void processMessage(SIPMessage sipMessage) throws Exception {
		try {
			if (sipMessage.getFrom() == null
					|| // sipMessage.getFrom().getTag()
					// == null ||
					sipMessage.getTo() == null
					|| sipMessage.getCallId() == null
					|| sipMessage.getCSeq() == null
					|| sipMessage.getViaHeaders() == null) {
				String badmsg = sipMessage.encode();
				if (sipStack.isLoggingEnabled()) {
					sipStack.logWriter.logDebug(">>> Dropped Bad Msg");
					sipStack.logWriter.logDebug(badmsg);
				}

				return;
			}

			ViaList viaList = sipMessage.getViaHeaders();
			// For a request
			// first via header tells where the message is coming from.
			// For response, this has already been recorded in the outgoing
			// message.
			if (sipMessage instanceof SIPRequest) {
				Via v = (Via) viaList.getFirst();
				Hop hop = sipStack.addressResolver.resolveAddress(v.getHop());
				this.peerPort = hop.getPort();
				this.peerProtocol = v.getTransport();
				try {
					this.peerAddress = mySock.getInetAddress();
					// Check to see if the received parameter matches
					// the peer address and tag it appropriately.

					// JvB: dont do this. It is both costly and incorrect
					// Must set received also when it is a FQDN, regardless whether
					// it resolves to the correct IP address
					// InetAddress sentByAddress = InetAddress.getByName(hop.getHost());
					// JvB: if sender added 'rport', must always set received					
					if ( v.hasParameter(Via.RPORT) 
						|| !hop.getHost().equals(this.peerAddress.getHostAddress())) {
							v.setParameter(Via.RECEIVED, this.peerAddress.getHostAddress() );
					}
					// @@@ hagai
					// JvB: technically, may only do this when Via already contains
					// rport
					v.setParameter(Via.RPORT, Integer.toString(this.peerPort));
				} catch (java.text.ParseException ex) {
					InternalErrorHandler
							.handleException(ex, sipStack.logWriter);
				}
				// Use this for outgoing messages as well.
				if (!this.isCached) {
					((TCPMessageProcessor) this.messageProcessor)
							.cacheMessageChannel(this);
					this.isCached = true;
					String key = IOHandler.makeKey(mySock.getInetAddress(),
							this.peerPort);
					sipStack.ioHandler.putSocket(key, mySock);
				}
			}

			// System.out.println("receiver address = " + receiverAddress);

			// Foreach part of the request header, fetch it and process it

			long receptionTime = System.currentTimeMillis();

			if (sipMessage instanceof SIPRequest) {
				// This is a request - process the request.
				SIPRequest sipRequest = (SIPRequest) sipMessage;
				// Create a new sever side request processor for this
				// message and let it handle the rest.

				if (sipStack.isLoggingEnabled()) {
					sipStack.logWriter.logDebug("----Processing Message---");
				}

				// Check for reasonable size - reject message
				// if it is too long.
				if (sipStack.getMaxMessageSize() > 0
						&& sipRequest.getSize()
								+ (sipRequest.getContentLength() == null ? 0
										: sipRequest.getContentLength()
												.getContentLength()) > sipStack
								.getMaxMessageSize()) {
					SIPResponse sipResponse = sipRequest
							.createResponse(SIPResponse.MESSAGE_TOO_LARGE);
					byte[] resp = sipResponse.encodeAsBytes();
					this.sendMessage(resp, false);
					throw new Exception("Message size exceeded");
				}

				ServerRequestInterface sipServerRequest = sipStack
						.newSIPServerRequest(sipRequest, this);
				if (sipStack.isLoggingEnabled()) {
					sipStack.logWriter.logDebug("---- sipServerRequest = "
							+ sipServerRequest);
				}

				if (sipServerRequest != null) {
					try {
						sipServerRequest.processRequest(sipRequest, this);
					} finally {
						if (sipServerRequest instanceof SIPTransaction) {
							SIPServerTransaction sipServerTx = (SIPServerTransaction) sipServerRequest;
							if (!sipServerTx.passToListener())
								((SIPTransaction) sipServerRequest)
										.releaseSem();
						}
					}
				} else {
					this.sipStack.logWriter
							.logWarning("Dropping request -- could not acquire semaphore in 10 sec");
					SIPResponse  response = sipRequest.createResponse(Response.SERVICE_UNAVAILABLE);
					// Service is overloaded -- send back an error and drop the request.
					this.sendMessage(response);
				}
				if (this.sipStack.serverLog
						.needsLogging(ServerLog.TRACE_MESSAGES)) {

					sipStack.serverLog.logMessage(sipMessage,
							this.getPeerHostPort().toString() , this
							.getMessageProcessor().getIPAddress()
							.getHostAddress()
							+ ":" + this.getMessageProcessor().getPort(),
							false, receptionTime);

				}
			} else {
				SIPResponse sipResponse = (SIPResponse) sipMessage;
				if (sipResponse.getStatusCode() == 100) sipResponse.getTo().removeParameter("tag");
				try {
					sipResponse.checkHeaders();
				} catch (ParseException ex) {
					if (sipStack.isLoggingEnabled())
						sipStack.logWriter.logError("Dropping Badly formatted response message >>> " + sipResponse);
					return;
				}
				// This is a response message - process it.
				// Check the size of the response.
				// If it is too large dump it silently.
				if (sipStack.getMaxMessageSize() > 0
						&& sipResponse.getSize()
								+ (sipResponse.getContentLength() == null ? 0
										: sipResponse.getContentLength()
												.getContentLength()) > sipStack
								.getMaxMessageSize()) {
					if (sipStack.isLoggingEnabled())
						this.sipStack.logWriter
								.logDebug("Message size exceeded");
					return;

				}
				ServerResponseInterface sipServerResponse = sipStack
						.newSIPServerResponse(sipResponse, this);
				if (sipServerResponse != null) {
					try {
						sipServerResponse.processResponse(sipResponse, this);
					} finally {
						if (sipServerResponse instanceof SIPTransaction
								&& !((SIPTransaction) sipServerResponse)
										.passToListener())
							((SIPTransaction) sipServerResponse).releaseSem();
					}
				} else {
					sipStack
							.getLogWriter()
							.logWarning(
									"Application is blocked -- could not acquire semaphore -- dropping response");
				}
			}
		} finally {
		}
	}

	/**
	 * This gets invoked when thread.start is called from the constructor.
	 * Implements a message loop - reading the tcp connection and processing
	 * messages until we are done or the other end has closed.
	 */
	public void run() {
		String message;
		Pipeline hispipe = null;
		// Create a pipeline to connect to our message parser.
		hispipe = new Pipeline(myClientInputStream, sipStack.readTimeout,
				((SIPTransactionStack) sipStack).timer);
		// Create a pipelined message parser to read and parse
		// messages that we write out to him.
		myParser = new PipelinedMsgParser(this, hispipe, this.sipStack
				.getMaxMessageSize());
		// Start running the parser thread.
		myParser.processInput();
		// bug fix by Emmanuel Proulx
		int bufferSize = 4096;
		this.tcpMessageProcessor.useCount++;
		this.isRunning = true;
		try {
			while (true) {
				try {
					byte[] msg = new byte[bufferSize];
					int nbytes = myClientInputStream.read(msg, 0, bufferSize);
					// no more bytes to read...
					if (nbytes == -1) {
						hispipe.write("\r\n\r\n".getBytes("UTF-8"));
						try {
							if (sipStack.maxConnections != -1) {
								synchronized (tcpMessageProcessor) {
									tcpMessageProcessor.nConnections--;
									tcpMessageProcessor.notify();
								}
							}
							hispipe.close();
							mySock.close();
						} catch (IOException ioex) {
						}
						return;
					}
					hispipe.write(msg, 0, nbytes);

				} catch (IOException ex) {
					// Terminate the message.
					try {
						hispipe.write("\r\n\r\n".getBytes("UTF-8"));
					} catch (Exception e) {
						// InternalErrorHandler.handleException(e);
					}

					try {
						if (sipStack.isLoggingEnabled())
							sipStack.logWriter
									.logDebug("IOException  closing sock " + ex);
						try {
							if (sipStack.maxConnections != -1) {
								synchronized (tcpMessageProcessor) {
									tcpMessageProcessor.nConnections--;
									// System.out.println("Notifying!");
									tcpMessageProcessor.notify();
								}
							}
							mySock.close();
							hispipe.close();
						} catch (IOException ioex) {
						}
					} catch (Exception ex1) {
						// Do nothing.
					}
					return;
				} catch (Exception ex) {
					InternalErrorHandler
							.handleException(ex, sipStack.logWriter);
				}
			}
		} finally {
			this.isRunning = false;
			this.tcpMessageProcessor.remove(this);
			this.tcpMessageProcessor.useCount--;
		}

	}

	protected void uncache() {
		this.tcpMessageProcessor.remove(this);
	}

	/**
	 * Equals predicate.
	 * 
	 * @param other
	 *            is the other object to compare ourselves to for equals
	 */

	public boolean equals(Object other) {

		if (!this.getClass().equals(other.getClass()))
			return false;
		else {
			TCPMessageChannel that = (TCPMessageChannel) other;
			if (this.mySock != that.mySock)
				return false;
			else
				return true;
		}
	}

	/**
	 * Get an identifying key. This key is used to cache the connection and
	 * re-use it if necessary.
	 */
	public String getKey() {
		if (this.key != null) {
			return this.key;
		} else {
			this.key = MessageChannel.getKey(this.peerAddress, this.peerPort,
					"TCP");
			return this.key;
		}
	}

	/**
	 * Get the host to assign to outgoing messages.
	 * 
	 * @return the host to assign to the via header.
	 */
	public String getViaHost() {
		return myAddress;
	}

	/**
	 * Get the port for outgoing messages sent from the channel.
	 * 
	 * @return the port to assign to the via header.
	 */
	public int getViaPort() {
		return myPort;
	}

	/**
	 * Get the port of the peer to whom we are sending messages.
	 * 
	 * @return the peer port.
	 */
	public int getPeerPort() {
		return peerPort;
	}

	public int getPeerPacketSourcePort() {
		return this.peerPort;
	}

	public InetAddress getPeerPacketSourceAddress() {
		return this.peerAddress;
	}

	/**
	 * TCP Is not a secure protocol.
	 */
	public boolean isSecure() {
		return false;
	}
}