rvmegacotermevent.c

h.248协议源码

			return rvFalse;
		}
		/* Stop signals for the first accepted digit */
		if(strlen(rvStringGetData(&x->dialString))==1 && !rvMegacoRequestedEventGetKeepActiveFlag(x->digitMapReqEvent)) 
			rvMegacoTermEventStopSignals(x);

		if(x->digitMapStat==RV_MEGACODIGITMAP_PARTIALMATCH || 
				x->digitMapStat==RV_MEGACODIGITMAP_FULLMATCH) {
			x->digitMapMedia = media; /* Save for later processing */
			rvMegacoTermDigitMapTimerRestart(x,timerDuration*1000);
			/* If event is explicitly required,send a notify */
			processRequestedEvent(x,name,media,args);
			return rvFalse;
		}
		else if(x->digitMapStat==RV_MEGACODIGITMAP_UNAMBIGUOUSMATCH) {
			/* If is explicitly required,process and accumulate */
			processAccumulateEvent(x,name,media,args);
			generateDMEventCompletion(x,media);
			/* Notify */
			sendNotify(x);
			processEmbFields(x,x->digitMapReqEvent);
		}
	}

	return rvTrue;
}

/* Function to be called if digitmap timer expires 
   If status was RV_MEGACODIGITMAP_PARTIALMATCH, send notify with dd/Meth=PM.
   If status was RV_MEGACODIGITMAP_FULLMATCH send notify with dd/Meth=FM
*/
void rvMegacoTermDigitMapTimerProcess(RvMegacoTerm * x) {
	generateDMEventCompletion(x,x->digitMapMedia);
	sendNotify(x);
	processEmbFields(x,x->digitMapReqEvent);
}

static void  queueBlockedEvent(RvMegacoTerm * x,RvMegacoPackageItem* name,
							   RvMdmMediaStream * media,const RvMegacoParameterList * args);


void rvMegacoTermPostEvent(RvTimer* t,void* data) {
	RvMegacoTerm* x = (RvMegacoTerm*)data;
	RvMegacoTermMgr* mgr = rvMegacoTermGetTermMgr(x);

	/* If the termination is processing a command, the events will
	   be processed later */
	if(!x->blockedEventState ) {
	   if(!rvQueueFull(&mgr->eventQueue) )
	       rvQueuePush(&mgr->eventQueue,x);
	    else
		   /* If the queue is full, delay the event so the
		      function doesn't block */
		   rvTimerReset(&x->blockedEventTimer,RV_MEGACOTERM_EVENTTIME);
	}
}


/* All events are queued to avoid deadlock if the event is process
   in the context of the user application.*/
RvBool rvMegacoTermQueueObsEvent(RvMdmTerm* mdmTerm,const char* pkg,const char* id,
						  RvMdmMediaStream* media,RvMegacoParameterList * args) {
	RvMegacoTerm * term = (RvMegacoTerm *)rvMdmTermGetXTerm_(mdmTerm);
	RvMegacoPackageItem item;
	RvBool retval = rvTrue;

	if(term->inactive)
		return retval;

	rvMegacoPackageItemConstructA(&item,pkg,id,term->alloc);

	rvMutexLock(&term->blockedEventMutex);

	queueBlockedEvent(term,&item,media,args);

	rvMegacoTermPostEvent(NULL,term);

	rvMutexUnlock(&term->blockedEventMutex);

	/* Destruct local objects */
	rvMegacoPackageItemDestruct(&item);
	return retval;
}

/* Process the events comming from timer manager or dedicated thread */
RvBool rvMegacoTermProcessObsEvent(RvMdmTerm* mdmTerm,const char* pkg,const char* id,
						  RvMdmMediaStream* media,RvMegacoParameterList * args) {
	RvMegacoTerm * term = (RvMegacoTerm *)rvMdmTermGetXTerm_(mdmTerm);
	RvMegacoPackageItem item;
	RvBool retval = rvFalse;

	if(term->inactive)
		return rvTrue;

	rvMegacoPackageItemConstructA(&item,pkg,id,term->alloc);

	rvMutexLock(&term->mutex);
	if(term->bufferActive==rvTrue) {
		addEventToBuffer(term,&item,media,args);
	}
	else {
		if(term->digitMapActive && isInDigitMapPkg(term,&item) )
			retval = processDigitMapEvent(term,&item,media,args);
		else 
			retval = processNotifyEvent(term,&item,media,args);
	}
	rvMutexUnlock(&term->mutex);

	/* Destruct local objects */
	rvMegacoPackageItemDestruct(&item);
	return retval;
}


