kazoo : event key/subkey & documentation corrections

modules/kazoo/README

@@ -111,7 +111,7 @@ Luis Azedo
    1.18. Set presentity_table parameter
    1.19. Set dialog_expires parameter
    1.20. Set presence_expires parameter
-   1.21. Set presence_expires parameter
+   1.21. Set mwi_expires parameter
    1.22. kazoo_publish usage
    1.23. kazoo_query usage
    1.24. kazoo_subscribe usage
@@ -242,6 +242,9 @@ Chapter 1. Admin Guide
    event_route[kazoo:consumer-event[-payload_key_value[-payload_subkey_val
    ue]]]
 
+   we can set the key/subkey pair on a subscription base. check the
+   payload on subscribe.
+
    Example 1.1. define the event route
 ...
 modparam("kazoo", "amqp_consumer_event_key", "Event-Category")
@@ -257,6 +260,14 @@ xlog("L_INFO", "received $(kzE{kz.json,Event-Package}) update for $(kzE{kz.json
 ...
 }
 
+event_route[kazoo:consumer-event-presence]
+{
+# presence is the value extracted from Event-Category field in json payload
+xlog("L_INFO", "received $(kzE{kz.json,Event-Package}) update for $(kzE{kz.json
+,From})");
+...
+}
+
 event_route[kazoo:consumer-event-event-category-event-name]
 {
 # event-category is the name of the amqp_consumer_event_key parameter
@@ -265,6 +276,13 @@ event_route[kazoo:consumer-event-event-category-event-name]
 ...
 }
 
