//******************************************************************************
//
// Restricted Rights
// WARNING: This is a restricted distribution Zeligsoft file.
//
//******************************************************************************
//
// Copyright (C) 2006 Zeligsoft Inc. - All Rights Reserved
//

//******************************************************************************
//
//     FILE: zceSendReceiveDataWorker.cpp
//
//******************************************************************************

#include "zceSendReceiveDataWorker.h"

#include "zceSendReceiveDataPorts.h"

#include <sys/socket.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <netinet/in.h>
#include <unistd.h>
#include <iostream>
#include "../../Constants.h"
#include "../../Support.h"
#include "../../Configuration.h"

zceSendReceiveDataWorker* theWorker;

void zceSendReceiveDataWorker::workerThreadFunc()
{
	mIsThreadRunning = true;
	logToFile("Debug", "SendReceiveData", "Thread started!\n");

	while(1) // mIsDone initialize'de 0, release'de 1 yapılır.
	{
//		if(mIsStarted == 0)
//		{
//		   logToFile("Debug", "SendReceiveData", "Thread is breaking because of mIsStarted!\n");
//		   break;
//		}

		if(mIsDone == 1)
		{
		   logToFile("Debug", "SendReceiveData", "Thread is breaking because of mIsDone!\n");
           break;
		}

		if(isConnectedToTCPServer())
		{
			// Generate packet
			int packetLength = CONST_PACKET_SIZE;
			CF::OctetSequence_var dataPacketOctet = CF::OctetSequence(packetLength + 1);
			dataPacketOctet->length(CONST_PACKET_SIZE);

			CF::Properties options(0);
			options.length(0);

			// Create and zero out the buffer.
			int bufLength = CONST_PACKET_SIZE + 1;
			char buf[bufLength];
			memset(buf, 0x0, bufLength);

			// Read the modified line back from the server.
			int isReceived = recv(mSocketDescriptor, buf, CONST_PACKET_SIZE, MSG_WAITALL);
			if (isReceived > 0)
			{
				// Generate message
				for (unsigned int i = 0; i < CONST_PACKET_SIZE; i++)
				{
					dataPacketOctet[i] = buf[i];
				}

				// Send received data to the WF
				portDataOut_->processOctetMsg(dataPacketOctet, options);
			}
			else
			{
				logToFile("Error", "SendReceiveData", "Cannot get response from remote waveform");
				break;
			}
		}
		else
		{
			logToFile("Debug", "SendReceiveData", "Not connected to remote waveform");
		}

	} // while((!mIsDone) && (mIsStarted == 1))

	disconnectFromTCPServer();

	mIsThreadRunning = false;
	logToFile("Debug", "SendReceiveData", "Thread exited!\n");
} // workerThreadFunc

void* pushData(void* param)
{
	theWorker->workerThreadFunc();
	pthread_exit(0);
}

void zceSendReceiveDataWorker::connectToTCPServer()
{
	unsigned short int serverPort;
	struct sockaddr_in serverAddress;
	struct hostent *hostInfo;

	char* serverIp = strdup(CONF_TCP_SERVER_IP);
	serverPort = CONF_TCP_SERVER_PORT;

	// gethostbyname() takes a host name or ip address in "numbers and
	// dots" notation, and returns a pointer to a hostent structure,
	// which we'll need later.  It's not important for us what this
	// structure is actually composed of.
	hostInfo = gethostbyname(serverIp);
	if (hostInfo == NULL)
	{
		ostringstream message;
		message << "problem interpreting host: " << serverIp << "\n";
		logToFile("Error", "SendReceiveData", message.str());
		return;
	}

	// Create a socket.  "AF_INET" means it will use the IPv4 protocol.
	// "SOCK_STREAM" means it will be a reliable connection (i.e., TCP;
	// for UDP use SOCK_DGRAM), and I'm not sure what the 0 for the last
	// parameter means, but it seems to work.
	mSocketDescriptor = socket(AF_INET, SOCK_STREAM, 0);
	if (mSocketDescriptor < 0)
	{
		ostringstream message;
		message << "cannot create socket\n";
		logToFile("Error", "SendReceiveData", message.str());
		return;
	}

	// Connect to server.  First we have to set some fields in the
	// serverAddress structure.  The system will assign me an arbitrary
	// local port that is not in use.
	serverAddress.sin_family = hostInfo->h_addrtype;
	memcpy((char *) &serverAddress.sin_addr.s_addr, hostInfo->h_addr_list[0],
			hostInfo->h_length);
	serverAddress.sin_port = htons(serverPort);

	if (connect(mSocketDescriptor, (struct sockaddr *) &serverAddress, sizeof(serverAddress)) < 0)
	{
		ostringstream message;
		message << "cannot connect\n";
		logToFile("Error", "SendReceiveData", message.str());
		return;
	}
	else
	{
		mIsConnectedToTCP = true;
		ostringstream message;
		message << "connected to server\n";
		logToFile("Debug", "SendReceiveData", message.str());
	}

	free(serverIp);
}