/*---------------------------------------------------------------------------------*/
/* Functions used to process new events descriptor                                 */
/*---------------------------------------------------------------------------------*/
static const RvMdmEventInfo* validateRequestedEvent(RvMegacoTerm * x,
									 const RvMegacoRequestedEvent* reqEvent,
									 RvMegacoCommandReply * commReply) {
	const RvMdmEventInfo * info = NULL;
	RvMdmError error;
	const RvMegacoPackageItem* name;
	RvMdmTermMgr * mdmMgr = x->context->termMgr->mdmTermMgr;

	rvMdmErrorConstruct_(&error);

	name = rvMegacoRequestedEventGetName(reqEvent);
	if( rvMdmTermIsPkgSupported_(rvMegacoTermGetMdmTerm(x),rvMegacoPackageItemGetPackage(name)) ) {
		info = rvMdmTermMgrGetEventInfo_(mdmMgr,&name->package,&name->item,&error);
	}
	/* Event not supported - report error and halt processing */
	if(info==NULL) {
		rvMegacoTermSendErrorMsg(x,&error,RV_MEGACOERROR_NOTSPT_EVENT,
								"Termination unequipped to detect requested Event",commReply);

	}

	return info;
}

/* The application is responsible for processing events */
static RvBool procesEventParameters(RvMegacoTerm * x,const RvMegacoRequestedEvent* reqEvent,
									const RvMdmEventInfo * info,RvMegacoCommandReply * commReply) {
	RvMdmMediaStream * mdmStream = NULL;
	const char * pkg,* item;
	RvMegacoStreamId streamId;
	RvMdmError error;
	rvMdmErrorConstruct_(&error);

	if(info->paramF!=NULL) {
		pkg =  rvMegacoPackageItemGetPackage(rvMegacoRequestedEventGetName(reqEvent));
		item = rvMegacoPackageItemGetItem(rvMegacoRequestedEventGetName(reqEvent));

		/* Get the stream id if any */
		streamId = rvMegacoRequestedEventGetStreamId(reqEvent);			
		if(streamId) 
			mdmStream = rvMegacoTermGetMediaStream(x,streamId);

		if( !info->paramF(rvMegacoTermGetMdmTerm(x),pkg,item,mdmStream,
						  rvMegacoRequestedEventGetParameterList(reqEvent),&error) ) {
			/* Generate error msg  */
			rvMegacoTermSendErrorMsg(x,&error,RV_MEGACOERROR_NOTSPT_EVENT,
									"Media Gateway unequipped to detect requested Event",commReply);
			return rvFalse;
		}	
	}

	return rvTrue;
}

/* Enable specific detection of events if required */
static void enableEventDetection(RvMegacoTerm * x,const RvMegacoRequestedEvent* reqEvent,
								 const RvMdmEventInfo * info) {
	const char * pkg,* item;

	if(info->enableF!=NULL) {
		pkg =  rvMegacoPackageItemGetPackage(rvMegacoRequestedEventGetName(reqEvent));
		item = rvMegacoPackageItemGetItem(rvMegacoRequestedEventGetName(reqEvent));
		info->enableF(rvMegacoTermGetMdmTerm(x),pkg,item);
	}

	return;
}


/* Take one event from the event buffer and process it */
/* Note: are digitMap events accumulated during buffer processing? */
static RvBool processEventFromBuffer(RvMegacoTerm* x) {
	RvBool wasProcessed = rvFalse;

	if(!x->bufferActive)
		return rvTrue;
	/*
	1. If the EventBuffer is empty, the MG waits for detection of events
       based on the new EventsDescriptor
	2. If the event in the queue is in the events listed in the new
       EventsDescriptor, the MG acts on the event and removes the
       event from the EventBuffer.  The time stamp of the Notify shall
       be the time the event was actually detected.  The MG then waits
       for a new EventsDescriptor

	3. If the event is not in the new EventsDescriptor, the MG SHALL
	   discard the event and repeat from step 1.
*/
	x->bufferActive = rvFalse; /* Allow processing events (1) */

	while(!rvListEmpty(&x->eventBuffer) && !wasProcessed) {
		const char* id,* pkg;
		RvMdmMediaStream * media;
		unsigned int streamId;
		RvMegacoObservedEvent* obsEvent;

		/* Retrieve an event from the buffer */
		obsEvent = rvListBack(&x->eventBuffer);
		/* Get the event data */
		streamId = rvMegacoObservedEventGetStreamId(obsEvent);
		media =	rvMegacoTermGetMediaStream(x,streamId);
		pkg = rvMegacoPackageItemGetPackage(rvMegacoObservedEventGetName(obsEvent));
		id  = rvMegacoPackageItemGetItem(rvMegacoObservedEventGetName(obsEvent));
		x->bufferedTime = rvMegacoObservedEventGetTimeStamp(obsEvent);
		/* Process the event */
		wasProcessed = rvMegacoTermProcessObsEvent(rvMegacoTermGetMdmTerm(x),id,pkg,media,NULL);
		rvListPopBack(RvMegacoObservedEvent)(&x->eventBuffer);
	}

	/* Reset the buffered time to NULL */
	x->bufferedTime = NULL;

	if(wasProcessed) /*At least one event was notified,restore the state to buffer accumulate */
		x->bufferActive = rvFalse;

	return rvTrue;
}

