myvtm/Framework/spbase/SpClientSessionFunction.cpp

#include "stdafx.h"

#include "SpBase.h"
#include "SpMisc.h"
#include "SpAsyncWait.h"
#include "SpEntity.h"
#include "SpClientSessionFunction.h"
#include "dbgutil.h"

#define DEFAULT_CONNECT_TIMEOUT			10000		//10s
#define WILD_POINTER_JUDGE			0x123456789ABCDEF0

SpClientSessionFunction::SpClientSessionFunction( SpEntity *pEntity, CClientSessionBase *pClientSessionBase, const char *pszRemoteEntity ) 
	: SpAsyncWait(pEntity), m_local_seq(0), m_pEntity(pEntity)
	, m_pClientSessionBase(pClientSessionBase), m_uac(NULL), m_value(WILD_POINTER_JUDGE)
{

	int rc;
	sp_ses_uac_callback cb;
	int remote_entity_id = SP_INVALID_SVC_ID;
	int remote_epid = SP_INVALID_MOD_ID;
	sp_env_t *env = sp_get_env();
	sp_entity_t *ent = sp_mod_mgr_find_entity_by_name(env->mod_mgr, pszRemoteEntity);
	if (ent && ent->service_flag && (ent->state == EntityState_Idle || ent->state == EntityState_Busy)) {
		remote_entity_id = ent->cfg->idx;
		remote_epid = ent->mod->cfg->idx;
		
	}
	if (remote_entity_id != SP_INVALID_SVC_ID) {
		cb.user_data = this;
		cb.on_connect = &__on_connect;
		cb.on_close = &__on_close;
		cb.on_destroy = &__on_destroy;
		rc = sp_ses_uac_create(pEntity->get_ses_mgr(), remote_epid, remote_entity_id, &cb, &m_uac);
		//DbgWithLink(LOG_LEVEL_INFO, LOG_TYPE_SYSTEM).setAPI(__FUNCTION__)("uac create, remote_epid:%d", remote_epid);
	}
}

SpClientSessionFunction::~SpClientSessionFunction()
{
	m_value = 0;
}

ErrorCodeEnum SpClientSessionFunction::Begin(const char *pszParam)
{
	int rc;

	if (m_uac) {
		iobuffer_t *conn_pkt = iobuffer_create(-1, -1);
		iobuffer_write(conn_pkt, IOBUF_T_STR, pszParam, -1);
		IncrementRef();
		rc = sp_ses_uac_async_connect(m_uac, DEFAULT_CONNECT_TIMEOUT+1, &conn_pkt);
		if (rc != 0) {
			DecrementRef();
		}
		if (conn_pkt)
			iobuffer_dec_ref(conn_pkt);
	} else {
		rc = Error_NetBroken;
	}

	return SpTranslateError(rc);
}

ErrorCodeEnum SpClientSessionFunction::Close()
{
	int rc;

	if (m_uac) {
		rc = sp_ses_uac_close(m_uac);
		if (rc == 0) {
			sp_ses_uac_destroy(m_uac);
		}
	} else {
		rc = Error_NotInit;
	}

	return SpTranslateError(rc);
}

void SpClientSessionFunction::OnAnswerWaited(ErrorCodeEnum Error)
{
	if (Error == Error_Succeed) {
		if (AsyncGetAnswer() == Error_Succeed) {
			m_pClientSessionBase->OnConnectSucceed();
		}
	}
}

ErrorCodeEnum SpClientSessionFunction::OnewayCall(DWORD dwMessageID, DWORD dwMessageSignature, CAutoBuffer Buffer)
{
	ErrorCodeEnum Error;

	if (m_uac) {
		int rc;
		iobuffer_t *info_pkt = iobuffer_create(-1, Buffer.GetCount()+32);
		int v = Buffer.GetCount();
		if (v > 0)
			iobuffer_write(info_pkt, IOBUF_T_BUF, &Buffer[0], v);
		rc = sp_ses_uac_send_info(m_uac, (int)dwMessageID, (int)dwMessageSignature, &info_pkt);
		if (info_pkt)
			iobuffer_dec_ref(info_pkt);
		Error = SpTranslateError(rc);
	} else {
		Error = Error_NotInit;
	}

	return Error;
}

