scqueue.cxx

Windows CE 6.0 Server 源码

//
// Copyright (c) Microsoft Corporation.  All rights reserved.
//
//
// Use of this source code is subject to the terms of the Microsoft shared
// source or premium shared source license agreement under which you licensed
// this source code. If you did not accept the terms of the license agreement,
// you are not authorized to use this source code. For the terms of the license,
// please see the license agreement between you and Microsoft or, if applicable,
// see the SOURCE.RTF on your install media or the root of your tools installation.
// THE SOURCE CODE IS PROVIDED "AS IS", WITH NO WARRANTIES.
//
/*++


Module Name:

    scqueue.cxx

Abstract:

    Small client queue class support


--*/
#include <sc.hxx>

#include <scpacket.hxx>
#include <scqman.hxx>
#include <scqueue.hxx>
#include <scsman.hxx>
#include <sccomp.hxx>

#include <mq.h>

void ScQueue::Init (void) {
	fInitialized	= FALSE;

	sFile			= NULL;
	pPackets		= NULL;

	pJournal		= NULL;

	lpszQueueHost	= NULL;
	lpszQueueName	= NULL;
	lpszFormatName	= NULL;
	lpszQueueLabel	= NULL;

	memset (&qp, 0, sizeof(qp));

	tCreation		= scutil_now ();
	tModification	= tCreation;

	iQueueSizeB     = 0;

	fDenyAll		= FALSE;
	uiOpenRecv		= 0;
	uiOpen			= 0;

	llSeqID         = 0;
	uiSeqNum        = 0;
	uiFirstUnackedSeqN = 0;

	hkReceived      = 0;

	hUpdateEvent	= CreateEvent (NULL, FALSE, FALSE, NULL);

	pSess    = NULL;
}


static BOOL CheckQueueTypeSupported(int qType, BOOL fLocalOnly) {
	SVSUTIL_ASSERT(gMem->IsLocked());

	if ((qType == SCFILE_QP_FORMAT_HTTPS) || (qType == SCFILE_QP_FORMAT_HTTP))
		return gMachine->fUseSRMP;
	else if ((qType == SCFILE_QP_FORMAT_TCP) || (qType == SCFILE_QP_FORMAT_OS)) {
		if (gMachine->fUseBinary)
			return TRUE;

		return fLocalOnly;  // Allow creation of local queues even if binary is disabled.
	}
	else {
		SVSUTIL_ASSERT(0);
	}

	return FALSE;	
}

void ScQueue::InitSequence (void) {
	if (qp.bIsIncoming || (! qp.bTransactional)) {
		SVSUTIL_ASSERT ((uiSeqNum == 0) && (llSeqID == 0));
		return;
	}

	if (llSeqID)
		return;

	llSeqID = ((LONGLONG)scutil_now ()) << 32;

	SVSUTIL_ASSERT (llSeqID != 0);
	SVSUTIL_ASSERT (uiSeqNum == 0);
}
//
//	Create new incoming or outgoing queue from backup file
//
ScQueue::ScQueue (WCHAR *a_lpszFileName, HRESULT *phr) {
	Init ();

	pPackets = SVSNewTree(gMem->pTreeNodeMem);

	sFile = new ScFile (a_lpszFileName, this);

	if ((NULL==pPackets) || (NULL==sFile)) {
		if (phr)
			*phr = MQ_ERROR;

		return;
	}

	if (! sFile->Restore ()) {
		if (phr)
			*phr = MQ_ERROR;

		return;
	}

	WCHAR *lpFormatName = scutil_MakeFormatName (lpszQueueHost, lpszQueueName, &qp);

	if (! lpFormatName) {
		if (phr)
			*phr = MQ_ERROR_ILLEGAL_QUEUE_PATHNAME;

		return;
	}

	lpszFormatName = svsutil_StringHashAlloc (gMem->pStringHash, lpFormatName);
	g_funcFree (lpFormatName, g_pvFreeData);

	if (NULL == lpszFormatName) {
		if (phr)
			*phr = MQ_ERROR_ILLEGAL_QUEUE_PATHNAME;

		return;
	}
	
	if (! qp.bIsIncoming)
		pSess = gSessionMan->GetSession (lpszQueueHost,qp.bFormatType);

	InitSequence ();

	fInitialized = TRUE;
}