/* Called from event process thread. Process one blocked event */
void rvMegacoTermProcessBlockedEvent(RvMegacoTerm* x) {

	rvMutexLock(&x->blockedEventMutex);
	/* If the termination is processing a command, leave the events in the
	   buffer so the queue doesn't block. Events will be processed after
	   the command ends */
	if(x->blockedEventState) {
		rvMutexUnlock(&x->blockedEventMutex);
		return;
	}

	/* if events arrived before mdm started, start mdm and stop the timer */
	rvMegacoTermMgrCheckStartMdm(rvMegacoTermGetTermMgr(x));

	rvMutexLock(&x->mutex);
	/* Process all pending events */
	while(!rvListEmpty(&x->blockedEventBuffer) ) {
		const char* id,* pkg;
		RvMdmMediaStream * media;
		unsigned int streamId;
		RvMegacoObservedEvent* obsEvent;
		RvMegacoParameterList* args;

		/* Retrieve a blocked event from the buffer */
		obsEvent = rvListFront(&x->blockedEventBuffer);
		/* Get the event data */
		streamId = rvMegacoObservedEventGetStreamId(obsEvent);
		media =	rvMegacoTermGetMediaStream(x,streamId);
		pkg = rvMegacoPackageItemGetPackage(rvMegacoObservedEventGetName(obsEvent));
		id  = rvMegacoPackageItemGetItem(rvMegacoObservedEventGetName(obsEvent));
		args = (RvMegacoParameterList*)rvMegacoObservedEventGetParameterList(obsEvent);
		if( rvMegacoParameterListIsEmpty(args) )
			args = NULL;
		/* Process the event */
		rvMegacoTermProcessObsEvent(rvMegacoTermGetMdmTerm(x),pkg,id,media,args);
		rvListPopFront(RvMegacoObservedEvent)(&x->blockedEventBuffer);
	}
	rvMutexUnlock(&x->mutex);
	rvMutexUnlock(&x->blockedEventMutex);
}

static void  queueBlockedEvent(RvMegacoTerm * x,RvMegacoPackageItem* name,
							       RvMdmMediaStream * media,const RvMegacoParameterList * args) {

	/* Construct observed event and add to linked list */
	RvMegacoObservedEvent * obsEvent = rvListAllocBack(RvMegacoObservedEvent)(&x->blockedEventBuffer);
	/* Don't bother to get the time because the event can be descarded 
	   and the delay should be small */
	x->bufferedTime = &nullTime;
	buildObservedEvent(x,obsEvent,name,media,args);

	x->bufferedTime = NULL;
}

/* For future use */
/* no msgs arrived in a while, send service change for this termination (we can assume we're
   not in a middle of a call at this point */
/*void processKeepAliveTimeout(RvTimer* timer, void* data) {
	RvMegacoTermTimer* megacoTimer = (RvMegacoTermTimer*)data;
	RvMegacoTerm* term = (RvMegacoTerm*)megacoTimer->data;
	RvMegacoTermMgr* mgr = rvMegacoTermGetTermMgr(term);
	rvMutexLock(&term->mutex);*/
	
	/* if no msgs arrived by now, send service change */
/*	if (term->msgCounter != mgr->msgCounter) {
		RvMegacoServiceChangeDescriptor sc;
		rvMdmServiceChangeConstruct(&sc);
		rvMdmServiceChangeSetMethod(&sc, RV_MEGACOSERVICECHANGEMETHOD_RESTART);
		rvMdmServiceChangeSetReason(&sc, RV_MEGACOSERVICECHANGEREASON_SERVICERESTORED);
		rvMegacoTermMgrSendTermServiceChangeUp(mgr, term, &sc);
		rvMdmServiceChangeDestruct(&sc);
	}
	rvMutexUnlock(&term->mutex);
	
}*/


