Function notify_ebr_subscriptions

modules/event_routing/ebr_data.c:458–676

int notify_ebr_subscriptions( ebr_event *ev, evi_params_t *params)
{
	ebr_subscription *sub, *sub_next, *sub_prev;
	ebr_filter *filter;
	ebr_ipc_job *job;
	evi_param_t *e_param;
	int matches;
	struct usr_avp *avps=(void*)-1;
	unsigned int my_time;

	LM_DBG("notification received for event %.*s, checking subscriptions\n",
		ev->event_name.len, ev->event_name.s);

	my_time = get_ticks();

	lock_get( &(ev->lock) );

	ev->last_timeout_check = my_time;

	/* check the EBR subscription on this event and apply the filters */
	sub_prev = NULL;
	for ( sub=ev->subs ; sub ; sub_prev=sub, sub=sub_next) {
		sub_next = sub->next;

		/* discard expired subscriptions */
		if (sub->expire<my_time) {
			LM_DBG("subscription type [%s] from process %d(pid %d) on "
				"event <%.*s> expired at %d\n", EBR_SUBS_TYPE(sub),
				sub->proc_no, pt[sub->proc_no].pid,
				sub->event->event_name.len, sub->event->event_name.s,
				sub->expire );
			/* fire the job, if we deal with an WAIT */
			if (sub->flags&EBR_SUBS_TYPE_WAIT) {
				job =(ebr_ipc_job*)shm_malloc( sizeof(ebr_ipc_job) );
				if (job==NULL) {
					LM_ERR("failed to allocated new IPC job, skipping..\n");
					continue; /* with the next subscription */
				}
				job->ev = ev;
				job->data = sub->data;
				job->flags = sub->flags;
				job->tm = sub->tm;
				job->avps = NULL;
				/* sent the event notification via IPC to resume on the
				 * subscribing process */
				if (ipc_send_job( sub->proc_no, ebr_ipc_type , (void*)job)<0) {
					LM_ERR("failed to send job via IPC, skipping...\n");
					shm_free(job);
					continue; /* keep it and try next time */
				}
			} else
			/* resume if an sync/blocking WAIT */
			if (sub->flags&EBR_SUBS_TYPE_SWAIT) {
				struct swait_pack *swait_data = (struct swait_pack*)sub->data;
				cond_lock(&swait_data->cond);
				cond_signal(&swait_data->cond);
				cond_unlock(&swait_data->cond);
				/* the "swait_data" will be freed by the waiting proc,