Add support to send a command to an RTPEngine specified by the sock_var (#3617)

* Add support to send a command to an RTPEngine specified by the sock_var

* Removed some logging problems that the linter complained about.

* Updates to support memory corruption edge case

* Fixed casting typo

* Update some comments

* Implement developer suggestions

1. If the set variable is passed in, use it. Otherwise use the existing logic. 

2. Move setting of the socket STR outside of the loop.

* Update README

* Update rtpengine_admin.xml documentation

Author: NormB

modules/rtpengine/README

@@ -442,6 +442,15 @@ modparam("rtpengine", "ping_enabled", yes)
 
 1.5. Exported Functions
 
+    Many rtpengine_* functions accept a "sock_var" parameter that
+    will be populated with the socket of the RTPEngine chosen for
+    the particular operation. The format of the data stored in
+    "sock_var" is: "proto:ip:port".  If the "sock_var" has
+    been specified and it is non-NULL then it will be used to
+    determine the specific RTPEngine to use.  Note that the socket
+    specified by "sock_var" must be a member of the current RTPEngine
+    Set context.
+
 1.5.1.  rtpengine_use_set(setid)
 
    Sets the ID of the RTP proxy set to be used for the next

modules/rtpengine/doc/rtpengine_admin.xml

@@ -78,6 +78,16 @@
 		You can use the <xref linkend="param_extra_failover_error"/> parameter
 		to extend the above list.
 	</para>
+	<para>
+		Many rtpengine_* functions accept a "sock_var" parameter that
+		will be populated with the socket of the RTPEngine chosen for
+		the particular operation. The format of the data stored in
+		"sock_var" is: "proto:ip:port".  If the "sock_var" has
+		been specified and it is non-NULL then it will be used to
+		determine the specific RTPEngine to use.  Note that the socket
+		specified by "sock_var" must be a member of the current RTPEngine
+		Set context.
+	</para>
 	</section>
 
 	<section id="dependencies" xreflabel="Dependencies">

modules/rtpengine/rtpengine.c

@@ -344,6 +344,7 @@ static int add_rtpengine_socks(struct rtpe_set * rtpe_list, char * rtpengine);
 static int fixup_set_id(void ** param);
 static int fixup_free_set_id(void ** param);
 static int set_rtpengine_set_f(struct sip_msg * msg, rtpe_set_link_t *set_param);
+static int set_rtpengine_set_from_avp(struct sip_msg *msg);
 static struct rtpe_set * select_rtpe_set(int id_set);
 static struct rtpe_node *select_rtpe_node(str, struct rtpe_set *, struct rtpe_ignore_node *);
 static struct rtpe_node *lookup_rtpe_node(struct rtpe_set * rtpe_list, str *rtpe_url);
@@ -1865,7 +1866,6 @@ static int connect_rtpengines(int force_test)
 		return 0;
 	LM_DBG("[Re]connecting sockets (%d > %d)\n", *rtpe_no, rtpe_number);
 
-
 	if (*rtpe_no > rtpe_number) {
 		rtpe_socks = (int*)pkg_realloc(rtpe_socks, *rtpe_no * sizeof(int));
 		if (rtpe_socks==NULL) {
@@ -2827,30 +2827,56 @@ static bencode_item_t *rtpe_function_call(bencode_buffer_t *bencbuf, struct sip_
 	pv_value_t val;
 	struct rtpe_ignore_node *ignore_list = NULL;
 	int ret;
-
 	memset(&ng_flags, 0, sizeof(ng_flags));
 	error.len = 0;
 	error.s = "";
+	pv_value_t socket_val;
 
 	/*** get & init basic stuff needed ***/
 	if (rtpe_function_call_prepare(bencbuf, msg, op, &ng_flags, flags_str, body_in, extra_dict,&err) < 0)
 		goto error;
 
-	/*** send it out ***/
-	if (!set && (set=rtpe_ctx_set_get())==NULL )
-		set = *default_rtpe_set;
+	/*** set can be passed into this routine. If it is NULL, then attempt to determine it. ***/
+	if (!set) {
+		if ((set=rtpe_ctx_set_get())==NULL)
+			set = *default_rtpe_set;
+	}
+
+	/*** If the spvar "sock_var" has been specified, parse it into a (socket_val) STR variable ***/
+	if (spvar) {
+		memset(&socket_val, 0, sizeof(pv_value_t));
+		pv_get_spec_value(msg, spvar, &socket_val);
+	}
 
 	failed_node = NULL;
 
 	RTPE_START_READ();
 	do {
-		if (snode && snode->s) {
+		/*
+		 * Many rtpengine_* commands allow a "sock_var" to be specified by the script writer.
+		 * The "sock_var" will be populated by the rtpengine_* command with the RTPEngine that
+		 * was chosen for the operation.
+		 *
+		 * The "sock_var" can be used by subsequent rtpengine_* commands to direct the command
+		 * toward a specific RTPEngine instance.
+		 */
+		if (spvar && (socket_val.rs.len > 0)) {
+			LM_DBG("Sending command [%d] to RTPEngine socket: [%.*s] set id: [%d]\n", op, (int)(socket_val.rs.len), (char *)(socket_val.rs.s), set->id_set);
+			node = lookup_rtpe_node(set, &socket_val.rs);
+			if (node == NULL) {
+				RTPE_STOP_READ();
+				goto error;
+			}
+
+		} else if (snode && snode->s) {
 			if ((node = get_rtpe_node(snode, set)) == NULL && op == OP_OFFER)
 				node = select_rtpe_node(ng_flags.call_id, set, ignore_list);
 			snode = NULL;
+
 		} else {
 			node = select_rtpe_node(ng_flags.call_id, set, ignore_list);
 		}
+
 		if (!node) {
 			if (!err && !error.len)
 				err = "no available proxies";
@@ -3026,6 +3052,7 @@ rtpe_test(struct rtpe_node *node, int isdisabled, int force)
 {
 	bencode_buffer_t bencbuf;
 	bencode_item_t *dict;
+	bencode_item_t *resp;
 	char *cp;
 	int ret;
 
@@ -3055,14 +3082,14 @@ rtpe_test(struct rtpe_node *node, int isdisabled, int force)
 		goto error;
 	}
 
-	dict = bencode_decode_expect(&bencbuf, cp, ret, BENCODE_DICTIONARY);
-	if (!dict || bencode_dictionary_get_strcmp(dict, "result", "pong")) {
+	resp = bencode_decode_expect(&bencbuf, cp, ret, BENCODE_DICTIONARY);
+	if (!resp || bencode_dictionary_get_strcmp(resp, "result", "pong")) {
 		LM_ERR("proxy responded with invalid response\n");
 		goto error;
 	}
 
 	if (isdisabled)
-		LM_INFO("rtp proxy <%s> found, support for it %senabled\n",
+		LM_DBG("rtpengine <%s> found, support for it %senabled\n",
 				node->rn_url.s, force == 0 ? "re-" : "");
 
 	bencode_buffer_free(&bencbuf);