//----------------------------------------------------------------------------
//
//  Copyright (C) Intel Corporation, 2006 - 2007.
//
//  File:       TcpConsumer.cpp
//
//  Contents:   Handles outgoing TCP messages.
//
//  Notes:      
//----------------------------------------------------------------------------

//===================================================
//  INCLUDES
//===================================================
// Include for ignoring SIGPIPE under linux	
#if !defined (ACE_WIN32)	
#include <ace/Signal.h>
#endif /* !(ACE_WIN32) */
#include <ace/Message_Block.h>
#include <ace/SOCK_Stream.h>

#include "TcpSvcHandler.h" 
#include "TcpConsumer.h"
#include "global.h"
#include "OptionsUtils.h"

//===================================================
//  SocksSupplier Implementation
//===================================================
Tcp_Consumer::~Tcp_Consumer()
{
	ACE_DEBUG((MY_DEBUG "----->Tcp_Consumer dtor\n"));
	ACE_DEBUG((MY_TRACE ACE_TEXT("Tcp_Consumer::dtor .message counter remain in queue= %d. (%x)\n"),  _msg_counter, _svc_handler));
	
}

// Constructor (set the svc handler)
Tcp_Consumer::Tcp_Consumer(Tcp_Svc_Handler* handler): _msg_counter(0)
{
	_svc_handler = handler;
}

//-----------------------------------------
// Get reference to socket stream
//-----------------------------------------
ACE_SOCK_Stream& Tcp_Consumer::getPeer(void)  const
{
	return (ACE_SOCK_Stream &)_svc_handler->peer();
}

// Put message in this tunnel handler queue
STATUS Tcp_Consumer::putq(ACE_Message_Block *mb, ACE_Time_Value *tv)
{
	ACE_TRACE(ACE_TEXT("Tcp_Consumer::putq"));
	if (_svc_handler == NULL)
		return STATUS_FATAL_ERROR;

	ACE_DEBUG((MY_TRACE ACE_TEXT("Tcp_Consumer:putq (%x)\n"), _svc_handler));
	{
		// every time we enqueue message, a new notification is created in reactor.
		// we count those notification so we will know when the reactor does'nt contain 
		// any notification dispacher for this handler (important for deletion)
		ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex ,
			lock2,
			_svc_handler->_dispatch_output_counter_mutex,
			STATUS_LOCK_FAILURE);

		_svc_handler->_dispatch_output_counter++;
	}
	if (_svc_handler->msg_queue()->is_full())
	{
		ACE_DEBUG((MY_DEBUG 
			ACE_TEXT("Tcp_Consumer: Queue is full\n")));
	}
	_msg_counter++;
	
	if (_svc_handler->putq (mb, tv) == -1)
	{
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("TCP consumer failed to put message in queue\n")));

		if (errno == EWOULDBLOCK)
			return STATUS_QUEUE_TIMEOUT;
		else
			return STATUS_FAILURE;
	}
	return STATUS_SUCCESS;
}

