//---------------------------------------------------------------------------------
// <copyright file="SCPExecutor.cpp" company="LiquidRobotics">
//
// Copyright (c) 2014 Liquid Robotics, Inc.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.
//
// This software is associated with a release.  Liquid Robotics makes no
// representation that the design, interfaces or component distribution will
// remain unchanged in future releases.
//
// </copyright>
//
// <summary>
//   Implementation of the SCPExecutor class, which is responsible for periodically
//	 starting a PPP session, reading a file with a standard name and then executing
//   it, then disconnecting.
// </summary>
//
// <owner>Mike Cookson</owner>
//---------------------------------------------------------------------------------

#include "SCPExecutor.h"
#include "LqrSMACLineMachine.h"
#include "LqrSMACUptime.h"
#include <fstream>
#include <boost/algorithm/string.hpp>
#include <boost/bind.hpp>
#include <boost/lexical_cast.hpp>

using namespace std;
using namespace LiquidRobotics::SMACFramework;

namespace LiquidRobotics {
namespace PPPHostLib {

IMPLEMENT_INNER_EXCEPTION(SCPExecutor, AbortDelay, std::runtime_error);
IMPLEMENT_INNER_EXCEPTION(SCPExecutor, AlreadyInitialized, std::runtime_error);
IMPLEMENT_INNER_EXCEPTION(SCPExecutor, NotInitialized, std::runtime_error);
IMPLEMENT_INNER_EXCEPTION(SCPExecutor, ShutdownRequested, std::runtime_error);

SCPExecutor::SCPExecutor(const SCPExecutorSettings& _settings, Logger::Ptr _logger)
: settings(_settings)
, runningSCP(false)
, logger(_logger)
, queue(_settings.GetQueueCapacity())
, pThread(NULL)
{
}

SCPExecutor::~SCPExecutor()
{
	Shutdown();
}

void SCPExecutor::Initialize()
{
	Logger::Locked(logger)->Info << "SCPExecutor initializing" << endl;

	if (pThread)
		throw AlreadyInitialized("This SCPExecutor instance was already initialized");

	lineMachine.reset(new LineMachine);
	lineMachineConnection = lineMachine->Subscribe(boost::bind(&SCPExecutor::LineHandler, this, _1, _2));

	pThread = new boost::thread(boost::bind(&SCPExecutor::BackgroundThread, this));
}

void SCPExecutor::LineHandler(LineMachine& sender, const string& receivedLine)
{
	const QueueEntry::Type entryType = runningSCP ? QueueEntry::SCP_OUTPUT : QueueEntry::USER_OUTPUT;

	if (logger)
	{
		Logger::Lock logLock(*logger);
		logger->Info << "Received " << entryType << " '" << receivedLine << "'" << endl;
	}

	queue.Write(QueueEntry(entryType, receivedLine));
}

void SCPExecutor::QueueConnect()
{
	Lock lock(*this);

	if (!pThread)
		throw NotInitialized("QueueConnect: this SCPExecutor instance has not been initialized");

	queue.Write(QueueEntry(QueueEntry::ABORT_DELAY));
}

void SCPExecutor::Shutdown()
{
	Lock lock(*this);
	Logger::Locked(logger)->Info << "SCPExecutor starting to shut down" << endl;

	if (pThread)
	{
		boost::thread* pThreadSaved = pThread;

		pThread = NULL;
		queue.Write(QueueEntry(QueueEntry::SHUT_DOWN));
		lock.unlock();

		pThreadSaved->join();
	}
}

SCPExecutor::ScopedConnectionPtr SCPExecutor::Subscribe(ConnectionSignalType::slot_type slot)
{
	ScopedConnectionPtr result;

	result.reset(new ScopedConnection(connectionStateChanged.connect(slot)));
	return result;
}

void SCPExecutor::BackgroundThread()
{
	bool looping = true;
	bool first = true;

	Logger::Locked(logger)->Notice << "SCPExecutor background thread starting" << endl;

	while(pppConnection.reset(), pppHost.reset(), looping)
	{
		try
		{
			if (first)
			{
				first = false;

				float startTime = GetSystemUpTimeSeconds();
				float waitTime = 10;
				float elapsedTime;

				do
				{
					try
					{
						elapsedTime = GetSystemUpTimeSeconds()-startTime;
						float thisDelay = (elapsedTime > waitTime) ? 0 : (waitTime-elapsedTime);

						if (thisDelay)
							Delay(thisDelay, "startup settling");
					}
					catch(AbortDelay& ex) { }

					elapsedTime = GetSystemUpTimeSeconds()-startTime;

				} while(elapsedTime < waitTime);
			}

			Connect();

			if (RunSCPVerb())
				RunUserCommand();
		}
		catch(ShutdownRequested& ex)
		{
			Logger::Locked(logger)->Notice << "SCPExecutor thread received shutdown request and will now exit (" << ex.what() << ")" << endl;
			looping = false;
		}
		catch(exception& ex)
		{
			Logger::Locked(logger)->Error << "SCPExecutor thread caught " << typeid(ex).name() << " '" << ex.what() << "'" << endl;
		}

		if (looping)
			try
			{
				Disconnect();
			}
			catch(ShutdownRequested& ex)
			{
				Logger::Locked(logger)->Notice << "SCPExecutor thread received shutdown request and will now exit (" << ex.what() << ")" << endl;
				looping = false;
			}
			catch(exception& ex)
			{
				Logger::Locked(logger)->Error << "SCPExecutor thread caught " << typeid(ex).name() << " '" << ex.what() << "'" << endl;
			}

		if (looping)
			try
			{
				float nextCallInterval = GetNextCallInterval();

				if (nextCallInterval > 0)
					Delay(nextCallInterval, "call-in interval");
				else
					while(true) Delay(3600, "call-in interval (indefinite)");
			}
			catch(AbortDelay& ex)
			{
				Logger::Locked(logger)->Notice << "Aborted call-in interval wait" << endl;
			}
			catch(ShutdownRequested& ex)
			{
				Logger::Locked(logger)->Notice << "SCPExecutor thread received shutdown request and will now exit (" << ex.what() << ")" << endl;
				looping = false;
			}
			catch(exception& ex)
			{
				Logger::Locked(logger)->Error << "SCPExecutor thread caught " << typeid(ex).name() << " '" << ex.what() << "'" << endl;
			}
	}

	Logger::Locked(logger)->Notice << "SCPExecutor thread exiting" << endl;
}

void SCPExecutor::Connect()
{
	if (logger)
	{
		Logger::Lock logLock(*logger);
		logger->Info << "SCPExecutor starting to connect" << endl;
	}

	PPPHost::Ptr pppHost2 = pppHost;

	if (!pppHost2)
	{
		pppHost.reset(new PPPHost(settings.GetPPPHostSettings(), logger));

		try
		{
			pppConnection = pppHost->Subscribe(boost::bind(&SCPExecutor::PPPHostCallback, this, _1, _2));
		}
		catch(...)
		{
			pppHost.reset();
			throw;
		}

		pppHost2 = pppHost;
	}

	if (pppHost2 && (pppHost2->GetConnectionStatus() != PPPHost::CONNECTED))
	{
		pppHost2->Connect();

		bool looping = true;

		while(looping)
		{
			QueueEntry item = queue.Read();

			switch(item.type)
			{
			case QueueEntry::CONNECTION_STATE_CHANGE:

				if (item.newStatus == PPPHost::CONNECTED)
				{
					string serialNumberFileName = settings.GetModemSerialNumberFilename();
					boost::trim(serialNumberFileName);

					if (serialNumberFileName.length())
					{
						string modemSerialNumber = pppHost2->GetModemSerialNumber();

						{
							Logger::Lock logLock(*logger);
							logger->Info << "Saving modem serial number '" << modemSerialNumber << "' to " << serialNumberFileName << endl;
						}

						ofstream ofs(serialNumberFileName.c_str());

						// if the write stream is no good then throw invalid argument exception
						if (!ofs.good())
						{
							Logger::Lock logLock(*logger);
							logger->Error << "Could not open " << serialNumberFileName << " for write" << endl;
						}
						else
							try
							{
								ofs << modemSerialNumber;
							}
							catch(exception& ex)
							{
								Logger::Lock logLock(*logger);
								logger->Error << "Caught exception " << typeid(ex).name() << " '" << ex.what() << "' while saving modem serial number to file" << endl;
							}
					}

					string simCardFileName = settings.GetModemSimCardFilename();
					boost::trim(simCardFileName);

					if (simCardFileName.length())
					{
						string simCardNumber = pppHost2->GetModemSimCardNumber();

						{
							Logger::Lock logLock(*logger);
							logger->Info << "Saving modem SIM card number '" << simCardNumber << "' to " << simCardFileName << endl;
						}

						ofstream ofs(simCardFileName.c_str());

						// if the write stream is no good then throw invalid argument exception
						if (!ofs.good())
						{
							Logger::Lock logLock(*logger);
							logger->Error << "Could not open " << simCardFileName << " for write" << endl;
						}
						else
							try
							{
								ofs << simCardNumber;
							}
							catch(exception& ex)
							{
								Logger::Lock logLock(*logger);
								logger->Error << "Caught exception " << typeid(ex).name() << " '" << ex.what() << "' while saving SIM card number to file" << endl;
							}
					}

					looping = false;
				}

				break;

			case QueueEntry::SHUT_DOWN:

				throw ShutdownRequested("Shut down detected while awaiting connect");

			default:

				if (logger)
				{
					Logger::Lock logLock(*logger);
					logger->Info << "Unhandled message type " << item.type << " while awaiting connect (ignored)" << endl;
				}

				break;
			}
		}
	}
}

void SCPExecutor::PPPHostCallback(PPPHost& sender, ConnectionStatus connectionStatus)
{
	queue.Write(QueueEntry(QueueEntry::CONNECTION_STATE_CHANGE, connectionStatus));
	connectionStateChanged(sender, connectionStatus);
}

void SCPExecutor::Delay(float seconds, const string& reason)
{
	if (logger)
	{
		Logger::Lock logLock(*logger);
		logger->Info << "SCPExecutor delaying " << seconds << " seconds for " << reason << endl;
	}

	float startTime = GetSystemUpTimeSeconds();
	bool timeout = false;

	while(!timeout)
	{
		const float elapsed = GetSystemUpTimeSeconds()-startTime;
		const unsigned millisecRemaining = (elapsed >= seconds) ? 0 : (unsigned) ceil(1000*(seconds-elapsed));
		QueueType::Lock qlock(queue, boost::defer_lock);

		if (queue.AwaitReadable(millisecRemaining, &qlock, false))
		{
			QueueEntry item = queue.Read();

			switch(item.type)
			{
			case QueueEntry::ABORT_DELAY:

				throw AbortDelay("Aborting Delay since ABORT_DELAY message was received");

			case QueueEntry::SHUT_DOWN:
			{
				stringstream msg;
				msg << "SCPExecutor detected object shutdown while delaying for " << reason;
				throw ShutdownRequested(msg.str());
			}

			default:

				if (logger)
				{
					Logger::Lock logLock(*logger);
					logger->Info << "Unhandled message type " << item.type << " while delaying for " << reason << endl;
				}

				break;
			}
		}
		else
			timeout = true;
	}
}

void SCPExecutor::Disconnect()
{
	if (logger)
	{
		Logger::Lock logLock(*logger);
		logger->Info << "SCPExecutor starting to disconnect" << endl;
	}

	PPPHost::Ptr pppHost2 = pppHost;

	if (pppHost2 && (pppHost2->GetConnectionStatus() != PPPHost::DISCONNECTED))
	{
		pppHost2->Disconnect();

		bool looping = true;

		while(looping)
		{
			QueueEntry item = queue.Read();

			switch(item.type)
			{
			case QueueEntry::CONNECTION_STATE_CHANGE:

				if (item.newStatus == PPPHost::DISCONNECTED)
					looping = false;

				break;

			case QueueEntry::SHUT_DOWN:

				throw ShutdownRequested("Shut down detected while awaiting disconnect");

			default:

				if (logger)
				{
					Logger::Lock logLock(*logger);
					logger->Info << "Unhandled message type " << item.type << " while awaiting disconnect (ignored)" << endl;
				}

				break;
			}
		}
	}

	pppConnection.reset();
	pppHost.reset();
}

bool SCPExecutor::RunSCPVerb()
{
	if (logger)
	{
		Logger::Lock logLock(*logger);
		logger->Info << "SCPExecutor running SCP verb '" << settings.GetSCPVerb() << "'" << endl;
	}

	runningSCP = true;
	lineMachine->Reset();

	AsynchronousProcessExecutor cmdExec(logger, SIGTERM, true, SIGKILL, 500, true, 1024);
	ScopedConnectionPtr execConnection = cmdExec.SubscribeExitSignal(boost::bind(&SCPExecutor::APEExitHandler, this, _1, _2));
	ScopedConnectionPtr execOutputConnection = cmdExec.SubscribeReportSignal(boost::bind(&SCPExecutor::APEReportHandler, this, _1, _2, _3));

	cmdExec.RunAsynch(settings.GetSCPVerb(), true, true);

	bool looping = true;
	bool retval = true;
	bool killed = false;

	while(looping)
	{
		QueueEntry item = queue.Read();

		switch(item.type)
		{
		case QueueEntry::CONNECTION_STATE_CHANGE:

			if (!killed && (item.newStatus == PPPHost::DISCONNECTED))
			{
				cmdExec.Kill();
				killed = true;
			}

			break;

		case QueueEntry::SCP_EXITED:

			if (item.resultCode)
				retval = false;

			looping = false;  // TODO: parse output stream or return code to determine if scp got a file
			break;

		case QueueEntry::SCP_OUTPUT:

			if (item.line.find("No such file or directory") != string::npos)
				retval = false;

			break;

		case QueueEntry::SHUT_DOWN:

			throw ShutdownRequested("Shut down detected while running SCP verb");

		default:

			if (logger)
			{
				Logger::Lock logLock(*logger);
				logger->Info << "Unhandled message type " << item.type << " while running SCP verb (ignored)" << endl;
			}

			break;
		}
	}

	return retval && !killed;
}

void SCPExecutor::APEExitHandler(AsynchronousProcessExecutor& sender, int resultCode)
{
	const QueueEntry::Type entryType = runningSCP ? QueueEntry::SCP_EXITED : QueueEntry::USER_PROGRAM_EXITED;
	bool exited = WIFEXITED(resultCode);
	int exitStatus = exited ? WEXITSTATUS(resultCode) : 1;

	if (logger)
	{
		Logger::Lock logLock(*logger);
		logger->Info << entryType << " exited=" << exited << " exitStatus=" << exitStatus << " signaled=" << WIFSIGNALED(resultCode) << endl;
	}

	queue.Write(QueueEntry(entryType, exitStatus));
}

void SCPExecutor::APEReportHandler(AsynchronousProcessExecutor& sender, const ByteArray& dataReceived, bool fromStdErr)
{
	lineMachine->Feed(dataReceived);
}

void SCPExecutor::RunUserCommand()
{
	if (logger)
	{
		Logger::Lock logLock(*logger);
		logger->Info << "SCPExecutor running user command '" << settings.GetRunVerb() << "'" << endl;
	}

	runningSCP = false;
	lineMachine->Reset();

	AsynchronousProcessExecutor cmdExec(logger, SIGTERM, true, SIGKILL, 500, true, 1024);
	ScopedConnectionPtr execConnection = cmdExec.SubscribeExitSignal(boost::bind(&SCPExecutor::APEExitHandler, this, _1, _2));
	ScopedConnectionPtr execOutputConnection = cmdExec.SubscribeReportSignal(boost::bind(&SCPExecutor::APEReportHandler, this, _1, _2, _3));

	cmdExec.RunAsynch(settings.GetRunVerb(), true, true);

	bool looping = true;
	bool killed = false;

	while(looping)
	{
		QueueEntry item = queue.Read();

		switch(item.type)
		{
		case QueueEntry::CONNECTION_STATE_CHANGE:

			if (!killed && (item.newStatus == PPPHost::DISCONNECTED))
			{
				cmdExec.Kill();
				killed = true;
			}

			break;

		case QueueEntry::USER_PROGRAM_EXITED:

			looping = false;
			break;

		case QueueEntry::SHUT_DOWN:

			throw ShutdownRequested("Shut down detected while running user command");

		default:

			if (logger)
			{
				Logger::Lock logLock(*logger);
				logger->Info << "Unhandled message type " << item.type << " while running user command (ignored)" << endl;
			}

			break;
		}
	}
}

// Lockable concept used internally and usable by external actors
void SCPExecutor::lock()
{
	mutex.lock();
}

bool SCPExecutor::try_lock()
{
	return mutex.try_lock();
}

void SCPExecutor::unlock()
{
	mutex.unlock();
}

float SCPExecutor::GetNextCallInterval()
{
	float nextCallInterval = settings.GetCallInterval();
	string verb = settings.GetNextCallIntervalVerb();

	boost::trim(verb);

	if (verb.length())
	{
		{
			Logger::Lock logLock(*logger);
			logger->Info << "Executing next call interval verb '" << verb << "'" << endl;
		}

		FILE* fp = popen(verb.c_str(), "r");

		if (fp != NULL)
		{
			stringstream stm;
			char c;

			while(c = fgetc(fp), !feof(fp) && !ferror(fp))
				stm << c;

			pclose(fp);

			string content = stm.str();
			boost::trim(content);

			if (content.length())
			{
				try
				{
					nextCallInterval = boost::lexical_cast<float>(content);
				}
				catch(exception& ex)
				{
					Logger::Lock logLock(*logger);
					logger->Error << "Caught " << typeid(ex).name() << " '" << ex.what() << "'"
									 " while converting next call interval content (" << content << ") to float" << endl;
				}
			}
			else
			{
				Logger::Lock logLock(*logger);
				logger->Info << "Next call interval verb returned blank" << endl;
			}
		}
		else
		{
			Logger::Lock logLock(*logger);
			logger->Error << "popen() failed for next call interval verb" << endl;
		}
	}

	{
		Logger::Lock logLock(*logger);
		logger->Info << "Next call interval is " << nextCallInterval << " seconds" << endl;
	}

	return nextCallInterval;
}


ostream& operator<<(ostream& out, const SCPExecutor::QueueEntry::Type& entryType)
{
	switch(entryType)
	{
	case SCPExecutor::QueueEntry::UNSPECIFIED:				out << "UNSPECIFIED";				break;
	case SCPExecutor::QueueEntry::CONNECTION_STATE_CHANGE:	out << "CONNECTION_STATE_CHANGED";	break;
	case SCPExecutor::QueueEntry::SHUT_DOWN:				out << "SHUT_DOWN";					break;
	case SCPExecutor::QueueEntry::SCP_EXITED:				out << "SCP_EXITED";				break;
	case SCPExecutor::QueueEntry::SCP_OUTPUT:				out << "SCP_OUTPUT";				break;
	case SCPExecutor::QueueEntry::USER_OUTPUT:				out << "USER_OUTPUT";				break;
	case SCPExecutor::QueueEntry::USER_PROGRAM_EXITED:		out << "USER_PROGRAM_EXITED";		break;
	default:												out << (int) entryType << "?";		break;
	}

	return out;
}

}
}