ErrorCodeEnum SpClientSessionFunction::OnewayCall(DWORD dwMessageID, DWORD dwMessageSignature)
{
	ErrorCodeEnum Error;

	if (m_uac) {
		int rc;
		iobuffer_t *info_pkt = iobuffer_create(-1, -1);
		rc = sp_ses_uac_send_info(m_uac, (int)dwMessageID, (int)dwMessageSignature, &info_pkt);
		if (info_pkt)
			iobuffer_dec_ref(info_pkt);
		Error = SpTranslateError(rc);
	} else {
		Error = Error_NotInit;
	}

	return Error;
}

ErrorCodeEnum SpClientSessionFunction::OnewayCall(DWORD dwMessageID, DWORD dwMessageSignature, CAutoBuffer Buffer, linkContext& pContext)
{
	ErrorCodeEnum Error;

	if (m_uac) {
		int rc;
		iobuffer_t* info_pkt = iobuffer_create(-1, Buffer.GetCount() + 32);
		int v = Buffer.GetCount();
		if (v > 0)
			iobuffer_write(info_pkt, IOBUF_T_BUF, &Buffer[0], v);
		if (pContext.checkEmpty())
			pContext.AutoGenerate();
		iobuffer_set_linkInfo(info_pkt, pContext.bussinessId.GetData(), pContext.traceId.GetData(), pContext.spanId.GetData(), pContext.parentSpanId.GetData());
		rc = sp_ses_uac_send_info(m_uac, (int)dwMessageID, (int)dwMessageSignature, &info_pkt);
		if (info_pkt)
			iobuffer_dec_ref(info_pkt);
		Error = SpTranslateError(rc);

	}
	else {
		Error = Error_NotInit;
	}

	return Error;
}
ErrorCodeEnum SpClientSessionFunction::OnewayCall(DWORD dwMessageID, DWORD dwMessageSignature, linkContext& pContext)
{
	ErrorCodeEnum Error;

	if (m_uac) {
		int rc;
		iobuffer_t* info_pkt = iobuffer_create(-1, -1);
		if (pContext.checkEmpty())
			pContext.AutoGenerate();
		iobuffer_set_linkInfo(info_pkt, pContext.bussinessId.GetData(), pContext.traceId.GetData(), pContext.spanId.GetData(), pContext.parentSpanId.GetData());
		rc = sp_ses_uac_send_info(m_uac, (int)dwMessageID, (int)dwMessageSignature, &info_pkt);
		if (info_pkt)
			iobuffer_dec_ref(info_pkt);
		Error = SpTranslateError(rc);
	}
	else {
		Error = Error_NotInit;
	}


	return Error;
}

SessionStateEnum SpClientSessionFunction::GetCurrentState()
{
	SessionStateEnum State = SessionState_NotInit;

	if (m_uac) {
		int state = sp_ses_uac_get_state(m_uac);
		switch (state) {
			case SP_SES_STATE_INIT:
			case SP_SES_STATE_CONNECTING:
				break;
			case SP_SES_STATE_CONNECTED:
				State = SessionState_Live;
				break;
			case SP_SES_STATE_TERM:
			case SP_SES_STATE_ERROR:
				State = SessionState_Close;
				break;
			default:
				TOOLKIT_ASSERT(0);
				State = SessionState_NotInit;
				break;
		}
	} else {
		State = SessionState_NotInit;
	}

	return State;
}

ErrorCodeEnum SpClientSessionFunction::CloseSession()
{
	return Close();
}

