feat: support POST

Author: soedirgo

sql/pg_net--0.1.sql

@@ -1,12 +1,21 @@
 create schema if not exists net;
 
+create domain http_method as text
+check (
+    value ilike 'get' or
+    value ilike 'post'
+);
+
 -- Store pending requests. The background worker reads from here
 -- API: Private
 create table net.http_request_queue(
     id bigserial primary key,
+    method http_method not null,
     url text not null,
     params jsonb not null,
     headers jsonb not null,
+    content_type text,
+    body text,
     timeout_milliseconds int not null,
     -- Available for delete after this date
     delete_after timestamp not null
@@ -85,8 +94,8 @@ declare
     respone_rec net.http_response;
 begin
     -- Add to the request queue
-    insert into net.http_request_queue(url, params, headers, timeout_milliseconds, delete_after)
-    values (url, params, headers, timeout_milliseconds, timezone('utc', now()) + ttl)
+    insert into net.http_request_queue(method, url, params, headers, content_type, body, timeout_milliseconds, delete_after)
+    values ('GET', url, params, headers, null, null, timeout_milliseconds, timezone('utc', now()) + ttl)
     returning id
     into request_id;
 
@@ -101,6 +110,53 @@ begin
 end
 $$;
 
+-- Interface to make an async request
+-- API: Public
+create or replace function net.http_post(
+    -- url for the request
+    url text,
+    -- ContentType of the POST request
+    content_type text,
+    -- body of the POST request
+    body text,
+    -- key/value pairs to be url encoded and appended to the `url`
+    params jsonb DEFAULT '{}'::jsonb,
+    -- key/values to be included in request headers
+    headers jsonb DEFAULT '{}'::jsonb,
+    -- the maximum number of milliseconds the request may take before being cancelled
+    timeout_milliseconds int DEFAULT 1000,
+    -- the minimum amount of time the response should be persisted
+    ttl interval default '3 days',
+    -- when `true`, return immediately. when `false` wait for the request to complete before returning
+    async bool default true
+)
+    -- request_id reference
+    returns bigint
+    strict
+    volatile
+    parallel safe
+    language plpgsql
+as $$
+declare
+    request_id bigint;
+    respone_rec net.http_response;
+begin
+    -- Add to the request queue
+    insert into net.http_request_queue(method, url, params, headers, content_type, body, timeout_milliseconds, delete_after)
+    values ('POST', url, params, headers, content_type, body, timeout_milliseconds, timezone('utc', now()) + ttl)
+    returning id
+    into request_id;
+
+    -- If request is async, return id immediately
+    if async then
+        return request_id;
+    end if;
+
+    -- If sync, wait for the request to complete before returning
+    perform net._await_response(request_id);
+    return request_id;
+end
+$$;
 
 -- Collect respones of an http request
 -- API: Public

src/worker.c

@@ -23,6 +23,7 @@
 
 #include "utils/jsonb.h"
 
+#include <curl/curl.h>
 #include <curl/multi.h>
 
 PG_MODULE_MAGIC;
@@ -102,7 +103,7 @@ header_cb(void *contents, size_t size, size_t nmemb, void *userp)
 	return realsize;
 }
 
-static int init(CURLM *cm, char *url, int64 id, HTAB *curlDataMap)
+static int init(CURLM *cm, char *method, char *url, char *contentType, char *reqBody, int64 id, HTAB *curlDataMap)
 {
 	CURL *eh = curl_easy_init();
 
@@ -120,6 +121,25 @@ static int init(CURLM *cm, char *url, int64 id, HTAB *curlDataMap)
 		cdata->headers = headers;
 	}
 
+	if (contentType) {
+		struct curl_slist *reqHeaders = NULL;
+		reqHeaders = curl_slist_append(reqHeaders, contentType);
+		curl_easy_setopt(eh, CURLOPT_HTTPHEADER, reqHeaders);
+	}
+
+	if (strcasecmp(method, "GET") == 0) {
+		if (reqBody) {
+			curl_easy_setopt(eh, CURLOPT_POSTFIELDS, reqBody);
+			curl_easy_setopt(eh, CURLOPT_CUSTOMREQUEST, "GET");
+		}
+	} else if (strcasecmp(method, "POST") == 0) {
+		if (reqBody) {
+			curl_easy_setopt(eh, CURLOPT_POSTFIELDS, reqBody);
+		}
+	} else {
+		elog(ERROR, "error: Unsupported request method %s\n", method);
+	}
+
 	curl_easy_setopt(eh, CURLOPT_WRITEFUNCTION, body_cb);
 	curl_easy_setopt(eh, CURLOPT_WRITEDATA, cdata->body);
 	curl_easy_setopt(eh, CURLOPT_HEADERFUNCTION, header_cb);
@@ -204,7 +224,7 @@ worker_main(Datum main_arg)
 
 		appendStringInfo(&select_query, "\
 			SELECT\
-			  q.id, q.url\
+			  q.id, q.method, q.url, q.content_type, q.body\
 			FROM net.http_request_queue q \
 			LEFT JOIN net.http_response r ON q.id = r.id \
 			WHERE r.id IS NULL");
@@ -223,12 +243,30 @@ worker_main(Datum main_arg)
 
 			for (int j = 0; j < SPI_processed; j++)
 			{
+				    StringInfoData content_type;
+
 					int64 id = DatumGetInt64(SPI_getbinval(SPI_tuptable->vals[j], SPI_tuptable->tupdesc, 1, &tupIsNull));
-					char *url = TextDatumGetCString(SPI_getbinval(SPI_tuptable->vals[j], SPI_tuptable->tupdesc, 2, &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));
+					Datum contentTypeBin;
+					Datum bodyBin;
+					char *contentType = NULL;
+					char *body = NULL;
+
+					contentTypeBin = SPI_getbinval(SPI_tuptable->vals[j], SPI_tuptable->tupdesc, 4, &tupIsNull);
+					if (!tupIsNull) {
+						initStringInfo(&content_type);
+						appendStringInfo(&content_type, "Content-Type: %s", TextDatumGetCString(contentTypeBin));
+						contentType = content_type.data;
+					}
+					bodyBin = SPI_getbinval(SPI_tuptable->vals[j], SPI_tuptable->tupdesc, 5, &tupIsNull);
+					if (!tupIsNull) body = TextDatumGetCString(bodyBin);
+
+					elog(DEBUG1, "Making a %s request to %s with id %ld", method, url, id);
 
-					elog(DEBUG1, "Making a request to %s with id %ld", url, id);
+					res = init(cm, method, url, contentType, body, id, curlDataMap);
 
-					res = init(cm, url, id, curlDataMap);
+					/* pfree(content_type.data); */
 
 					if(res) {
 						elog(ERROR, "error: init() returned %d\n", res);