Final changes

Author: davidtrihy-genesys

modules/rest_client/rest_cb.c

@@ -111,7 +111,7 @@ size_t header_func(char *ptr, size_t size, size_t nmemb, void *userdata)
 
 int timer_cb(CURLM *multi_handle, long timeout_ms, void *cbp)
 {
-	LM_DBG("multi_handle timer called %d\n", timeout_ms);
+	LM_DBG("multi_handle timer called %ld\n", timeout_ms);
 	long *p = (long*) cbp;
 
 	*p = timeout_ms;
@@ -127,5 +127,5 @@ int prereq_callback(void *cbp,
 {
 	enum curl_status *p = (enum curl_status*) cbp;
 	*p = CURL_REQUEST_SENT;
-	return CURL_PREREQFUNC_OK;
+	return 0;
 }

modules/rest_client/rest_client.c

@@ -41,6 +41,7 @@
 #include "../tls_mgm/api.h"
 #include "rest_client.h"
 #include "rest_methods.h"
+#include "rest_sockets.h"
 #include "../../ssl_init_tweaks.h"
 #include "../../pt.h"
 #include "../../redact_pii.h"
@@ -52,7 +53,7 @@ long connection_timeout = 20; /* s */
 long connect_poll_interval = 20; /* ms */
 long connection_timeout_ms;
 int max_async_transfers = 100;
-int max_host_connection = 10;
+int max_connections = 100;
 long curl_timeout = 20;
 char *ssl_capath;
 unsigned int max_transfer_size = 10240; /* KB (10MB) */
@@ -113,6 +114,16 @@ static int w_async_rest_put(struct sip_msg *msg, async_ctx *ctx,
 			str *url, str *body, str *_ctype, pv_spec_t *body_pv,
 			pv_spec_t *ctype_pv, pv_spec_t *code_pv);
 
+// Temporary to expose in script
+static int w_async_rest_get_v2(struct sip_msg *msg, async_ctx *ctx, str *url,
+				pv_spec_t *body_pv, pv_spec_t *ctype_pv, pv_spec_t *code_pv);
+static int w_async_rest_post_v2(struct sip_msg *msg, async_ctx *ctx,
+			str *url, str *body, str *_ctype, pv_spec_t *body_pv,
+			pv_spec_t *ctype_pv, pv_spec_t *code_pv);
+static int w_async_rest_put_v2(struct sip_msg *msg, async_ctx *ctx,
+			str *url, str *body, str *_ctype, pv_spec_t *body_pv,
+			pv_spec_t *ctype_pv, pv_spec_t *code_pv);
+
 static int w_rest_append_hf(struct sip_msg *msg, str *hfv);
 static int w_rest_init_client_tls(struct sip_msg *msg, str *tls_client_dom);
 int validate_curl_http_version(const int *http_version);
@@ -148,6 +159,25 @@ static const acmd_export_t acmds[] = {
 		{CMD_PARAM_VAR,0,0},
 		{CMD_PARAM_VAR|CMD_PARAM_OPT,0,0},
 		{CMD_PARAM_VAR|CMD_PARAM_OPT,0,0}, {0,0,0}}},
+	{"rest_get_v2",(acmd_function)w_async_rest_get_v2, {
+		{CMD_PARAM_STR,0,0},
+		{CMD_PARAM_VAR,0,0},
+		{CMD_PARAM_VAR|CMD_PARAM_OPT,0,0},
+		{CMD_PARAM_VAR|CMD_PARAM_OPT,0,0}, {0,0,0}}},
+	{"rest_post_v2",(acmd_function)w_async_rest_post_v2, {
+		{CMD_PARAM_STR,0,0},
+		{CMD_PARAM_STR,0,0},
+		{CMD_PARAM_STR|CMD_PARAM_OPT,0,0},
+		{CMD_PARAM_VAR,0,0},
+		{CMD_PARAM_VAR|CMD_PARAM_OPT,0,0},
+		{CMD_PARAM_VAR|CMD_PARAM_OPT,0,0}, {0,0,0}}},
+	{"rest_put_v2",(acmd_function)w_async_rest_put_v2, {
+		{CMD_PARAM_STR,0,0},
+		{CMD_PARAM_STR,0,0},
+		{CMD_PARAM_STR|CMD_PARAM_OPT,0,0},
+		{CMD_PARAM_VAR,0,0},
+		{CMD_PARAM_VAR|CMD_PARAM_OPT,0,0},
+		{CMD_PARAM_VAR|CMD_PARAM_OPT,0,0}, {0,0,0}}},
 	{0,0,{{0,0,0}}}
 };
 
