Adding body handler to nxt_http_request_header_send().

src/nxt_h1proto.c

@@ -45,7 +45,7 @@ static void nxt_h1p_conn_request_body_read(nxt_task_t *task, void *obj,
     void *data);
 static void nxt_h1p_request_local_addr(nxt_task_t *task, nxt_http_request_t *r);
 static void nxt_h1p_request_header_send(nxt_task_t *task,
-    nxt_http_request_t *r);
+    nxt_http_request_t *r, nxt_work_handler_t body_handler);
 static void nxt_h1p_request_send(nxt_task_t *task, nxt_http_request_t *r,
     nxt_buf_t *out);
 static nxt_buf_t *nxt_h1p_chunk_create(nxt_task_t *task, nxt_http_request_t *r,
@@ -996,7 +996,8 @@ static const nxt_str_t  nxt_http_server_error[] = {
 #define UNKNOWN_STATUS_LENGTH  nxt_length("HTTP/1.1 65536\r\n")
 
 static void
-nxt_h1p_request_header_send(nxt_task_t *task, nxt_http_request_t *r)
+nxt_h1p_request_header_send(nxt_task_t *task, nxt_http_request_t *r,
+    nxt_work_handler_t body_handler)
 {
     u_char              *p;
     size_t              size;
@@ -1079,6 +1080,7 @@ nxt_h1p_request_header_send(nxt_task_t *task, nxt_http_request_t *r)
             if (http11) {
                 if (n != NXT_HTTP_NOT_MODIFIED
                     && n != NXT_HTTP_NO_CONTENT
+                    && body_handler != NULL
                     && !h1p->websocket)
                 {
                     h1p->chunked = 1;
@@ -1165,6 +1167,19 @@ nxt_h1p_request_header_send(nxt_task_t *task, nxt_http_request_t *r)
     c->write = header;
     c->write_state = &nxt_h1p_request_send_state;
 
+    if (body_handler != NULL) {
+        /*
+         * The body handler will run before c->io->write() handler,
+         * because the latter was inqueued by nxt_conn_write()
+         * in engine->write_work_queue.
+         */
+        nxt_work_queue_add(&task->thread->engine->fast_work_queue,
+                           body_handler, task, r, NULL);
+
+    } else {
+        header->next = nxt_http_buf_last(r);
+    }
+
     nxt_conn_write(task->thread->engine, c);
 
     if (h1p->websocket) {

src/nxt_http.h

@@ -175,7 +175,8 @@ struct nxt_http_pass_s {
 typedef struct {
     void (*body_read)(nxt_task_t *task, nxt_http_request_t *r);
     void (*local_addr)(nxt_task_t *task, nxt_http_request_t *r);
-    void (*header_send)(nxt_task_t *task, nxt_http_request_t *r);
+    void (*header_send)(nxt_task_t *task, nxt_http_request_t *r,
+         nxt_work_handler_t body_handler);
     void (*send)(nxt_task_t *task, nxt_http_request_t *r, nxt_buf_t *out);
     nxt_off_t (*body_bytes_sent)(nxt_task_t *task, nxt_http_proto_t proto);
     void (*discard)(nxt_task_t *task, nxt_http_request_t *r, nxt_buf_t *last);
@@ -195,7 +196,8 @@ nxt_http_request_t *nxt_http_request_create(nxt_task_t *task);
 void nxt_http_request_error(nxt_task_t *task, nxt_http_request_t *r,
     nxt_http_status_t status);
 void nxt_http_request_read_body(nxt_task_t *task, nxt_http_request_t *r);
-void nxt_http_request_header_send(nxt_task_t *task, nxt_http_request_t *r);
+void nxt_http_request_header_send(nxt_task_t *task, nxt_http_request_t *r,
+    nxt_work_handler_t body_handler);
 void nxt_http_request_ws_frame_start(nxt_task_t *task, nxt_http_request_t *r,
     nxt_buf_t *ws_frame);
 void nxt_http_request_send(nxt_task_t *task, nxt_http_request_t *r,

src/nxt_http_error.c

@@ -51,12 +51,10 @@ nxt_http_request_error(nxt_task_t *task, nxt_http_request_t *r,
     r->resp.content_length = NULL;
     r->resp.content_length_n = nxt_length(error);
 
-    nxt_http_request_header_send(task, r);
-
     r->state = &nxt_http_request_send_error_body_state;
 
-    nxt_work_queue_add(&task->thread->engine->fast_work_queue,
+    nxt_http_request_header_send(task, r, nxt_http_request_send_error_body);
-                       nxt_http_request_send_error_body, task, r, NULL);
+
     return;
 
 fail:

src/nxt_http_request.c

@@ -369,7 +369,8 @@ nxt_http_request_read_body(nxt_task_t *task, nxt_http_request_t *r)
 
 
 void
-nxt_http_request_header_send(nxt_task_t *task, nxt_http_request_t *r)
+nxt_http_request_header_send(nxt_task_t *task, nxt_http_request_t *r,
+    nxt_work_handler_t body_handler)
 {
     u_char            *p, *end;
     nxt_http_field_t  *server, *date, *content_length;
@@ -430,7 +431,7 @@ nxt_http_request_header_send(nxt_task_t *task, nxt_http_request_t *r)
     }
 
     if (nxt_fast_path(r->proto.any != NULL)) {
-        nxt_http_proto[r->protocol].header_send(task, r);
+        nxt_http_proto[r->protocol].header_send(task, r, body_handler);
     }
 
     return;

src/nxt_router.c

@@ -3531,7 +3531,7 @@ nxt_router_response_ready_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg,
             nxt_buf_chain_add(&r->out, b);
         }
 
-        nxt_http_request_header_send(task, r);
+        nxt_http_request_header_send(task, r, nxt_http_request_send_body);
 
         if (r->websocket_handshake
             && r->status == NXT_HTTP_SWITCHING_PROTOCOLS)
@@ -3575,11 +3575,6 @@ nxt_router_response_ready_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg,
         } else {
             r->state = &nxt_http_request_send_state;
         }
-
-        if (r->out) {
-            nxt_work_queue_add(&task->thread->engine->fast_work_queue,
-                               nxt_http_request_send_body, task, r, NULL);
-        }
     }
 
     return;