ScFile * ScQueue::CreateScFileFromName(WCHAR *lpszQueueHost, WCHAR *lpszQueueName, ScQueueParms *a_pqp, HRESULT *phr) {
	ScFile *sFile;

	WCHAR *lpszFileName  = scutil_MakeFileName (lpszQueueHost, lpszQueueName, a_pqp);
	if (lpszFileName==NULL) {
		if (phr)
			*phr = MQ_ERROR;
		return NULL;
	}
	
	sFile = new ScFile (lpszFileName, this);
	g_funcFree(lpszFileName, g_pvFreeData);

	if (sFile==NULL) {
		if (phr)
			*phr = MQ_ERROR;
		return NULL;
	}

	return sFile;
}

//
// Create new incoming queue based on path name
//
ScQueue::ScQueue (WCHAR *a_lpszPathName, WCHAR *a_lpszLabel, ScQueueParms *a_pqp, HRESULT *phr){
	SVSUTIL_ASSERT (a_pqp->bIsIncoming);

	Init ();

	WCHAR *l_lpszQueueHost, *l_lpszQueueName;
	if (! scutil_ParsePathName (a_lpszPathName, l_lpszQueueHost, l_lpszQueueName, a_pqp)) {
		if (phr)
			*phr = MQ_ERROR_ILLEGAL_QUEUE_PATHNAME;

		return;
	}

	if (! a_pqp->bIsIncoming) {
		g_funcFree (l_lpszQueueName, g_pvFreeData);
		g_funcFree (l_lpszQueueHost, g_pvFreeData);

		if (phr)
			*phr = MQ_ERROR_ILLEGAL_QUEUE_PATHNAME;

		return;
	}

	if (! CheckQueueTypeSupported(a_pqp->bFormatType,a_pqp->bIsIncoming)) {
		g_funcFree (l_lpszQueueName, g_pvFreeData);
		g_funcFree (l_lpszQueueHost, g_pvFreeData);

		if (phr)
			*phr = MQ_ERROR_QUEUE_NOT_AVAILABLE;

		return;
	}

	SVSUTIL_ASSERT ((! a_pqp->bIsJournal) || a_pqp->bIsIncoming);
	SVSUTIL_ASSERT (! (a_pqp->bIsJournal && a_pqp->bHasJournal));
	SVSUTIL_ASSERT ((! a_pqp->bIsJournalOn) || a_pqp->bHasJournal);

	lpszQueueName = svsutil_StringHashAlloc (gMem->pStringHash, l_lpszQueueName);
	g_funcFree (l_lpszQueueName, g_pvFreeData);

	lpszQueueHost = svsutil_StringHashAlloc (gMem->pStringHash, l_lpszQueueHost);
	g_funcFree (l_lpszQueueHost, g_pvFreeData);

	if ((NULL==lpszQueueName) || (NULL==lpszQueueHost)) {
		if (phr)
			*phr = MQ_ERROR;
		return;
	}

	WCHAR *lpFormatName = scutil_MakeFormatName (a_lpszPathName, a_pqp);

	if (! lpFormatName) {
		if (phr)
			*phr = MQ_ERROR_ILLEGAL_QUEUE_PATHNAME;

		return;
	}

	lpszFormatName = svsutil_StringHashAlloc (gMem->pStringHash, lpFormatName);
	g_funcFree (lpFormatName, g_pvFreeData);

	if (NULL==lpszFormatName) {
		if (phr)
			*phr = MQ_ERROR_ILLEGAL_QUEUE_PATHNAME;

		return;
	}

	qp = *a_pqp;

	scutil_uuidgen (&qp.guid);

	pPackets = SVSNewTree(gMem->pTreeNodeMem);

	if (NULL == pPackets) {
		if (phr)
			*phr = MQ_ERROR;
		return;
	}

	if (a_lpszLabel) {
		lpszQueueLabel = svsutil_wcsdup(a_lpszLabel);
		if (NULL == lpszQueueLabel) {
			if (phr)
				*phr = MQ_ERROR;
			return;
		}
	}

	tCreation = tModification = scutil_now();
	sFile = CreateScFileFromName(lpszQueueHost, lpszQueueName, a_pqp, phr);
	if (!sFile)
		return;

	if (! sFile->CreateNew()) {
		if (phr) {
			if (gQueueMan->FindIncomingByFormat (lpszFormatName) || gQueueMan->FindOutgoingByFormat (lpszFormatName))
				*phr = MQ_ERROR_QUEUE_EXISTS;
			else
				*phr = MQ_ERROR_INSUFFICIENT_RESOURCES;
		}

		return;
	}

	InitSequence ();

	fInitialized = TRUE;
}

