#include "LuaChannel.h"
#include <thread>
#include <queue>
#include <map>
#include <string>
#include <mutex>
#include <condition_variable>
#include <chrono>
using namespace std;

class Channel
{
public:
	queue<string> bus;
	mutex mLock;
	condition_variable cond;
};

map<int, Channel> gChannels;
mutex gChannelLock;


void ChannelSend(Channel& c, const string& data)
{
	unique_lock<mutex> ulk(c.mLock);
	c.bus.push(data);
}

int ChannelRead(Channel& c, string& out_data, int timeout=-1)
{
	unique_lock<mutex> ulk(c.mLock);
	if (c.bus.empty())
	{
		if (timeout > 0)
		{
			if (cv_status::timeout == c.cond.wait_for(ulk, chrono::seconds(timeout)))
			{
				return 0;  // timed out.
			}
			out_data = c.bus.front();
			c.bus.pop();
			return 1;
		}
		else if(timeout == 0)
		{
			return 0;
		}
		else
		{
			c.cond.wait(ulk);
			out_data = c.bus.front();
			c.bus.pop();
			return 1;
		}
	}
	else
	{
		out_data = c.bus.front();
		c.bus.pop();
		return 1;
	}
}

Channel& AcquireChannel(int channelID)
{
	unique_lock<mutex> ulk(gChannelLock);
	return gChannels[channelID];
}

int InitChannel(lua_State* L)
{
	return 0;
}

class VMInfo
{
public:
	unique_ptr<thread> ptd;
	unique_ptr<LuaVM> pvm;
	int wait_join;

	VMInfo() : wait_join(0), pvm(new LuaVM)
	{

	}

	~VMInfo()
	{
		if (ptd.get() && ptd.get()->joinable())
		{
			if (wait_join)
			{
				ptd.get()->join();
			}
			else
			{
				pvm.release();
				ptd.get()->detach();
			}
		}
	}
};

void VMRunner(VMInfo& vm, string code)
{

	if (!vm.wait_join)
	{
		vm.pvm.reset();
	}
}

int CreateVM(lua_State* L)
{
	VMInfo* vm = new (lua_newuserdata(L, sizeof(VMInfo))) VMInfo;
	string code(lua_tostring(L, 1));
	vm->ptd.reset(new thread(VMRunner, *vm, code));
	return 1;
}