void zceSendReceiveDataWorker::connectToTCPClient()
{
	unsigned short int listenPort;
	struct sockaddr_in serverAddress;
	struct sockaddr_in clientAddress;
	socklen_t clientAddressLength;
	listenPort = CONF_TCP_SERVER_PORT;

	// Create socket for listening for client connection requests.
	mListenSocket = socket(AF_INET, SOCK_STREAM, 0);
	if (mListenSocket < 0)
	{
		logToFile("Error", "SendReceiveData",
				"cannot create listen socket for WF");
		return;
	}

	// Bind listen socket to listen port.  First set various fields in
	// the serverAddress structure, then call bind().
	// htonl() and htons() convert long integers and short integers
	// (respectively) from host byte order (on x86 this is Least
	// Significant Byte first) to network byte order (Most Significant
	// Byte first).
	serverAddress.sin_family = AF_INET;
	serverAddress.sin_addr.s_addr = htonl(INADDR_ANY);
	serverAddress.sin_port = htons(listenPort);

	if (bind(mListenSocket, (struct sockaddr *) &serverAddress,	sizeof(serverAddress)) < 0)
	{
		logToFile("Error", "SendReceiveData", "cannot create bind socket for WF");
		return;
	}

	// Wait for connections from clients.
	// This is a non-blocking call; i.e., it registers this program with
	// the system as expecting connections on this socket, and then
	// this thread of execution continues on.
	listen(mListenSocket, 1);

	ostringstream oss;
	oss << "Waiting for TCP connection of WF on port " << listenPort << " ...\n";
	logToFile("Debug", "SendReceiveData", oss.str());

	// Accept a connection with a client that is requesting one.  The
	// accept() call is a blocking call; i.e., this thread of
	// execution stops until a connection comes in.
	// connectSocket is a new socket that the system provides,
	// separate from listenSocket.  We *could* accept more
	// connections on listenSocket, before connectSocket is closed,
	// but this program doesn't do that.
	clientAddressLength = sizeof(clientAddress);
	mSocketDescriptor = accept(mListenSocket, (struct sockaddr *) &clientAddress, &clientAddressLength);
	if (mSocketDescriptor < 0)
	{
		logToFile("Error", "SendReceiveData", "cannot accept connection from WF");
		return;
	}

	mIsConnectedToTCP = true;

	// Show the IP address of the client.
	// inet_ntoa() converts an IP address from binary form to the
	// standard "numbers and dots" notation.
	cout << "  connected to " << inet_ntoa(clientAddress.sin_addr);

	// Show the client's port number.
	// ntohs() converts a short int from network byte order (which is
	// Most Significant Byte first) to host byte order (which on x86,
	// for example, is Least Significant Byte first).
	cout << ":" << ntohs(clientAddress.sin_port) << "\n";
}

void zceSendReceiveDataWorker::disconnectFromTCPServer()
{
	if (isConnectedToTCPServer())
	{
		close(mSocketDescriptor);
		close(mListenSocket);
		mIsConnectedToTCP = false;
	}
}

bool zceSendReceiveDataWorker::isConnectedToTCPServer()
{
	return mIsConnectedToTCP;
}

zceSendReceiveDataWorker::zceSendReceiveDataWorker( const SendReceiveData_params& execParams, zcePushPorts_OctetSeqConsumerUsesPort* portDataOut ZCE_EXC_ENV_ARG ):
portDataOut_(portDataOut)
#if defined(DEBUG_MSGS)
, debug_(0)
#endif

{
	ZCE_ASSERT_EXCEPTION_VOID;
#if defined(DEBUG_MSGS)

	debug_.rdbuf(std::cout.rdbuf());
#endif

	initExecParams(execParams ZCE_EXC_ENV_PARAM );

	// TODO : Implement constructor for zceSendReceiveDataWorker
}

zceSendReceiveDataWorker::~zceSendReceiveDataWorker()
{
	// TODO : Implement destructor for zceSendReceiveDataWorker
}

void zceSendReceiveDataWorker::initExecParams( const SendReceiveData_params& params ZCE_EXC_ENV_ARG)
{

	ZCE_ASSERT_EXCEPTION_VOID;
	execParams_.NAMING_CONTEXT_IOR = params.NAMING_CONTEXT_IOR;
	execParams_.NAME_BINDING = params.NAME_BINDING;
	execParams_.COMPONENT_IDENTIFIER = params.COMPONENT_IDENTIFIER;

	// TODO: perform some action based on these exec parameters
}

/*!	\brief
 *
 *	 Describe operation here.
 *
 *	\throw ::CF::LifeCycle::InitializeError
 *
 */
void zceSendReceiveDataWorker::CF_LifeCycle_initialize( int portIdentifier ZCE_EXC_ENV_ARG )
{
	ZCE_ASSERT_EXCEPTION_VOID;

	// Load initial configuration
	loadConfiguration();

	mIsDone=0;
	mIsStarted = 0;
	mPacketID = 0;
	mIsThreadRunning = false;
	mSocketDescriptor = -1; // Not connected
	mIsConnectedToTCP = false; // Not connected
	mThreadId = 0;
}

