//--------------------------------------------------------------------------------------------------------------------
// <copyright file="LqrSMACDomainSocket.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 DomainSocket class, which encapsulates a Unix domain socket
//   (i.e. a socket which is identified with a string-based name that is local to the
//   machine)
//
//   Example:
//
//		#include <iostream>
//		#include <boost/algorithm/string.hpp>
//		#include "DomainSocket.h"
//		#include "LqrSMACByteArray.h"
//		#include "LqrSMACSemaphore.h"
//
//		class MyClient
//		{
//		public:
//
//			MyClient(const string& _connectionName)
//			: sock(_connectionName, 1024)
//			{
//			}
//
//			static void Run()
//			{
//				MyClient client("socktest");
//				client.Invoke("Hello, world!");
//			}
//
//			void Invoke(const string& request)
//			{
//				sock.ConnectToServer();
//				sock.Write(request, true);
//
//				string s;
//				unsigned n = 1;
//
//				try
//				{
//					do
//					{
//						sock.Read(s);
//						cout << n++ << " " << s.length() << " bytes: " << s << endl;
//					} while(s.length());
//				}
//				catch(...)
//				{
//				}
//			}
//
//		private:
//
//			DomainSocket sock;
//			DomainSocket::ScopedConnectionPtr sockConn;
//		};
//
//		class MyServer
//		{
//		public:
//
//			MyServer(const string& _connectionName)
//			: sock(_connectionName, 1024)
//			{
//			}
//
//			static void Run()
//			{
//				MyServer server("socktest");
//
//				server.Invoke();
//			}
//
//			void Invoke()
//			{
//				DomainSocket::ScopedConnectionPtr subscription = sock.Subscribe(boost::bind(&MyServer::Callback, this, _1, _2));
//				sock.StartServer();
//
//				Semaphore sem;
//				sem.SetCount(0);
//				sem.Wait();
//
//				sock.StopServer();
//			}
//
//		private:
//
//			void Callback(DomainSocket& sender, DomainSocket::Ptr newSocket)
//			{
//				string input;
//
//				newSocket->Read(input);
//				newSocket->Write(string("1 ")+input);
//				sleep(1);
//				newSocket->Write(string("2 ")+input);
//				sleep(1);
//				newSocket->Write(string("3 ")+input);
//				sleep(1);
//			}
//
//			DomainSocket sock;
//			DomainSocket::ScopedConnectionPtr sockConn;
//		};
//
//		int main(int argc, char* argv[])
//		{
//			if ((argc == 2) && boost::iequals(argv[1], "--server"))
//				MyServer::Run();
//			else if ((argc == 2) && boost::iequals(argv[1], "--client"))
//				MyClient::Run();
//			else
//				cerr << "Invalid arguments" << endl;
//
//			return 0;
//		}
// </summary>
//
// <owner>Mike Cookson</owner>
//--------------------------------------------------------------------------------------------------------------------

#include "LqrSMACDomainSocket.h"
#include "LqrSMACGetErrorDescription.h"
#include <algorithm>
#include <sstream>
#include <errno.h>
#include <fcntl.h>
#include <stdlib.h>
#include <unistd.h>

using namespace std;
using namespace LiquidRobotics::SMACFramework;

#define DONOEINTR(cmd,statvar,errtemp) \
			do														\
			{														\
				statvar = (cmd);									\
				errtemp = errno;									\
			} while((statvar < 0) && (errtemp == EINTR))

inline string MakeSystemErrorMessage(const string& whatArg, int _errNum)
{
	stringstream msg;
	msg << whatArg << " (error #" << _errNum << ": " << GetErrorDescription(_errNum) << ")";
	string msgStr = msg.str();
	return msgStr;
}