+event_route[kazoo:consumer-event-event-category]
+{
+# event-category is the name of the amqp_consumer_event_key parameter
+# this event route is executed if we can't find the previous
+...
+}
+
 event_route[kazoo:consumer-event]
 {
 # this event route is executed if we can't find the previous
@@ -353,8 +371,8 @@ modparam("kazoo", "amqp_consumer_processes", 10)
 
 4.1.3. amqp_consumer_event_key(str)
 
-   The name of the field in json payload to compose the event name 1st
-   part
+   The default name of the field in json payload to compose the event name
+   1st part
 
    Default value is “Event-Category”.
 
@@ -365,8 +383,8 @@ modparam("kazoo", "amqp_consumer_event_key", "My-JSON-Field-Name")
 
 4.1.4. amqp_consumer_event_subkey(str)
 
-   The name of the field in json payload to compose the event name 2nd
-   part
+   The default name of the field in json payload to compose the event name
+   2nd part
 
    Default value is “Event-Name”.
 
@@ -418,10 +436,7 @@ while(true) // main  loop
 
 4.2.1. amqp_consumer_loop_count(int)
 
-   The database url.
-
-   If set, the module is a fully operational presence server. Otherwise,
-   it is used as a 'library', for its exported functions.
+   The consumer loop count.
 
    Default value is 10.
 
@@ -432,10 +447,7 @@ modparam("kazoo", "amqp_consumer_loop_count", 3)
 
 4.2.2. amqp_internal_loop_count(int)
 
-   The database url.
-
-   If set, the module is a fully operational presence server. Otherwise,
-   it is used as a 'library', for its exported functions.
+   The internal listen for commands loop count.
 
    Default value is 5.
 
@@ -446,10 +458,7 @@ modparam("kazoo", "amqp_internal_loop_count", 1)
 
 4.2.3. amqp_consumer_ack_loop_count(int)
 
-   The database url.
-
-   If set, the module is a fully operational presence server. Otherwise,
-   it is used as a 'library', for its exported functions.
+   The work ack loop count.
 
    Default value is 20.
 
@@ -593,7 +602,7 @@ modparam("kazoo", "presence_expires", 600)
 
    Default value is 3600.
 
-   Example 1.21. Set presence_expires parameter
+   Example 1.21. Set mwi_expires parameter
 ...
 modparam("kazoo", "mwi_expires", 600)
 ...
@@ -701,6 +710,8 @@ event_route[kazoo:consumer-event]
      * durable : int, default 0
      * no_ack : int, default 1
      * wait_for_consumer_ack : int, default 0
+     * event_key : str, no default
+     * event_subkey : str, no default
 
    This function must be called from event_route[kazoo:mod-init].
 

modules/kazoo/doc/kazoo_admin.xml

@@ -77,6 +77,9 @@ The worker process issues an event-route where we can act on the received payloa
     <para>
     Kazoo module will try to execute the event route from most significant to less significant. 
     define the event route like event_route[kazoo:consumer-event[-payload_key_value[-payload_subkey_value]]]
+    </para>
+    <para>
+    we can set the key/subkey pair on a subscription base. check the payload on subscribe.
     </para>
         <example>
         <title>define the event route</title>
@@ -94,6 +97,13 @@ xlog("L_INFO", "received $(kzE{kz.json,Event-Package}) update for $(kzE{kz.json,
 ...
 }
 
+event_route[kazoo:consumer-event-presence]
+{
+# presence is the value extracted from Event-Category field in json payload 
+xlog("L_INFO", "received $(kzE{kz.json,Event-Package}) update for $(kzE{kz.json,From})");
+...
+}
+
 event_route[kazoo:consumer-event-event-category-event-name]
 {
 # event-category is the name of the amqp_consumer_event_key parameter
@@ -102,6 +112,13 @@ event_route[kazoo:consumer-event-event-category-event-name]
 ...
 }
 
+event_route[kazoo:consumer-event-event-category]
+{
+# event-category is the name of the amqp_consumer_event_key parameter
+# this event route is executed if we can't find the previous 
+...
+}
+
 event_route[kazoo:consumer-event]
 {
 # this event route is executed if we can't find the previous 
@@ -213,7 +230,7 @@ modparam("kazoo", "amqp_consumer_processes", 10)
     <section>
         <title><varname>amqp_consumer_event_key</varname>(str)</title>
         <para>
-        The name of the field in json payload to compose the event name 1st part
+        The default name of the field in json payload to compose the event name 1st part
         </para>
         <para>
         <emphasis>Default value is <quote>Event-Category</quote>.</emphasis>
@@ -231,7 +248,7 @@ modparam("kazoo", "amqp_consumer_event_key", "My-JSON-Field-Name")
     <section>
         <title><varname>amqp_consumer_event_subkey</varname>(str)</title>
         <para>
-        The name of the field in json payload to compose the event name 2nd part
+        The default name of the field in json payload to compose the event name 2nd part
         </para>
         <para>
         <emphasis>Default value is <quote>Event-Name</quote>.</emphasis>
@@ -304,11 +321,7 @@ while(true) // main  loop
     <section>
         <title><varname>amqp_consumer_loop_count</varname>(int)</title>
         <para>
-        The database url.
-        </para>
-        <para>If set, the module is a fully operational
-        presence server. Otherwise, it is used as a 'library', for 
-        its exported functions.
+        The consumer loop count.
         </para>
         <para>
         <emphasis>Default value is 10.</emphasis>
@@ -326,11 +339,7 @@ modparam("kazoo", "amqp_consumer_loop_count", 3)
     <section>
         <title><varname>amqp_internal_loop_count</varname>(int)</title>
         <para>
-        The database url.
-        </para>
-        <para>If set, the module is a fully operational
-        presence server. Otherwise, it is used as a 'library', for 
-        its exported functions.
+        The internal listen for commands loop count.
         </para>
         <para>
         <emphasis>Default value is 5.</emphasis>
@@ -348,11 +357,7 @@ modparam("kazoo", "amqp_internal_loop_count", 1)
     <section>
         <title><varname>amqp_consumer_ack_loop_count</varname>(int)</title>
         <para>
-        The database url.
-        </para>
-        <para>If set, the module is a fully operational
-        presence server. Otherwise, it is used as a 'library', for 
-        its exported functions.
+        The work ack loop count.
         </para>
         <para>
         <emphasis>Default value is 20.</emphasis>
@@ -572,7 +577,7 @@ modparam("kazoo", "presence_expires", 600)
         <emphasis>Default value is 3600.</emphasis>
         </para>
         <example>
-        <title>Set <varname>presence_expires</varname> parameter</title>
+        <title>Set <varname>mwi_expires</varname> parameter</title>
         <programlisting format="linespecific">
 ...
 modparam("kazoo", "mwi_expires", 600)
@@ -710,6 +715,12 @@ event_route[kazoo:consumer-event]
             <listitem>
                <para>wait_for_consumer_ack : int, default 0</para> 
             </listitem>
+            <listitem>
+               <para>event_key : str, no default</para> 
+            </listitem>
+            <listitem>
+               <para>event_subkey : str, no default</para> 
+            </listitem>
         </itemizedlist>
         </para>
         <para>

modules/kazoo/kz_amqp.c

@@ -113,6 +113,20 @@ amqp_bytes_t kz_amqp_bytes_dup_from_str(str *src)
 	return kz_amqp_bytes_malloc_dup(amqp_cstring_bytes(src->s));
 }
 
+
+void kz_amqp_free_consumer_delivery(kz_amqp_consumer_delivery_ptr ptr)
+{
+	if(ptr == NULL)
+		return;
+	if(ptr->payload)
+		shm_free(ptr->payload);
+	if(ptr->event_key)
+		shm_free(ptr->event_key);
+	if(ptr->event_subkey)
+		shm_free(ptr->event_subkey);
+	shm_free(ptr);
+}
+
 void kz_amqp_free_bind(kz_amqp_bind_ptr bind)
 {
 	if(bind == NULL)
@@ -125,6 +139,10 @@ void kz_amqp_free_bind(kz_amqp_bind_ptr bind)
 		kz_amqp_bytes_free(bind->queue);
 	if(bind->routing_key.bytes)
 		kz_amqp_bytes_free(bind->routing_key);
+	if(bind->event_key.bytes)
+		kz_amqp_bytes_free(bind->event_key);
+	if(bind->event_subkey.bytes)
+		kz_amqp_bytes_free(bind->event_subkey);
 	shm_free(bind);
 }
 
@@ -180,7 +198,7 @@ kz_amqp_cmd_ptr kz_amqp_alloc_pipe_cmd()
 	return cmd;
 }
 