@@ -178,6 +208,28 @@ static const cmd_export_t cmds[] = {
 		{CMD_PARAM_VAR|CMD_PARAM_OPT,0,0},
 		{CMD_PARAM_VAR|CMD_PARAM_OPT,0,0}, {0,0,0}},
 		ALL_ROUTES},
+	{"rest_get_v2",(cmd_function)w_rest_get, {
+		{CMD_PARAM_STR,0,0},
+		{CMD_PARAM_VAR,0,0},
+		{CMD_PARAM_VAR|CMD_PARAM_OPT,0,0},
+		{CMD_PARAM_VAR|CMD_PARAM_OPT,0,0}, {0,0,0}},
+		ALL_ROUTES},
+	{"rest_post_v2",(cmd_function)w_rest_post, {
+		{CMD_PARAM_STR,0,0},
+		{CMD_PARAM_STR,0,0},
+		{CMD_PARAM_STR|CMD_PARAM_OPT,0,0},
+		{CMD_PARAM_VAR,0,0},
+		{CMD_PARAM_VAR|CMD_PARAM_OPT,0,0},
+		{CMD_PARAM_VAR|CMD_PARAM_OPT,0,0}, {0,0,0}},
+		ALL_ROUTES},
+	{"rest_put_v2",(cmd_function)w_rest_put, {
+		{CMD_PARAM_STR,0,0},
+		{CMD_PARAM_STR,0,0},
+		{CMD_PARAM_STR|CMD_PARAM_OPT,0,0},
+		{CMD_PARAM_VAR,0,0},
+		{CMD_PARAM_VAR|CMD_PARAM_OPT,0,0},
+		{CMD_PARAM_VAR|CMD_PARAM_OPT,0,0}, {0,0,0}},
+		ALL_ROUTES},
 	{"rest_append_hf",(cmd_function)w_rest_append_hf, {
 		{CMD_PARAM_STR,0,0}, {0,0,0}},
 		ALL_ROUTES},