namespace LiquidRobotics {
namespace SMACFramework {

DomainSocket::SystemError::SystemError(const string& whatArg)
: runtime_error(MakeSystemErrorMessage(whatArg, errno))
, errNum(errno)
{
}

DomainSocket::SystemError::SystemError(const string& whatArg, int _errNum)
: runtime_error(MakeSystemErrorMessage(whatArg, _errNum))
, errNum(_errNum)
{
}

string DomainSocket::SystemError::describe(int errNum)
{
	return GetErrorDescription(errNum);
}

DomainSocket::DomainSocket(const string& _advertisedName, unsigned _readWriteSize)
: advertisedName(_advertisedName)
, readWriteSize(_readWriteSize)
, socketfd(-1)
, numRequests(3)
, listenerRunning(false)
{
	MakeAddress(advertisedName, sa, salen);
}

DomainSocket::~DomainSocket()
{
	if (socketfd != -1)
		StopServer();

	if (socketfd != -1)
	{
		close(socketfd);
		socketfd = -1;
	}
}

void DomainSocket::ConnectToServer()
{
	if (socketfd != -1)
		throw runtime_error("DomainSocket::ConnectToServer: socket is open");

	sockaddr_un clientsa;
	unsigned clientsalen;
	int err;

	MakeAddress(advertisedName, clientsa, clientsalen);

	try
	{
		DONOEINTR(socket(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0), socketfd, err);

		if (socketfd < 0)
			throw SystemError("DomainSocket::ConnectToServer: socket(AF_UNIX,SOCK_STREAM,0) failed", err);

		int status;

		DONOEINTR(connect(socketfd, (const sockaddr *) &clientsa, clientsalen), status, err);

		if (status < 0)
			throw SystemError("DomainSocket::ConnectToServer: connect failed", err);

	}
	catch(...)
	{
		if (socketfd != -1)
		{
			close(socketfd);
			socketfd = -1;
		}

		throw;
	}
}

const string& DomainSocket::GetAdvertisedName() const
{
	return advertisedName;
}

int DomainSocket::GetFileDescriptor() const
{
	return socketfd;
}

DomainSocket::Ptr DomainSocket::ListenForConnection(int ms)
{
	if (socketfd == -1)
		throw runtime_error("DomainSocket::ListenForConnection: the socket is not open");

	DomainSocket::Ptr result;
	int waitStatus = Wait(socketfd, pipes.ReadPipe(), ms);

	if (waitStatus == 1)
	{
		int acceptfd = accept(socketfd, NULL, NULL);

		if (acceptfd >= 0)
		{
			int status0;
			int err;

			DONOEINTR(fcntl(acceptfd, F_GETFD), status0, err);

			int status = status0;

			if (status >= 0)
			{
				DONOEINTR(fcntl(acceptfd, F_SETFD, status0 | FD_CLOEXEC), status, err);

				result.reset(new DomainSocket(advertisedName, readWriteSize));
				result->socketfd = acceptfd;
			}
			else
				close(acceptfd);
		}
	}

	return result;
}

int DomainSocket::Wait(int fd0, int fd1, int ms)
{
	int status;
	int err;

	do
	{
		fd_set readSet;
		FD_ZERO(&readSet);

		const int fdmax = max(fd0, fd1);
		FD_SET(fd0, &readSet);
		FD_SET(fd1, &readSet);

		fd_set exSet = readSet;

		timeval tv;
		timeval* tvp = NULL;

		if (ms >= 0)
		{
			tv.tv_sec = ms/1000;
			tv.tv_usec = 1000*(ms%1000);
			tvp = &tv;
		}

		status = select(fdmax+1, &readSet, NULL, &exSet, tvp);
		err = errno;

		if ((status == -1) && (err != EINTR))
		{
			stringstream msg;
			msg << "select failed and returned " << GetErrorDescription(err);

			string s = msg.str();
			throw runtime_error(s);
		}

	} while(status < 0);

	return status;
}

void DomainSocket::Read(string& data)
{
	int status;
	int err;

	data.clear();
	data.resize(readWriteSize, '\xEE');
	DONOEINTR(read(socketfd, const_cast<char*>(data.data()), data.length()), status, err);

	if (status < 0)
	{
		data.clear();
		throw SystemError("DomainSocket::Read: failed read from socket", err);
	}

	data.resize(status);
}

void DomainSocket::ShutdownWrites()
{
	if (socketfd == -1)
		throw runtime_error("DomainSocket::ShutdownWrites: the socket is closed");

	int status;
	int err;

	DONOEINTR(shutdown(socketfd, SHUT_WR), status, err);

	if (status < 0)
		throw SystemError("DomainSocket::ShutdownWrites: failed while shutting down writes to the connection", err);
}

DomainSocket::ScopedConnectionPtr DomainSocket::Subscribe(SignalType::slot_type callback)
{
	ScopedConnectionPtr result;
	result.reset(new ScopedConnection(sig.connect(callback)));
	return result;
}

void DomainSocket::StartServer(bool enableCallbacks, int _numRequests)
{
	if (listeningThread)
		throw runtime_error("DomainSocket::StartServer: the listener thread has already been started");

	if (socketfd != -1)
		throw runtime_error("DomainSocket::StartServer: the socket is already open");

	try
	{
		numRequests = _numRequests;

		int err;

		DONOEINTR(socket(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0), socketfd, err);

		if (socketfd < 0)
			throw SystemError("DomainSocket::StartServer: socket(AF_UNIX,SOCK_STREAM,0) failed", err);

		int status;

		DONOEINTR(bind(socketfd, (const sockaddr*) &sa, salen), status, err);

		if (status < 0)
			throw SystemError("DomainSocket::StartServer: bind() failed", err);

		DONOEINTR(listen(socketfd, numRequests), status, err);

		if (status == -1)
			throw SystemError("DomainSocket::StartServer: listen() failed", err);

		startSem.SetCount(0);
		listenerRunning = true;
		listeningThread.reset(new Thread(boost::bind(&DomainSocket::ListenerThread, this)));
		startSem.Wait();
	}
	catch(...)
	{
		if (socketfd != -1)
		{
			close(socketfd);
			socketfd = -1;
		}

		throw;
	}
}

void DomainSocket::StopServer()
{
	if (listeningThread)
	{
		listenerRunning = false;

		if (pipes.WritePipeOpen())
		{
			char c = 'A';
			if (write(pipes.WritePipe(), &c, 1));
		}

		close(socketfd);
		socketfd = -1;
		listeningThread->join();
		listeningThread.reset();
	}
}

void DomainSocket::Write(const string& data, bool shutdown)
{
	Write(data.data(), data.length(), shutdown);
}

void DomainSocket::Write(const void* pData, unsigned numBytes, bool shutdown)
{
	if (!pData)
		throw invalid_argument("DomainSocket::Write: nil pointer");

	if (numBytes)
	{
		int status;
		int err;
		DONOEINTR(write(socketfd, pData, numBytes), status, err);

		if (status < 0)
			throw SystemError("DomainSocket::Write: failed write to socket", err);

		if (status != (int) numBytes)
		{
			stringstream msg;
			msg << "DomainSocket::Write: attempted to write " << numBytes << " bytes to socket, but only "
				<< status << " were accepted";
			string msgStr = msg.str();
			throw runtime_error(msgStr);
		}
	}

	if (shutdown)
		ShutdownWrites();
}

void DomainSocket::ListenerThread()
{
	startSem.Post();

	while(listenerRunning)
	{
		Ptr newItem = ListenForConnection();

		if (newItem)
			sig(*this, newItem);
	}
}

void DomainSocket::MakeAddress(const string& advertisedName, sockaddr_un& sa, socklen_t& salen)
{
	const unsigned minimal = sizeof(sa)-sizeof(sa.sun_path)+1;
	const unsigned added = min(advertisedName.length(), sizeof(sa.sun_path)-1);

	salen = minimal+added;

	memset(&sa, '\0', sizeof(sa));
	sa.sun_family = AF_UNIX;
	sa.sun_path[0] = '\0';

	if (salen > minimal)
		memcpy(sa.sun_path+1, advertisedName.data(), added);
}

}
}