-kz_amqp_bind_ptr kz_amqp_bind_alloc(str* exchange, str* exchange_type, str* queue, str* routing_key )
+kz_amqp_bind_ptr kz_amqp_bind_alloc_ex(str* exchange, str* exchange_type, str* queue, str* routing_key, str* event_key, str* event_subkey )
 {
     kz_amqp_bind_ptr bind = NULL;
 
@@ -223,6 +241,22 @@ kz_amqp_bind_ptr kz_amqp_bind_alloc(str* exchange, str* exchange_type, str* queu
 	    }
 	}
 
+	if(event_key != NULL) {
+		bind->event_key = kz_amqp_bytes_dup_from_str(event_key);
+	    if (bind->event_key.bytes == NULL) {
+			LM_ERR("Out of memory allocating for routing key\n");
+			goto error;
+	    }
+	}
+
+	if(event_subkey != NULL) {
+		bind->event_subkey = kz_amqp_bytes_dup_from_str(event_subkey);
+	    if (bind->event_subkey.bytes == NULL) {
+			LM_ERR("Out of memory allocating for routing key\n");
+			goto error;
+	    }
+	}
+
 	return bind;
 
 error:
@@ -230,6 +264,11 @@ error:
     return NULL;
 }
 
+kz_amqp_bind_ptr kz_amqp_bind_alloc(str* exchange, str* exchange_type, str* queue, str* routing_key )
+{
+	return kz_amqp_bind_alloc_ex(exchange, exchange_type, queue, routing_key, NULL, NULL );
+}
+
 void kz_amqp_init_connection_pool() {
 	if(kz_pool == NULL) {
 		kz_pool = (kz_amqp_conn_pool_ptr) shm_malloc(sizeof(kz_amqp_conn_pool));
@@ -923,6 +962,8 @@ int kz_amqp_subscribe(struct sip_msg* msg, char* payload)
 	str queue_s;
 	str routing_key_s;
 	str payload_s;
+	str key_s;
+	str subkey_s;
 	int passive = 0;
 	int durable = 0;
 	int exclusive = 0;
@@ -950,6 +991,8 @@ int kz_amqp_subscribe(struct sip_msg* msg, char* payload)
     json_extract_field("type", exchange_type_s);
     json_extract_field("queue", queue_s);
     json_extract_field("routing", routing_key_s);
+    json_extract_field("event_key", key_s);
+    json_extract_field("event_subkey", subkey_s);
 
     tmpObj = json_object_object_get(json_obj, "passive");
     if(tmpObj != NULL) {
@@ -1397,7 +1440,7 @@ int kz_amqp_consumer_fire_event(char *eventkey)
 
 }
 
-void kz_amqp_consumer_event(int child_no, char *payload)
+void kz_amqp_consumer_event(int child_no, char *payload, char* event_key, char* event_subkey)
 {
     json_obj_ptr json_obj = NULL;
     str ev_name = {0, 0}, ev_category = {0, 0};
@@ -1414,20 +1457,33 @@ void kz_amqp_consumer_event(int child_no, char *payload)
 		return;
     }
 
-    json_extract_field(dbk_consumer_event_key.s, ev_category);
-    json_extract_field(dbk_consumer_event_subkey.s, ev_name);
+    char* key = (event_key == NULL ? dbk_consumer_event_key.s : event_key);
+    char* subkey = (event_subkey == NULL ? dbk_consumer_event_subkey.s : event_subkey);
+
+    json_extract_field(key, ev_category);
+    json_extract_field(subkey, ev_name);
 
     sprintf(buffer, "kazoo:consumer-event-%.*s-%.*s",ev_category.len, ev_category.s, ev_name.len, ev_name.s);
     for (p=buffer ; *p; ++p) *p = tolower(*p);
     for (p=buffer ; *p; ++p) if(*p == '_') *p = '-';
     if(kz_amqp_consumer_fire_event(buffer) != 0) {
-        sprintf(buffer, "kazoo:consumer-event-%.*s-%.*s",dbk_consumer_event_key.len, dbk_consumer_event_key.s, dbk_consumer_event_subkey.len, dbk_consumer_event_subkey.s);
+        sprintf(buffer, "kazoo:consumer-event-%.*s",ev_category.len, ev_category.s);
         for (p=buffer ; *p; ++p) *p = tolower(*p);
         for (p=buffer ; *p; ++p) if(*p == '_') *p = '-';
         if(kz_amqp_consumer_fire_event(buffer) != 0) {
-            sprintf(buffer, "kazoo:consumer-event");
+            sprintf(buffer, "kazoo:consumer-event-%s-%s", key, subkey);
+            for (p=buffer ; *p; ++p) *p = tolower(*p);
+            for (p=buffer ; *p; ++p) if(*p == '_') *p = '-';
             if(kz_amqp_consumer_fire_event(buffer) != 0) {
-                LM_ERR("kazoo:consumer-event not found");
+                sprintf(buffer, "kazoo:consumer-event-%s", key);
+                for (p=buffer ; *p; ++p) *p = tolower(*p);
+                for (p=buffer ; *p; ++p) if(*p == '_') *p = '-';
+				if(kz_amqp_consumer_fire_event(buffer) != 0) {
+					sprintf(buffer, "kazoo:consumer-event");
+					if(kz_amqp_consumer_fire_event(buffer) != 0) {
+						LM_ERR("kazoo:consumer-event not found");
+					}
+				}
             }
         }
     }
@@ -1462,7 +1518,7 @@ void kz_amqp_consumer_loop(int child_no)
 				kz_amqp_consumer_delivery_ptr ptr;
 				if(read(data_pipe, &ptr, sizeof(ptr)) == sizeof(ptr)) {
 					LM_DBG("consumer %d received payload %s\n", child_no, ptr->payload);
-					kz_amqp_consumer_event(child_no, ptr->payload);
+					kz_amqp_consumer_event(child_no, ptr->payload, ptr->event_key, ptr->event_subkey);
 					if(ptr->channel > 0 && ptr->delivery_tag > 0) {
 						kz_amqp_cmd_ptr cmd = kz_amqp_alloc_pipe_cmd();
 						cmd->type = KZ_AMQP_ACK;
@@ -1472,8 +1528,7 @@ void kz_amqp_consumer_loop(int child_no)
 							LM_ERR("failed to send ack to AMQP Manager in process %d, write to command pipe: %s\n", getpid(), strerror(errno));
 						}
 					}
-					shm_free(ptr->payload);
-					shm_free(ptr);
+					kz_amqp_free_consumer_delivery(ptr);
 				}
 			}
     	}
