/** @file polledproxy.cpp
 *  MessageProxy compatible polled proxy for EA::Types::Functions
 *	@headerfile polledproxy.h "EABaseTypesMessage/polledproxy.h"
 *	@author Chris Burns (cburns@ea.com)
 *
 *	Core Technology UI
 *	(c) 2013 Electronic Arts Inc. 
 */

#include "EABaseTypesMessage/polledproxy.h"

#include "EABaseTypes/functioncallwrappers.h"

#include "EABaseTypesMessage/primitiveencoder.h"
#include "EABaseTypesMessage/messageproxy.h"
 
 namespace EA
 {
	namespace Types
	{
		typedef EASTLICA::HashMap<int, FunctionRef> AnonymousFunctionMap;

		class Session : public RefCountedICA<RefCounted>
		{
		public:

			Factory*				mFactory;
			FunctionRef				mCallback;

			int						mLocalSeq;
			int						mRemoteSeq;
			ArrayRef				mQueuedData;

			int						mAnonymousFunctionId;
			AnonymousFunctionMap	mAnonymousFunctions;

			Session(Factory* factory):
				RefCountedICA<RefCounted>(factory->GetAllocator()),
				mFactory(factory),
				mCallback(),
				mLocalSeq(0),
				mRemoteSeq(0),
				mQueuedData(factory->Array()),
				mAnonymousFunctionId(0),
				mAnonymousFunctions(factory->GetAllocator()) {}

			BaseTypeReturn EncodeFunction(FunctionIn function)
			{
				mAnonymousFunctions[mAnonymousFunctionId] = function;
			
				return mFactory->Number(mAnonymousFunctionId++);
			}

			BaseTypeReturn EncodeUserData(UserDataIn userdata)
			{
				EA_FAIL_MESSAGE("Currently userdata is not supported!");
				return NULL;
			}

			FunctionReturn DecodeFunction(BaseTypeIn data);

			UserDataReturn DecodeUserData(BaseTypeIn data)
			{
				EA_FAIL_MESSAGE("Currently userdata is not supported!");
				return NULL;
			}

			void Send(ObjectIn object)
			{
				++mLocalSeq;
				mQueuedData->push_back(object);
			
				if(mCallback != NULL)
				{
					ObjectRef o = mFactory->Object();

					o->insert("seq", mLocalSeq);
					o->insert("ack", mRemoteSeq);
					o->insert("data", mQueuedData);

					mCallback->Call<void>(o);
					mCallback = NULL;
				}
			}
		};

		class PolledProxyFunction : public Function
		{
		public:
		
			PolledProxyFunction(SessionIn session, int id):
				Function(session->mFactory),
				mSession(session),
				mId(id)
			{
#if !defined(EA_FINAL)
				SetDebugInfo(session->mFactory->String("Polled proxy function"));
#endif
			}
			
			virtual ~PolledProxyFunction()
			{
				ObjectRef object = mFactory->Object();
				
				object->insert("type", "AnonymousProxyRelease");
				object->insert("id", mId);
				
				mSession->Send(object);
			}
			
			virtual BaseTypeReturn Call(ArrayIn params)
			{
				ObjectRef object = mFactory->Object();
				
				object->insert("type", "AnonymousProxyCall");
				object->insert("id", mId);
				
				ArrayRef data = object->insertArray("data");
				ArrayRef metadata = object->insertArray("metadata");
				
				PrimitiveEncoder<Session> encoder(mSession.GetUnsafePointer(), data.GetUnsafePointer(), metadata.GetUnsafePointer());
				BaseTypeArrayDecoder<PrimitiveEncoder<Session> >::Decode(&encoder, params.GetUnsafePointer());
				
				mSession->Send(object);

				return NULL;
			}

			virtual BaseTypeReturn Clone()
			{
				AddRef();

				return this;
			}

			virtual bool Equals(FunctionIn pCompare)
			{
				return false; // nobody should ever compare these anyway
			}

			virtual EA::Type::TypeID GetFunctionType()
			{
				return EA::Type::TypeOf(*this);
			}

		private:

			SessionRef		mSession;
			int				mId;
		};

		FunctionReturn Session::DecodeFunction(BaseTypeIn data)
		{
			return new (mFactory) PolledProxyFunction(SessionIn(this), data->AsInt());
		}
	
		PolledProxy::PolledProxy(Factory* factory,
								 IProxyIn proxy)
		:	RefCountedICA<RefCounted>(factory->GetAllocator()),
			mFactory(factory),
			mProxy(proxy),
			mSessions(factory->GetAllocator())
		{
		}

		PolledProxy::~PolledProxy()
		{
		}

		ObjectReturn PolledProxy::ToObject()
		{
			ObjectRef that = mFactory->Object();

			that->insert("Poll", new (mFactory) Functor2<void, FunctionIn, ObjectIn>(mFactory, this, &PolledProxy::Poll));

			return that;
		}
		
		void PolledProxy::Poll(FunctionIn callback, ObjectIn object)
		{
			const EA::String& sessionName = (*object)["session"]->AsEAString();

			SessionRef session = NULL;

			SessionMap::iterator it = mSessions.find(sessionName);
			if(it == mSessions.end())
			{
				EA::Allocator::ICoreAllocator* allocator = mFactory->GetAllocator();
				session = CORE_NEW(allocator, "PolledProxy::Session", EA::Allocator::MEM_TEMP) Session(mFactory);
				mSessions[sessionName] = session;
			}
			else
			{
				session = (*it).second;
			}

			if(session->mCallback != NULL)
			{
				session->mCallback->Call<void>();
				session->mCallback = NULL;
			}

			int ack = (*object)["ack"]->AsInt();

			size_t numToPop = session->mQueuedData->size() - static_cast<size_t>(session->mLocalSeq - ack);
			
			if(numToPop > 0)
			{
				ArrayRef queuedData = mFactory->Array();
				for(size_t i = numToPop; i < session->mQueuedData->size(); ++i)
				{
					queuedData->push_back((*session->mQueuedData)[i]);
				}

				session->mQueuedData = queuedData;
			}

			ArrayRef data = (*object)["data"]->AsArray();

			int seq = (*object)["seq"]->AsInt();

			int numToReceive = 0;
			
			if(seq > session->mRemoteSeq)
			{
				numToReceive = seq - session->mRemoteSeq;

				session->mRemoteSeq = seq;
			}
			
			if(session->mQueuedData->size() > 0)
			{
				ObjectRef o = mFactory->Object();

				o->insert("seq", session->mLocalSeq);
				o->insert("ack", session->mRemoteSeq);
				o->insert("data", session->mQueuedData);

				callback->Call<void>(o);
			}
			else
			{
				session->mCallback = callback;
			}

			int length = static_cast<int>(data->size());

			int i = length - numToReceive;

			// if i is negative, this is junk data from a previous session - ignore it
			// we might be able to do interesting error handling here...?
			// perhaps tell the client that the session is closed to encourage it to make a brand new one?
			if(i < 0)
			{
				mSessions.erase(sessionName);
				return;
			}

			for(; i < length; ++i)
			{
				ObjectRef o = (*data)[i]->AsObject();

				const EA::String& type = (*o)["type"]->AsEAString();

				if(type == "Close")
				{
					mSessions.erase(sessionName);
					return;
				} 
				else if(type == "AnonymousProxyRelease")
				{
					session->mAnonymousFunctions.erase((*o)["id"]->AsInt());
				}
				else
				{
					ArrayRef params = mFactory->Array();
				
					BaseTypeArrayEncoder encoder(params.GetUnsafePointer());
					PrimitiveDecoderParams<Session> decoderParams(session.GetUnsafePointer(), (*o)["data"]->AsArray().GetUnsafePointer(), (*o)["metadata"]->AsArray().GetUnsafePointer());
					PrimitiveDecoder<Session, BaseTypeArrayEncoder>::Decode(&encoder, &decoderParams);
				
					FunctionRef function = NULL;
				
					if(type == "NamedProxyCall")
					{
						function = mProxy->GetFunction((*o)["path"]->AsCStr());
					}
					else if(type == "AnonymousProxyCall")
					{
						function = session->mAnonymousFunctions[(*o)["id"]->AsInt()];
					}
					else
					{
						EA_FAIL_FORMATTED(("Unrecognized type %s", type.c_str()));
					}
				
					function->Call(params);
				}
			}
		}
	}
}
 