httpc: introduce stream input/output interface

Add a streaming data input/output object for http.client. The
input/output object can be created using the same methods and the
same options as a normal request, but with a new option
{chunked = true}.

Closes #7845

@TarantoolBot document
Title: Stream input/output interface for http.client

An uncompleted io object has only headers and cookies fields. A
completed io object has the same fields as a normal request, but
without the `body` field.

The io object interface looks like the socket object interface
and should have the same description:

```
io_object:read(chunk[, timeout])
io_object:read(delimiter[, timeout])
io_object:read({chunk = chunk, delimiter = delimiter}[, timeout])
io_object:write(data[, timeout])
```

The difference is in the method `finish`. Unlike socket:close()
it has an optional parameter `timeout`:

```
io_object:finish([timeout])
```

Be careful, the call may yield a fiber. The idea is to wait
until a HTTP connection is finished by the server-side or
force finish the connection from client-time after a timeout
value.

The default timeout value is 10 seconds for all methods.

Usage example:

```lua
local io = httpc:post(url, nil, {chunked = true})

local write_chan = fiber.channel()
fiber.new(function()
    fiber.name("write to " .. url)
    while true do
        local data = write_chan:get()
        if data == nil then
            break
        end
        io:write(data, 1)
    end
end)

local recvdata
while recvdata = io:read('\r\n', 1) do
    local decoded = json.decode(recvdata)
    if condition(decoded) then
        write_chan:put(data)
    end
    if condition(decoded) then
        io:finish(1)
    end
end
write_chan:close()
```

See also:
* https://www.tarantool.io/en/doc/latest/reference/reference_lua/socket/#lua-function.socket_object.read
* #7845 (comment)
* #7845 (comment)

changelogs/unreleased/gh-7845-httpc-stream-io-interface.md

@@ -0,0 +1,3 @@
+## feature/lua/http client
+
+* Introduce stream input/output interface for http.client (gh-7845).

src/curl.c

@@ -91,6 +91,8 @@ curl_multi_process(CURLM *multi, curl_socket_t sockfd, int events)
 			errinj->bparam = false;
 #endif
 		fiber_cond_signal(&request->cond);
+		if (request->done_handler)
+			request->done_handler(request->done_handler_arg);
 	}
 }
 
@@ -277,6 +279,8 @@ curl_request_create(struct curl_request *curl_request)
 	curl_request->in_progress = false;
 	curl_request->code = CURLE_OK;
 	fiber_cond_create(&curl_request->cond);
+	curl_request->done_handler = NULL;
+	curl_request->done_handler_arg = NULL;
 	return 0;
 }
 
@@ -288,15 +292,42 @@ curl_request_destroy(struct curl_request *curl_request)
 	fiber_cond_destroy(&curl_request->cond);
 }
 
+/**
+ * Set diag if mcode != CURLM_OK.
+ */
+static inline void
+curl_diag_set_merror(CURLMcode mcode)
+{
+	if (mcode != CURLM_OK) {
+		switch (mcode) {
+		case CURLM_OUT_OF_MEMORY:
+			diag_set(OutOfMemory, 0, "curl", "internal");
+			break;
+		default:
+			errno = EINVAL;
+			diag_set(SystemError, "curl_multi_error: %s",
+				 curl_multi_strerror(mcode));
+		}
+	}
+}
+
 CURLMcode
-curl_execute(struct curl_request *curl_request, struct curl_env *env,
-	     double timeout)
+curl_request_start(struct curl_request *curl_request, struct curl_env *env)
 {
 	CURLMcode mcode;
 	curl_request->in_progress = true;
 	mcode = curl_multi_add_handle(env->multi, curl_request->easy);
-	if (mcode != CURLM_OK)
-		goto curl_merror;
+	curl_diag_set_merror(mcode);
+
+	return mcode;
+}
+
+CURLMcode
+curl_request_finish(struct curl_request *curl_request, struct curl_env *env,
+		    double timeout)
+{
+	CURLMcode mcode;
+
 	ERROR_INJECT_YIELD(ERRINJ_HTTP_RESPONSE_ADD_WAIT);
 	/* Don't wait on a cond if request has already failed or finished. */
 	if (curl_request->code == CURLE_OK && curl_request->in_progress) {
@@ -307,20 +338,8 @@ curl_execute(struct curl_request *curl_request, struct curl_env *env,
 		--env->stat.active_requests;
 	}
 	mcode = curl_multi_remove_handle(env->multi, curl_request->easy);
-	if (mcode != CURLM_OK)
-		goto curl_merror;
-
-	return CURLM_OK;
-
-curl_merror:
-	switch (mcode) {
-	case CURLM_OUT_OF_MEMORY:
-		diag_set(OutOfMemory, 0, "curl", "internal");
-		break;
-	default:
-		errno = EINVAL;
-		diag_set(SystemError, "curl_multi_error: %s",
-			 curl_multi_strerror(mcode));
-	}
+	curl_diag_set_merror(mcode);
+	curl_request->in_progress = false;
+
 	return mcode;
 }

src/curl.h

@@ -62,6 +62,12 @@ struct curl_env {
 	struct curl_stat stat;
 };
 
+/**
+ * CURL request completed handler
+ */
+typedef void
+(*curl_done_handler)(void *arg);
+
 /**
  * CURL Request
  */
@@ -77,6 +83,15 @@ struct curl_request {
 	 * until the handler (callback function) gives a signal within variable.
 	 * */
 	struct fiber_cond cond;
+	/**
+	 * The curl-driver calls the handler after the request execution has
+	 * been completed.
+	 */
+	curl_done_handler done_handler;
+	/**
+	 * The argument for done_handler.
+	 */
+	void *done_handler_arg;
 };
 
 /**
@@ -113,13 +128,22 @@ void
 curl_request_destroy(struct curl_request *curl_request);
 
 /**
- * Execute CURL request
+ * Start executing the CURL request
+ * @param curl_request request
+ * @param env environment
+ * @param curl_request request
+ */
+CURLMcode
+curl_request_start(struct curl_request *curl_request, struct curl_env *env);
+
+/**
+ * Wait for the CURL request to be completed or aborts the request by timeout
  * @param curl_request request
  * @param env environment
  * @param timeout - timeout of waiting for libcurl api
  */
 CURLMcode
-curl_execute(struct curl_request *curl_request, struct curl_env *env,
-	     double timeout);
+curl_request_finish(struct curl_request *curl_request, struct curl_env *env,
+		    double timeout);
 
 #endif /* TARANTOOL_CURL_H_INCLUDED */