@@ -1494,8 +1549,9 @@ int check_timeout(struct timeval *now, struct timeval *start, struct timeval *ti
 
 int consumer = 1;
 
-void kz_amqp_send_consumer_event_ex(char* payload, amqp_channel_t channel, uint64_t delivery_tag, int nextConsumer)
+void kz_amqp_send_consumer_event_ex(char* payload, char* event_key, char* event_subkey, amqp_channel_t channel, uint64_t delivery_tag, int nextConsumer)
 {
+	int len = 0;
 	kz_amqp_consumer_delivery_ptr ptr = (kz_amqp_consumer_delivery_ptr) shm_malloc(sizeof(kz_amqp_consumer_delivery));
 	if(ptr == NULL) {
 		LM_ERR("NO MORE SHARED MEMORY!");
@@ -1505,7 +1561,8 @@ void kz_amqp_send_consumer_event_ex(char* payload, amqp_channel_t channel, uint6
 	ptr->channel = channel;
 	ptr->delivery_tag = delivery_tag;
 	ptr->payload = payload;
-
+	ptr->event_key = event_key;
+	ptr->event_subkey = event_subkey;
 	if (write(kz_pipe_fds[consumer*2+1], &ptr, sizeof(ptr)) != sizeof(ptr)) {
 		LM_ERR("failed to send payload to consumer %d : %s\nPayload %s\n", consumer, strerror(errno), payload);
 	}
@@ -1520,7 +1577,7 @@ void kz_amqp_send_consumer_event_ex(char* payload, amqp_channel_t channel, uint6
 
 void kz_amqp_send_consumer_event(char* payload, int nextConsumer)
 {
-	kz_amqp_send_consumer_event_ex(payload, 0, 0, nextConsumer);
+	kz_amqp_send_consumer_event_ex(payload, NULL, NULL, 0, 0, nextConsumer);
 }
 
 void kz_amqp_fire_connection_event(char *event, char* host)
@@ -1732,6 +1789,8 @@ void kz_amqp_manager_loop(int child_no)
 						break;
 					case KZ_AMQP_CONSUMING:
 						kz_amqp_send_consumer_event_ex(kz_amqp_bytes_dup(envelope.message.body),
+								kz_amqp_bytes_dup(channels[idx].consumer->event_key),
+								kz_amqp_bytes_dup(channels[idx].consumer->event_subkey),
 								channels[idx].consumer->no_ack ? 0 : envelope.channel,
 								channels[idx].consumer->no_ack ? 0 : envelope.delivery_tag,
 								(firstLoop && dbk_single_consumer_on_reconnect) ? 0 : 1);

modules/kazoo/kz_amqp.h

@@ -84,6 +84,8 @@ typedef struct {
 	char* payload;
 	uint64_t delivery_tag;
 	amqp_channel_t channel;
+	char* event_key;
+	char* event_subkey;
 } kz_amqp_consumer_delivery, *kz_amqp_consumer_delivery_ptr;
 
 typedef struct {
@@ -91,6 +93,8 @@ typedef struct {
 	amqp_bytes_t exchange_type;
 	amqp_bytes_t routing_key;
 	amqp_bytes_t queue;
+	amqp_bytes_t event_key;
+	amqp_bytes_t event_subkey;
 	amqp_boolean_t passive;
 	amqp_boolean_t durable;
 	amqp_boolean_t exclusive;