[event_routing] added new sync wait_for_event() function

This is exeactly the same as the async function, but does inline blocking/waiting until the event is received.
The function is very useful in script routes that does not have async support

Author: bogdan-iancu

modules/event_routing/doc/event_routing_admin.xml

@@ -176,6 +176,39 @@ route[reg_done] {
 		</example>
 	</section>
 
+	<section id="func_wait_for_event" xreflabel="wait_for_event()">
+		<title>
+		<function moreinfo="none">wait_for_event(event,filter,timeout)</function>
+		</title>
+		<para>
+		Exactly as the async <xref linkend="afunc_wait_for_event"/> function,
+		but sync/blocking version. The script execution will block and wait
+		until the event is delivered or the timeout hits 
+		</para>
+		<para>
+		The function return 1 upon success (an event was received), -1 in error
+		case or -2 in timeout case (no event was received).
+		</para>
+		<para>
+		This function can be used from any type of route.
+		</para>
+		<example>
+		<title><function>wait_for_event</function> usage</title>
+		<programlisting format="linespecific">
+...
+# block until the callee to register
+$avp(filter) = "aor="+$rU+"@"+$rd
+wait_for_event("E_UL_AOR_INSERT",$avp(filter), 40);
+if ($rc>0) {
+	xlog("user $avp(aor) is now registered\n");
+	lookup("location");
+	t_relay();
+}
+</programlisting>
+		</example>
+	</section>
+
+
 	</section>
 
 

modules/event_routing/ebr_data.c

@@ -323,6 +323,7 @@ int add_ebr_subscription( struct sip_msg *msg, ebr_event *ev,
 		sub->tm.hash = 0;
 		sub->tm.label = 0;
 	}
+
 	LM_DBG("transaction reference is %X:%X\n",sub->tm.hash,sub->tm.label);
 
 	/* link subscription to the event */
@@ -332,7 +333,7 @@ int add_ebr_subscription( struct sip_msg *msg, ebr_event *ev,
 	lock_release( &(ev->lock) );
 
 	LM_DBG("new subscription [%s] on event %.*s/%d successfully added from "
-		"process %d\n", (flags&EBR_SUBS_TYPE_WAIT)?"WAIT":"NOTIFY",
+		"process %d\n", EBR_SUBS_TYPE(sub),
 		ev->event_name.len, ev->event_name.s, ev->event_id, process_no);
 
 	return 0;
@@ -481,8 +482,7 @@ int notify_ebr_subscriptions( ebr_event *ev, evi_params_t *params)
 		/* 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",
-				(sub->flags&EBR_SUBS_TYPE_WAIT)?"WAIT":"NOTIFY",
+				"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 );
@@ -505,6 +505,20 @@ int notify_ebr_subscriptions( ebr_event *ev, evi_params_t *params)
 					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, 
+				 * we will free here only the subcription (without the
+				 * data field) */
+				sub->data = NULL; /*just to avod earlier free*/
+				/* setting swait_data->ret_avps to -1 serves as an 
+				 * indication of a timeout */
+				swait_data->ret_avps = ((void*)(long)-1);
 			}
 
 			/* unlink it */
@@ -541,8 +555,7 @@ int notify_ebr_subscriptions( ebr_event *ev, evi_params_t *params)
 		if (matches) {
 
 			LM_DBG("subscription type [%s]from process %d(pid %d) matched "
-				"event, generating notification via IPC\n",
-				(sub->flags&EBR_SUBS_TYPE_WAIT)?"WAIT":"NOTIFY",
+				"event, generating notification via IPC\n", EBR_SUBS_TYPE(sub),
 				sub->proc_no, pt[sub->proc_no].pid);
 
 			/* convert the EVI params into AVP (only once) */
@@ -582,14 +595,34 @@ int notify_ebr_subscriptions( ebr_event *ev, evi_params_t *params)
 					if (job->data) shm_free(job->data);
 					shm_free(job);
 				}
-			} else {
+			} else
+			if (sub->flags&EBR_SUBS_TYPE_WAIT) {
 				/* 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);
 				}
 
+				/* unlink it */
+				if (sub_prev) sub_prev->next = sub_next;
+				else ev->subs = sub_next;
+				/* free it */
+				free_ebr_subscription(sub);
+				/* do not count us as prev, as we are removed */
+				sub = sub_prev;
+			} else
+			/* resume if an sync/blocking WAIT */
+			if (sub->flags&EBR_SUBS_TYPE_SWAIT) {
+				struct swait_pack *swait_data = (struct swait_pack*)sub->data;
+				swait_data->ret_avps = job->avps;
+				shm_free(job); /* we need only the AVPs in this scenario */
+				cond_lock(&swait_data->cond);
+				cond_signal(&swait_data->cond);
+				cond_unlock(&swait_data->cond);
+				/* and destroy the subscription as it is only one time
+				 * triggering */
+				sub->data = NULL; /* just to be sure it is not freed here */
 				/* unlink it */
 				if (sub_prev) sub_prev->next = sub_next;
 				else ev->subs = sub_next;
@@ -651,8 +684,7 @@ void ebr_timeout(unsigned int ticks, void* param)
 				continue;
 
 			LM_DBG("subscription type [%s] from process %d(pid %d) on "
-				"event <%.*s> expired at %d, now %d\n",
-				(sub->flags&EBR_SUBS_TYPE_WAIT)?"WAIT":"NOTIFY",
+				"event <%.*s> expired at %d, now %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, my_time );

modules/event_routing/ebr_data.h

@@ -24,6 +24,7 @@
 #define _MODULE_EBR_H
 
 #include "../../locking.h"
+#include "../../lib/cond.h"
 #include "../tm/t_lookup.h"
 
 #define EVI_ROUTING_NAME "routing"
@@ -42,8 +43,14 @@ struct _ebr_event;
 
 #define EBR_SUBS_TYPE_WAIT  (1<<0)
 #define EBR_SUBS_TYPE_NOTY  (1<<1)
-#define EBR_DATA_TYPE_ROUT  (1<<2)
-#define EBR_DATA_TYPE_FUNC  (1<<3)
+#define EBR_SUBS_TYPE_SWAIT (1<<2)
+#define EBR_DATA_TYPE_ROUT  (1<<3)
+#define EBR_DATA_TYPE_FUNC  (1<<4)
+
+#define EBR_SUBS_TYPE(_s) \
+	( ((_s)->flags&EBR_SUBS_TYPE_WAIT)?"WAIT": \
+		(((_s)->flags&EBR_SUBS_TYPE_SWAIT)?"SWAIT":"NOTIFY") )
+
 
 typedef struct usr_avp *(*ebr_pack_params_cb) (evi_params_t *params);
 
@@ -55,8 +62,8 @@ typedef struct _ebr_subscription {
 	ebr_pack_params_cb pack_params;
 	void *data;
 	int expire;
-	/* Transaction ID data */
-	struct tm_id tm ;
+	/* Transaction ID data , used for NOTIFY */
+	struct tm_id tm;
 	struct _ebr_subscription *next;
 } ebr_subscription;
 
@@ -70,6 +77,13 @@ typedef struct _ebr_event {
 	struct _ebr_event *next;
 } ebr_event;
 
+
+struct swait_pack {
+	gen_cond_t cond;
+	struct usr_avp *ret_avps;
+};
+
+
 ebr_event * search_ebr_event( const str *name );
 
 ebr_event * add_ebr_event( const str *name );

modules/event_routing/event_routing.c

@@ -22,6 +22,7 @@
 #include "../../ut.h"
 #include "../../ipc.h"
 #include "../../mod_fix.h"
+#include "../../lib/cond.h"
 #include "../../evi/evi_transport.h"
 #include "../../evi/evi_modules.h"
 #include "../tm/tm_load.h"
@@ -40,6 +41,8 @@ static int mod_init(void);
 static int cfg_validate(void);
 static int notify_on_event(struct sip_msg *msg, ebr_event* event,
 		pv_spec_t *avp_filter, void *route, int *timeout);
+static int wait_for_event_sync(struct sip_msg* msg,
+		ebr_event* event, pv_spec_t* avp_filter, int* timeout);
 static int wait_for_event(struct sip_msg* msg, async_ctx *ctx,
 		ebr_event* event, pv_spec_t* avp_filter, int* timeout);
 
@@ -83,6 +86,11 @@ static const cmd_export_t cmds[]={
 		{CMD_PARAM_STR, fix_notification_route, free_notification_route},
 		{CMD_PARAM_INT, 0 ,0}, {0,0,0}},
 		EVENT_ROUTE|REQUEST_ROUTE|ONREPLY_ROUTE|FAILURE_ROUTE|BRANCH_ROUTE},
+	{"wait_for_event",  (cmd_function)wait_for_event_sync, {
+		{CMD_PARAM_STR, fix_event_name, 0},
+		{CMD_PARAM_VAR, fixup_check_avp, 0},
+		{CMD_PARAM_INT, 0 ,0}, {0,0,0}},
+		ALL_ROUTES},
 	{"ebr_bind", (cmd_function)ebr_bind, {{0,0,0}}, 0},
 	{0,0,{{0,0,0}},0}
 };