//-----------------------------------------
// Handle output message.
// goes over the message queue - and send all its content.
//-----------------------------------------
STATUS Tcp_Consumer::handle_output(ACE_HANDLE h)
{
	ACE_TRACE(ACE_TEXT("Tcp_Supplier::handle_output"));
	if (getPeer().get_handle() == ACE_INVALID_HANDLE)
	{
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT ("socket handler is INVALID\n")));
		return STATUS_NETWORK_ERROR;
	}

	ACE_Message_Block *mb = NULL;
	int res = 0;
	int err = 0;
	int qcount = 0 ;
	STATUS rep = STATUS_SUCCESS;

	if (_svc_handler->msg_queue()->message_count() <= 0)
	{
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT ("tcpConsumer:handle_output - message count <=0\n")));
		return STATUS_SUCCESS;
	}
	_msg_counter--;
	
	qcount = _svc_handler->getq(mb);

	if (mb != 0 && qcount != -1) // qcount > 0
	{
		//if close connection message
		if ( mb->msg_type() == ACE_Message_Block::MB_HANGUP)
		{
		
			
			ACE_DEBUG((MY_DEBUG
				ACE_TEXT ("sending close message to mc\n")));
	
			// if there are more output notifications for this handler in reactor, we dont want to return STATUS_CONNECTION_CLOSED
			// because it can cause to delete this handler (while reactor still holding it)
			// only the last dispacher return STATUS_CONNECTION_CLOSED;
			ACE_GUARD_RETURN(	ACE_Recursive_Thread_Mutex ,
				dispatch_output_counter_lock,
				_svc_handler->_dispatch_output_counter_mutex,
				STATUS_LOCK_FAILURE);

			ACE_READ_GUARD_RETURN(ACE_Recursive_Thread_Mutex ,
				active_lock,
				_svc_handler->_active_mutex,
				STATUS_LOCK_FAILURE);

			ACE_WRITE_GUARD_RETURN(	ACE_RW_Thread_Mutex ,
				state_lock,
				_svc_handler->_state_mutex,
				STATUS_LOCK_FAILURE);

			_svc_handler->_tcp_state = Tcp_Svc_Handler::DISCONNECTED;
			if ((_svc_handler->_dispatch_output_counter == 0) && (_svc_handler->_active_counter == 1))
			{
				ACE_DEBUG((MY_TRACE ACE_TEXT("Tcp_Consumer::handle_output - got hangup message, and it IS the last dispacher\n")));
				rep = STATUS_CONNECTION_CLOSED;
			}
			else
			{
				ACE_DEBUG((MY_TRACE ACE_TEXT("Tcp_Consumer::handle_output - got hangup message, but its NOT the last dispacher, or some other thread is handling output for this service [%x]\n"), _svc_handler));
			}
		}

		else 
		{
			ACE_READ_GUARD_RETURN(	ACE_RW_Thread_Mutex ,
				lock,
				_svc_handler->_state_mutex,
				STATUS_LOCK_FAILURE);
			if (_svc_handler->state() == Tcp_Svc_Handler::DISCONNECTING)
			{
				ACE_DEBUG((MY_DEBUG
					ACE_TEXT ("DISCONNECT state - ignoring data\n")));

				rep = STATUS_SUCCESS;
			}
		
			else
			{
				lock.release();
// Handle SIGPIPE under linux	
#if !defined (ACE_WIN32)	
			 	ACE_Sig_Action no_sigpipe((ACE_SignalHandler) SIG_IGN);
				ACE_Sig_Action original_action;
				no_sigpipe.register_action(SIGPIPE, &original_action);
#endif /* !(ACE_WIN32) */

				res = this->getPeer().send_n(mb, &ACE_Time_Value(*getMaxTunnelTimeout()));

#if !defined (ACE_WIN32)	
			 	no_sigpipe.restore_action(SIGPIPE, original_action);	
#endif /* !(ACE_WIN32) */
			
				if ( (res < 0) || (res != mb->length()))
				{
					// Failed to send the entire message
					ACE_DEBUG((MY_DEBUG
						ACE_TEXT("TCPConsumer failed to send the entire message\n")));				
					rep = STATUS_NETWORK_ERROR;	
				}
			}
		}
		ACE_Message_Block::release(mb);
		
	}
	return rep;
}

//-----------------------------------------
// add message block to message queue, and
// register to write handler.
//-----------------------------------------
STATUS Tcp_Consumer::sendData(ACE_Message_Block* mb)
{
	ACE_TRACE(ACE_TEXT("Tcp_Consumer::sendData"));
	
	STATUS rep;

	rep = putq(mb);

	if (rep != STATUS_SUCCESS)  // failed to putq
	{
		ACE_Message_Block::release(mb);
		ACE_ERROR_RETURN((MY_DEBUG 
			ACE_TEXT("Tcp_Consumer::sendData - failed to send message block to message queue\n")), 
			rep);
	}

	return STATUS_SUCCESS;
}

//-----------------------------------------
// Close consummer
// This method is called in two cases:
// 1. By the Tunnel_Supplier (after a channel has been established) when 
//	  (a) AMT closed the channel. (b) The tunnel is closing.
// 2. By the SOCKS_Supplier if the channel creation failed.
//-----------------------------------------
void Tcp_Consumer::connectionClose()
{
	ACE_TRACE(ACE_TEXT("Tcp_Supplier::connectionClose"));
	
	ACE_Message_Block *mb;
	ACE_NEW(mb,ACE_Message_Block(0, ACE_Message_Block::MB_HANGUP));
	
	if (sendData(mb) != STATUS_SUCCESS)
	{
		ACE_DEBUG ((MY_DEBUG
				ACE_TEXT ("failed to send close message\n")));
	}
}

//-----------------------------------------
// close consummer
//-----------------------------------------
void Tcp_Consumer::handle_close()
{
	ACE_TRACE(ACE_TEXT("Tcp_Supplier::close"));
}