Add direct (same process) mode for packet stream

Signed-off-by: Andrew Opalach <andrew@akon.city>

2 files changed, 48 insertions, 39 deletions

diff --git a/src/curl/http.c b/src/curl/http.c
index 60d26f1..d040f53 100644
--- a/src/curl/http.c
+++ b/src/curl/http.c
@@ -6,17 +6,17 @@ static void check_status_codes(struct nn_curl *curl)
 {
     struct nn_http *http = (struct nn_http *)curl;
 
-    if (http->status_code <= 0) {
+    if (http->status_code <= 0L) {
         curl_easy_getinfo(curl->handle, CURLINFO_RESPONSE_CODE, &http->status_code);
-        if (http->status_code > 0) {
-            http->callback(http->userdata, NNWT_HTTP_RESPONSE_CODE, NULL, http->status_code);
+        if (http->status_code > 0L) {
+            http->callback(http->userdata, NNWT_HTTP_RESPONSE_CODE, NULL, &http->status_code);
         }
     }
 
-    if (http->content_length < 0 && http->status_code == 200) {
+    if (http->content_length < 0L && http->status_code == 200L) {
         curl_easy_getinfo(curl->handle, CURLINFO_CONTENT_LENGTH_DOWNLOAD_T, &http->content_length);
-        if (http->content_length >= 0) {
-            http->callback(http->userdata, NNWT_HTTP_CONTENT_LENGTH, NULL, (s64)http->content_length);
+        if (http->content_length >= 0L) {
+            http->callback(http->userdata, NNWT_HTTP_CONTENT_LENGTH, NULL, &http->content_length);
         }
     }
 }
@@ -35,7 +35,8 @@ static void http_handle_events(void *userdata, struct nn_curl *curl)
             CURLcode code = msg->data.result;
             if (code != CURLE_OK) {
                 al_log_error("http", "Curl error: %s.", curl_easy_strerror(code));
-                http->callback(http->userdata, NNWT_HTTP_ERROR, NULL, -1);
+                http->status_code = -1L;
+                http->callback(http->userdata, NNWT_HTTP_ERROR, NULL, &http->status_code);
                 return;
             }
 
@@ -45,26 +46,26 @@ static void http_handle_events(void *userdata, struct nn_curl *curl)
 
             check_status_codes(curl);
 
-            if (http->status_code == 200) {
-                http->callback(http->userdata, NNWT_HTTP_FINISHED, NULL, http->status_code);
-            } else if (http->status_code / 100 == 3) {
-                http->callback(http->userdata, NNWT_HTTP_REDIRECT, NULL, http->status_code);
+            if (http->status_code == 200L) {
+                http->callback(http->userdata, NNWT_HTTP_FINISHED, NULL, &http->status_code);
+            } else if (http->status_code / 100L == 3L) {
+                http->callback(http->userdata, NNWT_HTTP_REDIRECT, NULL, &http->status_code);
 
                 char *redirect_url = NULL;
                 curl_easy_getinfo(curl->handle, CURLINFO_REDIRECT_URL, &redirect_url);
                 if (!redirect_url) {
-                    http->callback(http->userdata, NNWT_HTTP_ERROR, NULL, http->status_code);
+                    http->callback(http->userdata, NNWT_HTTP_ERROR, NULL, &http->status_code);
                     return;
                 }
                 nn_curl_set_url(curl, al_str_cr(redirect_url));
 
-                http->status_code = -1;
-                http->content_length = -1;
+                http->status_code = -1L;
+                http->content_length = -1L;
                 if (!nn_curl_add_handle(curl)) {
                     return;
                 }
             } else {
-                http->callback(http->userdata, NNWT_HTTP_ERROR, NULL, http->status_code);
+                http->callback(http->userdata, NNWT_HTTP_ERROR, NULL, &http->status_code);
             }
 
             return;
@@ -81,8 +82,8 @@ static void http_handle_events(void *userdata, struct nn_curl *curl)
 
 bool nn_http_init(struct nn_http *http)
 {
-    http->status_code = -1;
-    http->content_length = -1;
+    http->status_code = -1L;
+    http->content_length = -1L;
     http->headers = NULL;
     return nn_curl_init(&http->curl);
 }
@@ -124,40 +125,49 @@ static size_t stream_write_callback(char *buffer, size_t size, size_t nmemb, voi
 {
     struct nn_http *http = (struct nn_http *)userdata;
     check_status_codes(&http->curl);
-    return http->callback(http->userdata, NNWT_HTTP_WRITE, (u8 *)buffer, size * nmemb);
+    size_t n = size * nmemb;
+    http->callback(http->userdata, NNWT_HTTP_WRITE, (u8 *)buffer, &n);
+    return n;
 }
 
 static size_t stream_read_callback(char *buffer, size_t size, size_t nmemb, void *userdata)
 {
     struct nn_http *http = (struct nn_http *)userdata;
-    return http->callback(http->userdata, NNWT_HTTP_READ, (u8 *)buffer, size * nmemb);
+    size_t n = size * nmemb;
+    http->callback(http->userdata, NNWT_HTTP_READ, (u8 *)buffer, &n);
+    return n;
 }
 
-static size_t request_callback(void *userdata, u8 op, u8 *buf, s64 int0)
+static void request_callback(void *userdata, u8 op, u8 *buf, void *opaque)
 {
     struct nn_http_request *request = (struct nn_http_request *)userdata;
     switch (op) {
     case NNWT_HTTP_READ: {
-        al_assert(request->pointer >= 0);
-
+        size_t n = *(size_t *)opaque;
         size_t size = nn_buffer_get_size(&request->payload);
-        if (request->pointer + int0 > (off_t)size) {
-            int0 = size - request->pointer;
+        if (request->pointer + n > size) {
+            n = size - request->pointer;
         }
 
-        nn_buffer_read(&request->payload, buf, request->pointer, int0);
-        request->pointer += int0;
+        nn_buffer_read(&request->payload, buf, request->pointer, n);
+        request->pointer += n;
+
+        *(size_t *)opaque = n;
 
         break;
     }
-    case NNWT_HTTP_WRITE:
-        nn_buffer_append(&request->response, buf, int0);
+    case NNWT_HTTP_WRITE: {
+        size_t n = *(size_t *)opaque;
+        nn_buffer_append(&request->response, buf, n);
         break;
+    }
     case NNWT_HTTP_RESPONSE_CODE:
         break;
-    case NNWT_HTTP_CONTENT_LENGTH:
-        nn_buffer_ensure_space(&request->response, int0);
+    case NNWT_HTTP_CONTENT_LENGTH: {
+        curl_off_t length = *(curl_off_t *)opaque;
+        nn_buffer_ensure_space(&request->response, length);
         break;
+    }
     case NNWT_HTTP_REDIRECT:
         break;
     case NNWT_HTTP_FINISHED:
@@ -167,7 +177,6 @@ static size_t request_callback(void *userdata, u8 op, u8 *buf, s64 int0)
         request->callback(request->userdata, request, false);
         break;
     }
-    return int0;
 }
 
 static bool init_request_internal(struct nn_http *http, u8 method)
@@ -201,7 +210,7 @@ static bool init_request_internal(struct nn_http *http, u8 method)
 }
 
 bool nn_http_request_stream(struct nn_http *http, u8 method, struct nn_event_loop *loop,
-    size_t (*callback)(void *, u8, u8 *, s64), void *userdata)
+    void (*callback)(void *, u8, u8 *, void *), void *userdata)
 {
     struct nn_curl *curl = &http->curl;
     al_assert(curl->loop == NULL);
@@ -230,9 +239,9 @@ bool nn_http_request(struct nn_http_request *request, u8 method, struct nn_event
     request->http.callback = request_callback;
     request->http.userdata = request;
 
-    size_t payload = nn_buffer_get_size(&request->payload);
-    if (payload > 0) {
-        curl_easy_setopt(curl->handle, CURLOPT_POSTFIELDSIZE, (s64)payload);
+    long payload = (long)nn_buffer_get_size(&request->payload);
+    if (payload > 0L) {
+        curl_easy_setopt(curl->handle, CURLOPT_POSTFIELDSIZE, payload);
     }
 
     request->callback = callback;
diff --git a/src/curl/http.h b/src/curl/http.h
index d5efcf9..952a74a 100644
--- a/src/curl/http.h
+++ b/src/curl/http.h
@@ -26,16 +26,16 @@ enum {
 
 struct nn_http {
     struct nn_curl curl;
-    s64 status_code;
+    long status_code;
     curl_off_t content_length;
     struct curl_slist *headers;
-    size_t (*callback)(void *, u8, u8 *, s64);
+    void (*callback)(void *, u8, u8 *, void *);
     void *userdata;
 };
 
 struct nn_http_request {
     struct nn_http http;
-    off_t pointer;
+    size_t pointer;
     struct nn_buffer payload;
     struct nn_buffer response;
     void (*callback)(void *, struct nn_http_request *, bool);
@@ -49,7 +49,7 @@ void nn_http_add_header(struct nn_http *http, str *header);
 void nn_http_set_range(struct nn_http *http, size_t start, size_t end);
 void nn_http_close(struct nn_http *http);
 bool nn_http_request_stream(struct nn_http *http, u8 method, struct nn_event_loop *loop,
-    size_t (*callback)(void *, u8, u8 *, s64), void *userdata);
+    void (*callback)(void *, u8, u8 *, void *), void *userdata);
 bool nn_http_request_init(struct nn_http_request *request);
 bool nn_http_request(struct nn_http_request *request, u8 method, struct nn_event_loop *loop,
     void (*callback)(void *, struct nn_http_request *, bool), void *userdata);