@@ -431,6 +439,75 @@ int api_wait_for_event(struct sip_msg *msg, async_ctx *ctx,
 	return _wait_for_event(msg, ctx, event, filters_cpy, timeout, pack_params);
 }
 
+
+static int wait_for_event_sync(struct sip_msg* msg,
+					ebr_event* event, pv_spec_t* avp_filter, int* timeout)
+{
+	struct swait_pack *swait_data;
+	ebr_filter *filters;
+	int rc = -1;
+
+	if (pack_ebr_filters(msg, avp_filter->pvp.pvn.u.isname.name.n,
+	                     &filters) < 0) {
+		LM_ERR("failed to build list of EBR filters\n");
+		goto done;
+	}
+
+	if (event->event_id == -1) {
+		/* do the init of the event*/
+		if (init_ebr_event(event) < 0) {
+			LM_ERR("failed to init event\n");
+			goto done;
+		}
+	}
+
+	swait_data = (struct swait_pack*)shm_malloc( sizeof(struct swait_pack) );
+	if (swait_data==NULL) {
+		LM_ERR("failed to allocated SWAIT data\n");
+		goto done;
+	}
+	if (cond_init(&swait_data->cond) != 0) {
+		LM_ERR("could not initialize subscription cond\n");
+		goto done;
+	}
+	swait_data->ret_avps = NULL; /*to be populated on resume*/
+
+	/* we have a valid EBR event here, let's subscribe on it */
+	if (add_ebr_subscription(msg, event, filters,
+	    *timeout, NULL, (void*)swait_data, EBR_SUBS_TYPE_SWAIT) < 0) {
+		LM_ERR("failed to add ebr subscription for event %d\n",
+		       event->event_id);
+		goto done1;
+	}
+
+	/* now just wait on the codition */
+	cond_lock(&swait_data->cond);
+	cond_wait(&swait_data->cond);
+	cond_unlock(&swait_data->cond);
+
+	/* timeout or event received ? */
+	if (swait_data->ret_avps==((void*)(long)-1) ) {
+		/* was a timeout */
+		rc = -2;
+	} else {
+		/* event data received */
+		if (swait_data->ret_avps!=NULL)
+			/* push the received event AVPs into the current list */
+			ebr_resume_from_wait( NULL, NULL, (void*)swait_data->ret_avps);
+		/* success */
+		rc = 1;
+	}
+
+	return rc == 0 ? 1 : rc;
+done1:
+	cond_destroy(&swait_data->cond);
+done:
+	shm_free(swait_data);
+	return rc;
+}
+
+
+
 /************ implementation of the EVI transport API *******************/
 
 static int ebr_match(evi_reply_sock *sock1, evi_reply_sock *sock2)