mqueue: add support for fetching bulk items

Author: razvancrainea

modules/mqueue/doc/mqueue_admin.xml

@@ -277,12 +277,16 @@ opensips-cli -x mq_get_size xyz
 	</section>
 	<section id="mi_mq_fetch" xreflabel="mq_fetch">
 		<title>mq_fetch</title>
-		<para>Fetch a key-value pair from a memory queue.</para>
+		<para>Fetch one (or up to limit) key-value pair from a memory queue.</para>
 		<para>Parameters:</para>
 		<itemizedlist>
 			<listitem>
 				<emphasis>name</emphasis> - the name of memory queue
 			</listitem>
+			<listitem>
+				<emphasis>limit</emphasis> (optional) - if used, an array
+				with up to <emphasis>limit</emphasis> records are being returned.
+			</listitem>
 		</itemizedlist>
 		<example>
 		<title><function>mq_fetch</function> usage</title>

modules/mqueue/mqueue_api.c

@@ -225,6 +225,24 @@ mq_pv_t *mq_pv_get(str *name)
 	return NULL;
 }
 
+/**
+ *
+ */
+mq_item_t *mq_head_fetch_item(mq_head_t *mh)
+{
+	mq_item_t *ret;
+
+	if(mh->ifirst == NULL)
+		return NULL;
+
+	ret = mh->ifirst;
+	mh->ifirst = mh->ifirst->next;
+	if(mh->ifirst == NULL)
+		mh->ilast = NULL;
+	mh->csize--;
+	return ret;
+}
+
 /**
  *
  */
@@ -245,21 +263,10 @@ int mq_head_fetch(str *name)
 		return -1;
 	lock_get(&mh->lock);
 
-	if(mh->ifirst == NULL) {
-		/* empty queue */
-		lock_release(&mh->lock);
-		return -2;
-	}
-
-	mp->item = mh->ifirst;
-	mh->ifirst = mh->ifirst->next;
-	if(mh->ifirst == NULL) {
-		mh->ilast = NULL;
-	}
-	mh->csize--;
+	mp->item = mq_head_fetch_item(mh);
 
 	lock_release(&mh->lock);
-	return 0;
+	return (mp->item?0:-2);
 }
 
 /**

modules/mqueue/mqueue_api.h

@@ -73,6 +73,7 @@ int mq_head_defined(void);
 void mq_destroy(void);
 int mq_head_add(str *name, int msize, int addmode);
 int mq_head_fetch(str *name);
+mq_item_t *mq_head_fetch_item(mq_head_t *mh);
 void mq_pv_free(str *name);
 int mq_item_add(str *qname, str *key, str *val);
 mq_head_t *mq_head_get(str *name);

modules/mqueue/mqueue_mod.c

@@ -54,6 +54,8 @@ mi_response_t *mi_get_size(const mi_params_t *params,
 								struct mi_handler *async_hdl);
 mi_response_t *mi_fetch(const mi_params_t *params,
 								struct mi_handler *async_hdl);
+mi_response_t *mi_fetch_bulk(const mi_params_t *params,
+								struct mi_handler *async_hdl);
 
 static pv_export_t mod_pvs[] = {
 	{ {"mqk",     sizeof("mqk") - 1},     1090, pv_get_mqk,     0,
@@ -97,8 +99,8 @@ static const stat_export_t mod_stats[] = {
 };
 
 #define MQH1 "Params: none ; Get the size of all memory queues."
-#define MQH2 "Params: [mqueue] ; Get the size of a memory queue."
-#define MQH3 "Params: [mqueue] ; Fetch a key-value pair from a memory queue."
+#define MQH2 "Params: mqueue; Get the size of a memory queue."
+#define MQH3 "Params: mqueue [limit]; Fetch one (or a max limit of) key-value pair from a memory queue."
 
 static const mi_export_t mi_cmds[] = {
 	{"mq_get_sizes", MQH1, 0, 0, {
@@ -111,6 +113,7 @@ static const mi_export_t mi_cmds[] = {
 	},
 	{"mq_fetch",     MQH3, 0, 0, {
 		{mi_fetch,     {"name", 0}},
+		{mi_fetch_bulk,{"name", "limit", 0}},
 		{EMPTY_MI_RECIPE}}
 	},
 	{EMPTY_MI_EXPORT}
@@ -422,3 +425,47 @@ mi_response_t *mi_fetch(const mi_params_t *params,
 	return NULL;
 }
 
+mi_response_t *mi_fetch_bulk(const mi_params_t *params,
+								struct mi_handler *async_hdl)
+{
+	mi_response_t *resp;
+	mi_item_t *resp_obj, *mq_item;
+	str mqueue_name;
+	int limit;
+	mq_head_t *mh;
+	mq_item_t *item = NULL;
+
+	if (get_mi_string_param(params, "name", &mqueue_name.s, &mqueue_name.len) < 0)
+		return init_mi_param_error();
+
+	if (get_mi_int_param(params, "limit", &limit) < 0 || limit < 1)
+		return init_mi_param_error();
+
+	mh = mq_head_get(&mqueue_name);
+	if (!mh)
+		return init_mi_error(404, MI_SSTR("No such queue"));
+
+	resp = init_mi_result_array(&resp_obj);
+	if (!resp)
+		return NULL;
+
+	lock_get(&mh->lock);
+	do {
+		item = mq_head_fetch_item(mh);
+		if (!item)
+			break;
+		mq_item = add_mi_object(resp_obj, NULL, 0);
+		if (add_mi_string_fmt(mq_item, MI_SSTR("key"), item->key.s, item->key.len) < 0)
+			break;
+		if (add_mi_string_fmt(mq_item, MI_SSTR("value"), item->val.s, item->val.len) < 0)
+			break;
+		shm_free(item);
+		item = 0;
+	} while (--limit > 0);
+	lock_release(&mh->lock);
+
+	if (item)
+		shm_free(item);
+
+	return resp;
+}