myvtm/Framework/spbase/SpAsyncWait.h

#pragma once

#include "SpBase.h"

#include "sp_rpc.h"
#include "sp_ses.h"
#include "sp_svc.h"
#include "sp_def.h"
#include "sp_env.h"

#include "spinlock.h"
#include "list.h"

class SpEntity;

class SpAsyncWait : public IAsynWaitSp
{
public:
	SpAsyncWait(SpEntity *m_pEntity);
	virtual ~SpAsyncWait();

	// IAsyncWaitSp
	virtual const char *GetFuctionName() { return ""; }
	virtual ErrorCodeEnum WaitAnswer(DWORD dwTimeout);
	virtual ErrorCodeEnum WaitAnswer(DWORD &dwUserError, CSimpleString& str, DWORD dwTimeout);
	virtual ErrorCodeEnum CancelWait();
	virtual bool IsPending();
	virtual DWORD GetMessageID() { return m_dwMessageID; }
	virtual ErrorCodeEnum AsyncGetAnswer(CAutoBuffer &ReceiveBuffer, bool &bEnd);
	virtual ErrorCodeEnum AsyncGetAnswer();
	virtual ErrorCodeEnum AsyncGetAnswer(CAutoBuffer &ReceiveBuffer, bool &bEnd, DWORD &dwUserError, CSimpleString& str);
	virtual ErrorCodeEnum AsyncGetAnswer(DWORD& dwUserError, CSimpleString& str);
	virtual ErrorCodeEnum AsyncGetAnswer(DWORD& dwUserError);

	virtual CSmartPointer<IReleasable> GetCallContext();
	virtual bool GetCallback(CSmartPointer<ICallbackListener> &pCallbackListener,CSmartPointer<IReleasable> &pContext);
	virtual void SetCallback(ICallbackListener *pCallback,IReleasable *pContext=NULL);
	virtual ErrorCodeEnum GetExpireTime(DWORD &dwWholeTime,DWORD &dwLeftTime) { return Error_NotImpl; }
	
	virtual ErrorCodeEnum Begin(const void *arg = NULL) { return Error_NotImpl; }
	virtual ErrorCodeEnum Close() { return Error_NotImpl; }
	virtual void OnAnswerWaited(ErrorCodeEnum Error) { }

	void SetMessageID(DWORD dwMessageID) { m_dwMessageID = dwMessageID; } 
	void ReceiveAnsPkt(int error, int end, iobuffer_t **ans_pkt, bool bReadUserCode = false);

	LONG IncrementRef() 
	{
		return InterlockedIncrement(m_pRefCount);
	}

	LONG DecrementRef() {
		LONG ret = InterlockedDecrement(m_pRefCount);
		if (ret == 0) {
			delete m_pRefCount;	// put delete here to avoid release it twice in case of CSmartPointer
			delete this;
		}
		return ret;
	}

	// return m_pRefCount as CSmartPointer ref count param, so it could be released by CSmartPointer
	// hence m_pRefCount should not be release in ~SpAsyncWait()
	LONG* GetRefCountPtr()
	{
		return m_pRefCount;
	}

protected:
	LONG *m_pRefCount;		// could be passed into CSmartPointer by GetRefCountPtr(), so release it carefully
	BOOL m_bEnd;
	BOOL m_bCancel;
	HANDLE m_sem;
	HANDLE m_evt_cancel;

	spinlock_t m_lock;
	CSmartPointer<IReleasable> m_pCallbackContext;
	ICallbackListener *m_pCallback;

	struct pending_pkt {
		struct list_head entry;
		iobuffer_t *pkt;
		DWORD dwSysError;
		DWORD dwUserError;
		CSimpleString str;
		int end;
		~pending_pkt() {
			if (pkt)
				iobuffer_dec_ref(pkt);
		}
	};
	struct list_head m_pending_pkt_list;
	pending_pkt *m_curr_pkt;
	int m_total_pkt;

	struct wait_callback_entry {
		struct list_head entry;
		int cancel;
		int cnt;
		wait_callback_entry() : cancel(0), cnt(0) {}
	};
	struct list_head m_pending_callback_list;
	strand_t *m_callback_strand;
	void threadpool_on_callback(struct wait_callback_entry *wce);
	static void __threadpool_on_callback(threadpool_t *threadpool, void *arg, param_size_t param1, param_size_t param2);

	DWORD m_dwMessageID;
	SpEntity *m_pEntity;

	friend ErrorCodeEnum WaitMultiAnswers(CAutoArray<IAsynWaitSp *> arrAsynWaits, DWORD &dwRetIndex, bool bWaitAll, DWORD dwTimeout);
};

class SpAsyncWaitRPC : public SpAsyncWait
{
public:
	SpAsyncWaitRPC(SpEntity *pEntity, iobuffer_t **req_pkt, int call_type, bool fetch_user = false);
	virtual ~SpAsyncWaitRPC();

	virtual ErrorCodeEnum Begin(const void *arg = NULL);
	virtual ErrorCodeEnum Close();

	static void __on_ans(sp_rpc_client_t *client, int error, iobuffer_t **ans_pkt, void *user_data);
	void on_ans(int error, iobuffer_t **ans_pkt);

private:
	int m_call_type;
	bool m_user;
	iobuffer_t *m_req_pkt;
	sp_rpc_client_t *m_rpc;
};

class SpAsyncWaitTsx : public SpAsyncWait
{
public:
	SpAsyncWaitTsx(SpEntity *pEntity, sp_ses_uac_t *uac, int timeout, int method_id, int method_sig, CAutoBuffer Buffer, int tsx_id);
	virtual ~SpAsyncWaitTsx();

	virtual ErrorCodeEnum Begin(const void *arg = NULL);
	virtual ErrorCodeEnum Close();
	virtual ErrorCodeEnum WaitAnswer(DWORD dwTimeout);
	virtual ErrorCodeEnum WaitAnswer(DWORD &dwUserError, CSimpleString& str, DWORD dwTimeout);

	static void __on_ans(sp_tsx_uac_t *tsx, int error, int end, iobuffer_t **ans_pkt, void *user_data);
	void on_ans(int error, int end, iobuffer_t **ans_pkt);
	void setLinkContext(linkContext& tmpContext);

private:
	iobuffer_t *m_req_pkt;
	int m_method_id;
	int m_method_sig;
	int m_timeout;
	int m_tsx_id;
	sp_tsx_uac_t *m_tsx;
};