feat: add timeout_milliseconds

Author: steve-chavez

src/worker.c

@@ -138,7 +138,7 @@ pg_text_array_to_slist(ArrayType *array, struct curl_slist *headers)
 	return headers;
 }
 
-static int init(CURLM *cm, char *method, char *url, struct curl_slist *reqHeaders, char *reqBody, int64 id, HTAB *curlDataMap)
+static int init(CURLM *cm, char *method, char *url, int timeout_milliseconds, struct curl_slist *reqHeaders, char *reqBody, int64 id, HTAB *curlDataMap)
 {
 	CURL *eh = curl_easy_init();
 
@@ -178,6 +178,7 @@ static int init(CURLM *cm, char *method, char *url, struct curl_slist *reqHeader
 	curl_easy_setopt(eh, CURLOPT_HEADER, 0L);
 	curl_easy_setopt(eh, CURLOPT_URL, url);
 	curl_easy_setopt(eh, CURLOPT_HTTPHEADER, reqHeaders);
+	curl_easy_setopt(eh, CURLOPT_TIMEOUT_MS, timeout_milliseconds);
 	curl_easy_setopt(eh, CURLOPT_PRIVATE, id);
 	if (log_min_messages <= DEBUG1)
 		curl_easy_setopt(eh, CURLOPT_VERBOSE, 1L);
@@ -286,6 +287,7 @@ worker_main(Datum main_arg)
 				q.id,\
 				q.method,\
 				q.url,\
+				q.timeout_milliseconds,\
 				array(\
 					select key || ': ' || value from jsonb_each_text(q.headers)\
 				),\
@@ -325,21 +327,22 @@ worker_main(Datum main_arg)
 						int64 id = DatumGetInt64(SPI_getbinval(SPI_tuptable->vals[j], SPI_tuptable->tupdesc, 1, &tupIsNull));
 						char *method = TextDatumGetCString(SPI_getbinval(SPI_tuptable->vals[j], SPI_tuptable->tupdesc, 2, &tupIsNull));
 						char *url = TextDatumGetCString(SPI_getbinval(SPI_tuptable->vals[j], SPI_tuptable->tupdesc, 3, &tupIsNull));
+						int32 timeout_milliseconds = DatumGetInt32(SPI_getbinval(SPI_tuptable->vals[j], SPI_tuptable->tupdesc, 4, &tupIsNull));
 
 						Datum headersBin;
 						Datum bodyBin;
 						ArrayType *pgHeaders;
 						char *body = NULL;
 
-						headersBin = SPI_getbinval(SPI_tuptable->vals[j], SPI_tuptable->tupdesc, 4, &tupIsNull);
+						headersBin = SPI_getbinval(SPI_tuptable->vals[j], SPI_tuptable->tupdesc, 5, &tupIsNull);
 						if (!tupIsNull) {
 							pgHeaders = DatumGetArrayTypeP(headersBin);
 							headers = pg_text_array_to_slist(pgHeaders, headers);
 						}
-						bodyBin = SPI_getbinval(SPI_tuptable->vals[j], SPI_tuptable->tupdesc, 5, &tupIsNull);
+						bodyBin = SPI_getbinval(SPI_tuptable->vals[j], SPI_tuptable->tupdesc, 6, &tupIsNull);
 						if (!tupIsNull) body = TextDatumGetCString(bodyBin);
 
-						res = init(cm, method, url, headers, body, id, curlDataMap);
+						res = init(cm, method, url, timeout_milliseconds, headers, body, id, curlDataMap);
 
 						if(res) {
 							elog(ERROR, "error: init() returned %d\n", res);

test/test_http_timeout.py

@@ -0,0 +1,52 @@
+import time
+
+import pytest
+from sqlalchemy import text
+
+def test_http_get_err_on_default_timeout(sess):
+    """net.http_get with default timeout errs on a slow reply"""
+
+    (request_id,) = sess.execute(
+        """
+        select net.http_get(url := 'http://localhost:8080/slow-reply');
+    """
+    ).fetchone()
+
+    sess.commit()
+
+    time.sleep(2)
+
+    (response,) = sess.execute(
+        text(
+            """
+        select error_msg from net._http_response where id = :request_id;
+    """
+        ),
+        {"request_id": request_id},
+    ).fetchone()
+
+    assert response == u'Timeout was reached'
+
+def test_http_get_succeed_with_gt_timeout(sess):
+    """net.http_get with a timeout greater than the default one succeeds on a slow reply"""
+
+    (request_id,) = sess.execute(
+        """
+        select net.http_get(url := 'http://localhost:8080/slow-reply', timeout_milliseconds := 2500);
+    """
+    ).fetchone()
+
+    sess.commit()
+
+    time.sleep(2.5)
+
+    (response,) = sess.execute(
+        text(
+            """
+        select content from net._http_response where id = :request_id;
+    """
+        ),
+        {"request_id": request_id},
+    ).fetchone()
+
+    assert 'after 2 seconds' in str(response)