//
//	Create non-local queue from format name
//
ScQueue::ScQueue (WCHAR *a_lpszFormatName, ScQueueParms *a_pqp, HRESULT *phr) {
	SVSUTIL_ASSERT (! a_pqp->bHasJournal);
	SVSUTIL_ASSERT (! a_pqp->bIsJournalOn);
	SVSUTIL_ASSERT (! a_pqp->bIsDeadLetter);
	SVSUTIL_ASSERT (! a_pqp->bIsMachineJournal);
	SVSUTIL_ASSERT (! a_pqp->bIsIncoming);
	SVSUTIL_ASSERT (! a_pqp->bIsJournal);

	Init ();

	qp = *a_pqp;
	scutil_uuidgen (&qp.guid);

	WCHAR *l_lpszQueueHost, *l_lpszQueueName;
	if (! scutil_ParseNonLocalDirectFormatName (a_lpszFormatName, l_lpszQueueHost, l_lpszQueueName, &qp)) {
		if (phr)
			*phr = MQ_ERROR_ILLEGAL_QUEUE_PATHNAME;

		return;
	}

	if (wcsicmp (l_lpszQueueHost, gMachine->lpszHostName) == 0) {
		g_funcFree (l_lpszQueueName, g_pvFreeData);
		g_funcFree (l_lpszQueueHost, g_pvFreeData);

		if (phr)
			*phr = MQ_ERROR_ILLEGAL_QUEUE_PATHNAME;

		return;
	}

	if (! CheckQueueTypeSupported(qp.bFormatType,a_pqp->bIsIncoming)) {
		g_funcFree (l_lpszQueueName, g_pvFreeData);
		g_funcFree (l_lpszQueueHost, g_pvFreeData);

		if (phr)
			*phr = MQ_ERROR_QUEUE_NOT_AVAILABLE;

		return;
	}

	lpszQueueName = svsutil_StringHashAlloc (gMem->pStringHash, l_lpszQueueName);
	g_funcFree (l_lpszQueueName, g_pvFreeData);

	lpszQueueHost = svsutil_StringHashAlloc (gMem->pStringHash, l_lpszQueueHost);
	g_funcFree (l_lpszQueueHost, g_pvFreeData);

	if ((NULL==lpszQueueName) || (NULL==lpszQueueHost)) {
		if (phr)
			*phr = MQ_ERROR;
		return;
	}

	//
	//	Override the quota setting...
	//
	qp.uiQuotaK = gMachine->uiDefaultOutQuotaK;

	sFile = CreateScFileFromName(lpszQueueHost, lpszQueueName, &qp, phr);
	if (!sFile)
		return;
	lpszFormatName = svsutil_StringHashAlloc (gMem->pStringHash, a_lpszFormatName);

	if (NULL==lpszFormatName) {
		if (phr)
			*phr = MQ_ERROR_ILLEGAL_QUEUE_PATHNAME;

		return;
	}

	pPackets = SVSNewTree(gMem->pTreeNodeMem);
	if (NULL == pPackets) {
		if (phr)
			*phr = MQ_ERROR;
		return;
	}

	tCreation = tModification = scutil_now();

	if (! sFile->CreateNew()) {
		if (phr) {
			if (gQueueMan->FindIncomingByFormat (lpszFormatName) || gQueueMan->FindOutgoingByFormat (lpszFormatName))
				*phr = MQ_ERROR_QUEUE_EXISTS;
			else
				*phr = MQ_ERROR_INSUFFICIENT_RESOURCES;
		}

		return;
	}

	SVSUTIL_ASSERT (! qp.bIsIncoming);

	pSess = gSessionMan->GetSession (lpszQueueHost,qp.bFormatType);

	InitSequence ();

	fInitialized = TRUE;
}

//
//	Delete queue...
//
static void freePacketData (void *pvData, void *pvArg) {
	ScPacket *pPacket = (ScPacket *)pvData;

	if (pPacket->pImage)
		g_funcFree (pPacket->pImage, g_pvFreeData);
}

ScQueue::~ScQueue (void) {