/*!	\brief
 *
 *	 Describe operation here.
 *
 *	\throw ::CF::LifeCycle::ReleaseError
 *
 */
void zceSendReceiveDataWorker::CF_LifeCycle_releaseObject( int portIdentifier ZCE_EXC_ENV_ARG )
{
	ZCE_ASSERT_EXCEPTION_VOID;

	mIsStarted=0;
	mIsDone=1;
}

/*!
 * \brief The start operation is provided to command a Resource implementing this
 * interface to start internal processing.
 *
 * Specification reference:  3.1.3.1.6.5.2
 *
 * The start operation puts the Resource in an operating condition.
 *
 * \return This operation does not return a value.
 *
 * \throw StartError The start operation shall raise the StartError exception if
 * an error occurs while starting the resource.
 *
 */
void zceSendReceiveDataWorker::CF_Resource_start( int portIdentifier ZCE_EXC_ENV_ARG )
{
	ZCE_ASSERT_EXCEPTION_VOID;

	theWorker = this;

	if(!CONF_IS_LOOP_DATA)
	{
		if (!mIsThreadRunning)
		{
			if(!isConnectedToTCPServer())
			{
				if(CONF_IS_SERVER)
				{
					connectToTCPClient();
				}
				else
				{
					connectToTCPServer();
				}
			}

			if(isConnectedToTCPServer())
			{
				// Join with main thread if previous thread is running
				if(mThreadId != 0)
				{
					logToFile("Debug", "SendReceiveData", "Joining thread\n");
					pthread_join(mThreadId, NULL);
					logToFile("Debug", "SendReceiveData", "Joined thread\n");
				}
				logToFile("Debug", "SendReceiveData", "Creating thread\n");
				pthread_create(&mThreadId, NULL, pushData, NULL);
				logToFile("Debug", "SendReceiveData", "Created thread\n");
			}
		}
	}

	mIsStarted = 1;
}

/*!
 * \brief The stop operation is provided to command a Resource implementing this
 * interface to stop internal processing.
 *
 * Specification reference:  3.1.3.1.6.5.1
 *
 * The stop operation shall disable all current operations and put the Resource in
 * a non-operating condition. Subsequent configure, query, and start operations
 * are not inhibited by the stop operation.
 *
 * \return This operation does not return a value.
 *
 * \throw StopError The stop operation shall raise the StopError exception if an
 * error occurs while stopping the resource.
 *
 */
void zceSendReceiveDataWorker::CF_Resource_stop( int portIdentifier ZCE_EXC_ENV_ARG )
{
	ZCE_ASSERT_EXCEPTION_VOID;

	mIsStarted = 0;
}

/*!	\brief
 *
 *	 Describe operation here.
 *
 *	\param msg ( ::CF::OctetSequence )
 *
 *	\param options ( ::CF::Properties )
 *
 */
void zceSendReceiveDataWorker::PushPorts_OctetSeqConsumer_processOctetMsg( int portIdentifier, const CF::OctetSequence& msg, const CF::Properties& options ZCE_EXC_ENV_ARG )
{
	ZCE_ASSERT_EXCEPTION_VOID;

	if((mIsStarted == 1) && (mIsDone == 0))
	{
		if(!CONF_IS_LOOP_DATA)
		{
			if(isConnectedToTCPServer())
			{
				int bufLength = CONST_PACKET_SIZE;
				char buf[bufLength];

				// Generate message
				for(unsigned int i = 0; i < msg.length(); i++)
				{
					buf[i] = msg[i];
				}

				// Send to TCP Server
				if(send(mSocketDescriptor, buf, CONST_PACKET_SIZE, MSG_WAITALL) < 0)
				{
					ostringstream message;
					message << "cannot send data\n";
					logToFile("Debug", "SendReceiveData", message.str());
					//close(mSocketDescriptor);
					//exit(1);
				}
				else
				{
					ostringstream message1;
					message1 << "Sended msg.len = " << msg.length() << "cons_pack = " << CONST_PACKET_SIZE << "-\n";
					logToFile("Debug", "SendReceiveData", message1.str());
				}
			}
		}
		else
		{
			// LOOP
			portDataOut_->processOctetMsg(msg, options);
		}

		ostringstream message;
		message << "send data length is " << msg.length() << "\n";
		logToFile("Debug", "SendReceiveData", message.str());
	}
}

void zceSendReceiveDataWorker::setportDataOut(
		zcePushPorts_OctetSeqConsumerUsesPort& portDataOut)
{
	portDataOut_ = &portDataOut;
}

#if defined(DEBUG_MSGS)
void zceSendReceiveDataWorker::setDebugStream(const std::ostream& stream)
{
	debug_.rdbuf(stream.rdbuf());
}

void zceSendReceiveDataWorker::debugMsg(const std::string& msg)
{
	if(debug_ != NULL)
	{
		debug_ << msg << std::endl;
	}
}
#endif

// End-Of-File
//------------------------------------------------------------------------------