ErrorCodeEnum SpClientSessionFunction::AsyncRequest(DWORD dwMessageID, DWORD dwMessageSignature, CAutoBuffer Buffer, CSmartPointer<IAsynWaitSp>& pAsynWaitSp, linkContext& pContext, DWORD dwTimeOut, DWORD dwReqId) {
	ErrorCodeEnum Error;

	if (!m_uac)
		return Error_NotInit;

	if (dwReqId == -1) {
		dwReqId = InterlockedIncrement((LONG*)&m_local_seq);
	}
	if (pContext.checkEmpty())
		pContext.AutoGenerate();
	SpAsyncWaitTsx* pAsyncWait = new SpAsyncWaitTsx(m_pEntity, m_uac, dwTimeOut, (int)dwMessageID, (int)dwMessageSignature, Buffer, (int)dwReqId);
	pAsyncWait->setLinkContext(pContext);

	Error = pAsyncWait->Begin();

	if (Error == Error_Succeed)
	{
		pAsynWaitSp.Attach(pAsyncWait, pAsyncWait->GetRefCountPtr());
		pAsyncWait->SetMessageID(dwMessageID);
	}

	pAsyncWait->DecrementRef();		// xkm@20150115
	return Error;
}
ErrorCodeEnum SpClientSessionFunction::AsyncRequest(DWORD dwMessageID, DWORD dwMessageSignature, CSmartPointer<IAsynWaitSp>& pAsynWaitSp, linkContext& pContext, DWORD dwTimeOut, DWORD dwReqId)
{
	CAutoBuffer Buffer;
	return AsyncRequest(dwMessageID, dwMessageSignature, Buffer, pAsynWaitSp, pContext, dwTimeOut, dwReqId);
}

ErrorCodeEnum SpClientSessionFunction::AsyncRequest(DWORD dwMessageID, DWORD dwMessageSignature, CAutoBuffer Buffer, CSmartPointer<IAsynWaitSp> &pAsynWaitSp, DWORD dwTimeOut, DWORD dwReqId)
{
	ErrorCodeEnum Error;

	if (!m_uac)
		return Error_NotInit;

	if (dwReqId == -1) {
		dwReqId = InterlockedIncrement((LONG*)&m_local_seq);
	}

	SpAsyncWaitTsx *pAsyncWait = new SpAsyncWaitTsx(m_pEntity, m_uac, dwTimeOut, (int)dwMessageID, (int)dwMessageSignature, Buffer, (int)dwReqId);

	Error = pAsyncWait->Begin();

	if (Error == Error_Succeed) 
	{
		pAsynWaitSp.Attach(pAsyncWait, pAsyncWait->GetRefCountPtr());
		pAsyncWait->SetMessageID(dwMessageID);
	}

	pAsyncWait->DecrementRef();		// xkm@20150115
	return Error;
}

ErrorCodeEnum SpClientSessionFunction::AsyncRequest(DWORD dwMessageID, DWORD dwMessageSignature, CSmartPointer<IAsynWaitSp> &pAsynWaitSp, DWORD dwTimeOut, DWORD dwReqId)
{
	CAutoBuffer Buffer;
	return AsyncRequest(dwMessageID, dwMessageSignature, Buffer, pAsynWaitSp, dwTimeOut, dwReqId);
}

void SpClientSessionFunction::__on_connect(sp_ses_uac_t *uac, int error, void *user_data)
{
	SpClientSessionFunction *pThis = static_cast<SpClientSessionFunction*>(user_data);
	if (pThis->isVaildPointer())
		pThis->on_connect(error);
}

void SpClientSessionFunction::__on_close(sp_ses_uac_t *uac, int error, void *user_data)
{
	SpClientSessionFunction *pThis = static_cast<SpClientSessionFunction*>(user_data);
	if (pThis->isVaildPointer())
		pThis->on_close(error);
}

void SpClientSessionFunction::__on_destroy(sp_ses_uac_t *uac, void *user_data)
{
	SpClientSessionFunction *pThis = static_cast<SpClientSessionFunction*>(user_data);
	if (pThis->isVaildPointer())
		pThis->on_destroy();
}

void SpClientSessionFunction::on_connect(int error)
{
	ReceiveAnsPkt(error, 1, NULL);
	DecrementRef();
}

void SpClientSessionFunction::on_close(int error)
{
	m_pClientSessionBase->OnClose(SpTranslateError(error));
}

void SpClientSessionFunction::on_destroy()
{
	if (m_pClientSessionBase) {
		m_pClientSessionBase->OnDestroy();
	}
}

bool SpClientSessionFunction::isVaildPointer()
{
	return m_value == WILD_POINTER_JUDGE;
}

const char * SpClientSessionFunction::GetRemoteEntityName()
{
	if (m_uac) {
		int remote_ent_idx = sp_ses_uac_get_remote_svc_id(m_uac);
		sp_entity_t *ent = sp_mod_mgr_find_entity_by_idx(sp_get_env()->mod_mgr, remote_ent_idx);
		if (ent) {
			return ent->cfg->name;
		}
	}
	return NULL;
}