@@ -200,7 +252,7 @@ static const trans_export_t trans[] = {
 };
 
 static int warm_pool_urls(modparam_t type, void *val) {
-	int num_conns;
+	unsigned int num_conns;
 	char *mod_param, *delim, *host;
 	size_t delim_index, string_end;
 	preconnect_urls *tmp;
@@ -272,7 +324,6 @@ static const param_export_t params[] = {
 	{ "connection_timeout",	INT_PARAM, &connection_timeout	},
 	{ "connect_poll_interval", INT_PARAM, &connect_poll_interval },
 	{ "max_async_transfers", INT_PARAM, &max_async_transfers },
-	{ "max_host_connection", INT_PARAM, &max_host_connection},
 	{ "max_transfer_size",	INT_PARAM, &max_transfer_size	},
 	{ "curl_timeout",		INT_PARAM, &curl_timeout		},
 	{ "ssl_capath",			STR_PARAM, &ssl_capath			},
@@ -282,6 +333,8 @@ static const param_export_t params[] = {
 	{ "enable_expect_100",	INT_PARAM, &enable_expect_100	},
 	{ "no_concurrent_connects",	INT_PARAM, &no_concurrent_connects	},
 	{ "curl_conn_lifetime",	INT_PARAM, &curl_conn_lifetime	},
+	{ "use_multi_socket_api",	INT_PARAM, &use_multi_socket_api	},
+	{ "max_connections",	INT_PARAM, &max_connections	},
 	{ "warm_pool_urls",		STR_PARAM|USE_FUNC_PARAM,
 		(void*)&warm_pool_urls },
 	{ 0, 0, 0 }
@@ -895,3 +948,142 @@ static int w_rest_init_client_tls(struct sip_msg *msg, str *tls_client_dom)
 {
 	return rest_init_client_tls(msg, tls_client_dom);
 }
+
+// Temporary duplication for feature toggle in script
+int async_rest_method_v2(enum rest_client_method method, struct sip_msg *msg,
+                         char *url, str *body, str *ctype, async_ctx *ctx,
+                         pv_spec_p body_pv, pv_spec_p ctype_pv, pv_spec_p code_pv)
+{
+	rest_async_param *param;
+	pv_value_t val;
+	char *host;
+	int read_fd, rc, lrc = RCL_OK;
+
+	param = pkg_malloc(sizeof *param);
+	if (!param) {
+		LM_ERR("no more shm\n");
+		return RCL_INTERNAL_ERR;
+	}
+	memset(param, '\0', sizeof *param);
+
+	if (no_concurrent_connects && (lrc=rcl_acquire_url(url, &host)) < RCL_OK)
+		return lrc;
+
+	param->timeout_s = (ctx->timeout_s && ctx->timeout_s < curl_timeout) ?
+			ctx->timeout_s : curl_timeout;
+
+	rc = start_async_http_req_v2(msg, method, url, body, ctype,
+			param, &param->body, ctype_pv ? &param->ctype : NULL, &read_fd);
+
+	/* error occurred; no transfer done */
+	if (read_fd == ASYNC_NO_IO) {
+		ctx->resume_param = NULL;
+		ctx->resume_f = NULL;
+		if (code_pv) {
+			val.flags = PV_VAL_INT|PV_TYPE_INT;
+			val.ri = 0;
+			if (pv_set_value(msg, (pv_spec_p)code_pv, 0, &val) != 0)
+				LM_ERR("failed to set output code pv\n");
+		}
+
+		/* keep default async status of NO_IO */
+		pkg_free(param);
+		goto done;
+	}
+
+	/* the TCP connection is established, async started with success */
+
+	if (lrc == RCL_OK_LOCKED)
+		rcl_release_url(host, rc == RCL_OK);
+
+	ctx->resume_f = resume_async_http_req_v2;
+	ctx->timeout_f = time_out_async_http_req_v2;
+	ctx->timeout_s = param->timeout_s;
+
+	param->method = method;
+	param->body_pv = (pv_spec_p)body_pv;
+	param->ctype_pv = (pv_spec_p)ctype_pv;
+	param->code_pv = (pv_spec_p)code_pv;
+	ctx->resume_param = param;
+
+	async_status = read_fd;
+	return 1;
+
+done:
+	if (lrc == RCL_OK_LOCKED)
+		rcl_release_url(host, rc == RCL_OK);
+	return rc;
+}
+
+static int w_async_rest_get_v2(struct sip_msg *msg, async_ctx *ctx, str *url,
+				pv_spec_t *body_pv, pv_spec_t *ctype_pv, pv_spec_t *code_pv)
+{
+	str url_nt;
+	int rc;
+
+	if (pkg_nt_str_dup(&url_nt, url) < 0) {
+		LM_ERR("No more pkg memory\n");
+		return RCL_INTERNAL_ERR;
+	}
+
+	LM_DBG("async rest get %.*s %p %p %p\n", url->len, url->s,
+			body_pv, ctype_pv, code_pv);
+
+	rc = async_rest_method_v2(REST_CLIENT_GET, msg, url_nt.s, NULL, NULL, ctx,
+				body_pv, ctype_pv, code_pv);
+
+	pkg_free(url_nt.s);
+	return rc;
+}
+
+static int w_async_rest_post_v2(struct sip_msg *msg, async_ctx *ctx,
+			str *url, str *body, str *_ctype, pv_spec_t *body_pv,
+			pv_spec_t *ctype_pv, pv_spec_t *code_pv)
+{
+	str ctype = { NULL, 0 };
+	str url_nt;
+	int rc;
+
+	if (pkg_nt_str_dup(&url_nt, url) < 0) {
+		LM_ERR("No more pkg memory\n");
+		return RCL_INTERNAL_ERR;
+	}
+
+	if (_ctype)
+		ctype = *_ctype;
+
+	LM_DBG("async rest post '%.*s' %p %p %p\n", url->len, url->s,
+			body_pv, ctype_pv, code_pv);
+
+	rc = async_rest_method_v2(REST_CLIENT_POST, msg, url_nt.s, body, &ctype, ctx,
+							body_pv, ctype_pv, code_pv);
+
+	pkg_free(url_nt.s);
+	return rc;
+}
+
+static int w_async_rest_put_v2(struct sip_msg *msg, async_ctx *ctx,
+			str *url, str *body, str *_ctype, pv_spec_t *body_pv,
+			pv_spec_t *ctype_pv, pv_spec_t *code_pv)
+{
+	str ctype = { NULL, 0 };
+	str url_nt;
+	int rc;
+
+	if (pkg_nt_str_dup(&url_nt, url) < 0) {
+		LM_ERR("No more pkg memory\n");
+		return RCL_INTERNAL_ERR;
+	}
+
+	if (_ctype)
+		ctype = *_ctype;
+
+	LM_DBG("async rest put '%.*s' %p %p %p\n",
+		url->len, url->s, body_pv, ctype_pv, code_pv);
+
+	rc = async_rest_method_v2(REST_CLIENT_PUT, msg, url_nt.s, body, &ctype, ctx,
+						body_pv, ctype_pv, code_pv);
+
+	pkg_free(url_nt.s);
+	return rc;
+}

modules/rest_client/rest_client.h

@@ -32,7 +32,7 @@ enum tr_rest_subtype {
 
 typedef struct _preconnect_urls {
 	char *url;
-	long connections;
+	unsigned int connections;
 	struct _preconnect_urls *next;
 } preconnect_urls;