/* Processing of the requested events descriptor 
   Note: DigitMap descriptor should be processed before this one 
0. Stop digitmap timer
1. Copy the requested event descriptor, initialize new observed events
2. Validate the requested events. If there are not supported,send an error reply.
3. For any auditable event, call the audit function and verify that the state is
   acceptable. If not, send error reply
4. For any event with digitmap parameters, find the pkg info,install new digitmap,
   start digitmap timer
5. If bufferActive is set:
   Take one event from buffer and process
*/
void rvMegacoTermProcessEvents(RvMegacoTerm * x,
								const RvMegacoEventsDescriptor * eventsDescr,
								RvMegacoCommandReply * commReply) {
	size_t i;
	const RvMegacoRequestedEvent* reqEvent;
	const RvMdmEventInfo* info;
/*	int numEvents;*/

	/* recovery mechanism: if events list is empty, start a timer. if no msgs arrived 
	   by the time it expires, send service change */
/*	if ((numEvents = rvMegacoEventsDescriptorGetNumEvents(eventsDescr)) == 0) {
		RvMegacoTermMgr* mgr = rvMegacoTermGetTermMgr(x);*/
		/* store global counter and start the timer */
/*		x->msgCounter = mgr->msgCounter;
		x->keepAliveTimer = rvMegacoTermTimerCreate(x->alloc , 60000, 
										  processKeepAliveTimeout, x, &x->mutex);
	}*/
	/* events list is not empty, stop timer */
/*	else if (x->keepAliveTimer) {
		if (x->keepAliveTimer->valid)
			rvMegacoTermTimerCancel(x->keepAliveTimer);
	}*/
		
	if(!rvMegacoEventsDescriptorIsSet(eventsDescr))
		return;

	/*0. Stop digitmap timer */
	rvMegacoTermDeactivateDigitMap(x);

	/* 1. Copy the requested event descriptor, initialize new observed events */
	rvMegacoEventsDescriptorCopy(&x->eventsDescr,eventsDescr);

	/*   Disable the explicit detection of events in the application */
	rvMdmTermClearDetectedEvents_(rvMegacoTermGetMdmTerm(x));

	rvMegacoObservedEventsDescriptorClear(&x->obsEventsDescr);
	/*Note: Set request id,should have special function or do construct/destruct*/
	x->obsEventsDescr.requestId = rvMegacoEventsDescriptorGetRequestId(&x->eventsDescr);

	for(i=0;i<rvMegacoEventsDescriptorGetNumEvents(&x->eventsDescr);i++) {
		reqEvent = rvMegacoEventsDescriptorGetEvent(&x->eventsDescr,i);
		/* 2. Validate the requested events. If they are not supported,send an error reply.*/
		/* Note: should the embbeded event get validated as well? */
		if((info = validateRequestedEvent(x,reqEvent,commReply))==NULL) { 
			return;
		}		

		/* 3. Let the application process the parameters of the requested event */
		if(!procesEventParameters(x,reqEvent,info,commReply) )
			return;

		/* Enable detection of events if required */
		enableEventDetection(x,reqEvent,info);

		/* 4. For any event with digitmap parameters, find the pkg info,install new digitmap,*/
		/*    start digitmap timer */
		if(!rvMegacoTermProcessDigitMapParameter(x,reqEvent,commReply))
			return;
	}
	/* 5. If bufferActive is set:
		  Take one event from buffer and process */
	processEventFromBuffer(x);
}

void rvMegacoTermProcessEventBuffer(RvMegacoTerm * x,const RvMegacoEventBufferDescriptor * eventBufferDescr,RvMegacoCommandReply * commReply) {
	rvMegacoEventBufferDescriptorCopy(&x->eventBufferDescr,eventBufferDescr);
}

void rvMegacoTermProcessEventQueue(RvThread* t,void * data) {
	RvMegacoTermMgr* mgr = (RvMegacoTermMgr*)data;
	RvMegacoTerm* term = NULL;

	while (rvQueuePop(&mgr->eventQueue, (RvQueueType *)&term) == 0) {
		/* Get a termination from the queue */
		if(term==NULL) 
			return; 
		rvMegacoTermProcessBlockedEvent(term);	
	}
}