HTTP keep-alive connections support.

1 files changed, 120 insertions, 640 deletions

diff --git a/src/nxt_router.c b/src/nxt_router.c
index 66ad6478..54e70a1a 100644
--- a/src/nxt_router.c
+++ b/src/nxt_router.c
@@ -7,6 +7,7 @@
 
 #include <nxt_router.h>
 #include <nxt_conf.h>
+#include <nxt_http.h>
 
 
 typedef struct {
@@ -35,15 +36,14 @@ typedef struct nxt_req_app_link_s nxt_req_app_link_t;
 
 
 typedef struct {
-    uint32_t             stream;
-    nxt_conn_t           *conn;
-    nxt_app_t            *app;
-    nxt_port_t           *app_port;
-    nxt_app_parse_ctx_t  *ap;
-    nxt_msg_info_t       msg_info;
-    nxt_req_app_link_t   *ra;
+    uint32_t                 stream;
+    nxt_app_t                *app;
+    nxt_port_t               *app_port;
+    nxt_app_parse_ctx_t      *ap;
+    nxt_msg_info_t           msg_info;
+    nxt_req_app_link_t       *ra;
 
-    nxt_queue_link_t     link;     /* for nxt_conn_t.requests */
+    nxt_queue_link_t         link;     /* for nxt_conn_t.requests */
 } nxt_req_conn_link_t;
 
 
@@ -199,14 +199,6 @@ static void nxt_router_app_port_release(nxt_task_t *task, nxt_port_t *port,
 static nxt_int_t nxt_router_app_port(nxt_task_t *task, nxt_app_t *app,
     nxt_req_app_link_t *ra);
 
-static void nxt_router_conn_init(nxt_task_t *task, void *obj, void *data);
-static void nxt_router_conn_http_header_parse(nxt_task_t *task, void *obj,
-    void *data);
-static nxt_sockaddr_t *nxt_router_local_addr(nxt_task_t *task, nxt_conn_t *c);
-static void nxt_router_conn_http_body_read(nxt_task_t *task, void *obj,
-    void *data);
-static void nxt_router_process_http_request(nxt_task_t *task,
-    nxt_conn_t *c, nxt_app_parse_ctx_t *ap);
 static void nxt_router_app_prepare_request(nxt_task_t *task,
     nxt_req_app_link_t *ra);
 static nxt_int_t nxt_python_prepare_msg(nxt_task_t *task, nxt_app_request_t *r,
@@ -215,16 +207,11 @@ static nxt_int_t nxt_php_prepare_msg(nxt_task_t *task, nxt_app_request_t *r,
     nxt_app_wmsg_t *wmsg);
 static nxt_int_t nxt_go_prepare_msg(nxt_task_t *task, nxt_app_request_t *r,
     nxt_app_wmsg_t *wmsg);
-static void nxt_router_conn_ready(nxt_task_t *task, void *obj, void *data);
-static void nxt_router_conn_close(nxt_task_t *task, void *obj, void *data);
 static void nxt_router_conn_free(nxt_task_t *task, void *obj, void *data);
-static void nxt_router_conn_error(nxt_task_t *task, void *obj, void *data);
-static void nxt_router_conn_timeout(nxt_task_t *task, void *obj, void *data);
 static void nxt_router_app_timeout(nxt_task_t *task, void *obj, void *data);
-static nxt_msec_t nxt_router_conn_timeout_value(nxt_conn_t *c, uintptr_t data);
 
-static void nxt_router_gen_error(nxt_task_t *task, nxt_conn_t *c, int code,
-    const char* str);
+static const nxt_http_request_state_t  nxt_http_request_send_state;
+static void nxt_http_request_send_body(nxt_task_t *task, void *obj, void *data);
 
 static nxt_router_t  *nxt_router;
 
@@ -245,7 +232,7 @@ nxt_router_start(nxt_task_t *task, void *data)
 
     rt = task->thread->runtime;
 
-    ret = nxt_app_http_init(task, rt);
+    ret = nxt_http_init(task, rt);
     if (nxt_slow_path(ret != NXT_OK)) {
         return ret;
     }
@@ -502,9 +489,8 @@ nxt_router_ra_update_peer(nxt_task_t *task, nxt_req_app_link_t *ra)
 static void
 nxt_router_ra_release(nxt_task_t *task, nxt_req_app_link_t *ra)
 {
-    nxt_mp_t             *mp;
-    nxt_conn_t           *c;
-    nxt_req_conn_link_t  *rc;
+    nxt_mp_t                *mp;
+    nxt_req_conn_link_t     *rc;
 
     nxt_assert(task->thread->engine == ra->work.data);
     nxt_assert(ra->use_count == 0);
@@ -514,18 +500,16 @@ nxt_router_ra_release(nxt_task_t *task, nxt_req_app_link_t *ra)
     rc = ra->rc;
 
     if (rc != NULL) {
-        c = rc->conn;
-
         if (nxt_slow_path(ra->err_code != 0)) {
-            nxt_router_gen_error(task, c, ra->err_code, ra->err_str);
+            nxt_http_request_error(task, rc->ap->request, ra->err_code);
 
         } else {
             rc->app_port = ra->app_port;
             rc->msg_info = ra->msg_info;
 
             if (rc->app->timeout != 0) {
-                c->read_timer.handler = nxt_router_app_timeout;
-                nxt_timer_add(task->thread->engine, &c->read_timer,
+                rc->ap->timer.handler = nxt_router_app_timeout;
+                nxt_timer_add(task->thread->engine, &rc->ap->timer,
                               rc->app->timeout);
             }
 
@@ -693,10 +677,6 @@ nxt_router_rc_unlink(nxt_task_t *task, nxt_req_conn_link_t *rc)
 
         rc->ap = NULL;
     }
-
-    nxt_queue_remove(&rc->link);
-
-    rc->conn = NULL;
 }
 
 
@@ -1079,6 +1059,12 @@ static nxt_conf_map_t  nxt_router_http_conf[] = {
     },
 
     {
+        nxt_string("idle_timeout"),
+        NXT_CONF_MAP_MSEC,
+        offsetof(nxt_socket_conf_t, idle_timeout),
+    },
+
+    {
         nxt_string("header_read_timeout"),
         NXT_CONF_MAP_MSEC,
         offsetof(nxt_socket_conf_t, header_read_timeout),
@@ -1089,6 +1075,12 @@ static nxt_conf_map_t  nxt_router_http_conf[] = {
         NXT_CONF_MAP_MSEC,
         offsetof(nxt_socket_conf_t, body_read_timeout),
     },
+
+    {
+        nxt_string("send_timeout"),
+        NXT_CONF_MAP_MSEC,
+        offsetof(nxt_socket_conf_t, send_timeout),
+    },
 };
 
 
@@ -1296,8 +1288,10 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
         skcf->large_header_buffers = 4;
         skcf->body_buffer_size = 16 * 1024;
         skcf->max_body_size = 2 * 1024 * 1024;
+        skcf->idle_timeout = 65000;
         skcf->header_read_timeout = 5000;
         skcf->body_read_timeout = 5000;
+        skcf->send_timeout = 5000;
 
         if (http != NULL) {
             ret = nxt_conf_map_object(mp, http, nxt_router_http_conf,
@@ -1308,7 +1302,7 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
             }
         }
 
-        skcf->listen->handler = nxt_router_conn_init;
+        skcf->listen->handler = nxt_http_conn_init;
         skcf->router_conf = tmcf->conf;
         skcf->router_conf->count++;
         skcf->application = nxt_router_listener_application(tmcf,
@@ -2377,92 +2371,20 @@ nxt_router_thread_exit_handler(nxt_task_t *task, void *obj, void *data)
 }
 
 
-static const nxt_conn_state_t  nxt_router_conn_read_header_state
-    nxt_aligned(64) =
-{
-    .ready_handler = nxt_router_conn_http_header_parse,
-    .close_handler = nxt_router_conn_close,
-    .error_handler = nxt_router_conn_error,
-
-    .timer_handler = nxt_router_conn_timeout,
-    .timer_value = nxt_router_conn_timeout_value,
-    .timer_data = offsetof(nxt_socket_conf_t, header_read_timeout),
-};
-
-
-static const nxt_conn_state_t  nxt_router_conn_read_body_state
-    nxt_aligned(64) =
-{
-    .ready_handler = nxt_router_conn_http_body_read,
-    .close_handler = nxt_router_conn_close,
-    .error_handler = nxt_router_conn_error,
-
-    .timer_handler = nxt_router_conn_timeout,
-    .timer_value = nxt_router_conn_timeout_value,
-    .timer_data = offsetof(nxt_socket_conf_t, body_read_timeout),
-    .timer_autoreset = 1,
-};
-
-
-static void
-nxt_router_conn_init(nxt_task_t *task, void *obj, void *data)
-{
-    size_t                   size;
-    nxt_conn_t               *c;
-    nxt_socket_conf_t        *skcf;
-    nxt_event_engine_t       *engine;
-    nxt_socket_conf_joint_t  *joint;
-
-    c = obj;
-    joint = data;
-
-    nxt_debug(task, "router conn init");
-
-    c->joint = joint;
-    joint->count++;
-
-    skcf = joint->socket_conf;
-    c->local = skcf->sockaddr;
-
-    size = skcf->header_buffer_size;
-    c->read = nxt_buf_mem_alloc(c->mem_pool, size, 0);
-
-    c->socket.data = NULL;
-
-    engine = task->thread->engine;
-    c->read_work_queue = &engine->fast_work_queue;
-    c->write_work_queue = &engine->fast_work_queue;
-
-    c->read_state = &nxt_router_conn_read_header_state;
-
-    nxt_conn_read(engine, c);
-}
-
-
-static const nxt_conn_state_t  nxt_router_conn_write_state
-    nxt_aligned(64) =
-{
-    .ready_handler = nxt_router_conn_ready,
-    .close_handler = nxt_router_conn_close,
-    .error_handler = nxt_router_conn_error,
-};
-
-
 static void
 nxt_router_response_ready_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg,
     void *data)
 {
     size_t               dump_size;
+    nxt_int_t            ret;
     nxt_buf_t            *b, *last;
-    nxt_conn_t           *c;
-    nxt_event_engine_t   *engine;
+    nxt_http_request_t   *r;
     nxt_req_conn_link_t  *rc;
+    nxt_app_parse_ctx_t  *ar;
 
     b = msg->buf;
     rc = data;
 
-    c = rc->conn;
-
     dump_size = nxt_buf_used_size(b);
 
     if (dump_size > 300) {
@@ -2477,16 +2399,16 @@ nxt_router_response_ready_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg,
         b = NULL;
     }
 
-    engine = task->thread->engine;
-
-    nxt_timer_disable(engine, &c->read_timer);
+    ar = rc->ap;
 
     if (msg->port_msg.last != 0) {
         nxt_debug(task, "router data create last buf");
 
-        last = nxt_buf_sync_alloc(c->mem_pool, NXT_BUF_SYNC_LAST);
+        last = nxt_http_request_last_buffer(task, ar->request);
         if (nxt_slow_path(last == NULL)) {
-            /* TODO pogorevaTb */
+            nxt_app_http_req_done(task, ar);
+            nxt_router_rc_unlink(task, rc);
+            return;
         }
 
         nxt_buf_chain_add(&b, last);
@@ -2495,8 +2417,8 @@ nxt_router_response_ready_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg,
 
     } else {
         if (rc->app->timeout != 0) {
-            c->read_timer.handler = nxt_router_app_timeout;
-            nxt_timer_add(engine, &c->read_timer, rc->app->timeout);
+            ar->timer.handler = nxt_router_app_timeout;
+            nxt_timer_add(task->thread->engine, &ar->timer, rc->app->timeout);
         }
     }
 
@@ -2509,16 +2431,67 @@ nxt_router_response_ready_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg,
         msg->buf = NULL;
     }
 
-    if (c->write == NULL) {
-        c->write = b;
-        c->write_state = &nxt_router_conn_write_state;
+    r = ar->request;
 
-        nxt_conn_write(task->thread->engine, c);
+    if (r->header_sent) {
+        nxt_buf_chain_add(&r->out, b);
+        nxt_http_request_send_body(task, r, NULL);
 
     } else {
-        nxt_debug(task, "router data attach out bufs to existing chain");
+        ret = nxt_http_parse_fields(&ar->resp_parser, &b->mem);
+        if (nxt_slow_path(ret != NXT_DONE)) {
+            goto fail;
+        }
+
+        r->resp.fields = ar->resp_parser.fields;
 
-        nxt_buf_chain_add(&c->write, b);
+        ret = nxt_http_fields_process(r->resp.fields,
+                                      &nxt_response_fields_hash, r);
+        if (nxt_slow_path(ret != NXT_OK)) {
+            goto fail;
+        }
+
+        if (nxt_buf_mem_used_size(&b->mem) != 0) {
+            nxt_buf_chain_add(&r->out, b);
+        }
+
+        r->state = &nxt_http_request_send_state;
+
+        nxt_http_request_header_send(task, r);
+    }
+
+    return;
+
+fail:
+
+    nxt_app_http_req_done(task, ar);
+    nxt_router_rc_unlink(task, rc);
+
+    nxt_http_request_error(task, r, NXT_HTTP_SERVICE_UNAVAILABLE);
+}
+
+
+static const nxt_http_request_state_t  nxt_http_request_send_state
+    nxt_aligned(64) =
+{
+    .ready_handler = nxt_http_request_send_body,
+    .error_handler = nxt_http_request_close_handler,
+};
+
+
+static void
+nxt_http_request_send_body(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_buf_t           *out;
+    nxt_http_request_t  *r;
+
+    r = obj;
+
+    out = r->out;
+
+    if (out != NULL) {
+        r->out = NULL;
+        nxt_http_request_send(task, r, out);
     }
 }
 
@@ -2562,98 +2535,12 @@ nxt_router_response_error_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg,
         }
     }
 
-    nxt_router_gen_error(task, rc->conn, 500,
-                         "Application terminated unexpectedly");
+    nxt_http_request_error(task, rc->ap->request, NXT_HTTP_SERVICE_UNAVAILABLE);
 
     nxt_router_rc_unlink(task, rc);
 }
 
 
-nxt_inline const char *
-nxt_router_text_by_code(int code)
-{
-    switch (code) {
-    case 400: return "Bad request";
-    case 404: return "Not found";
-    case 403: return "Forbidden";
-    case 408: return "Request Timeout";
-    case 411: return "Length Required";
-    case 413: return "Request Entity Too Large";
-    case 500:
-    default:  return "Internal server error";
-    }
-}
-
-
-static nxt_buf_t *
-nxt_router_get_error_buf(nxt_task_t *task, nxt_mp_t *mp, int code,
-    const char* str)
-{
-    nxt_buf_t  *b, *last;
-
-    b = nxt_buf_mem_alloc(mp, 16384, 0);
-    if (nxt_slow_path(b == NULL)) {
-        return NULL;
-    }
-
-    b->mem.free = nxt_sprintf(b->mem.free, b->mem.end,
-        "HTTP/1.0 %d %s\r\n"
-        "Content-Type: text/plain\r\n"
-        "Connection: close\r\n\r\n",
-        code, nxt_router_text_by_code(code));
-
-    b->mem.free = nxt_cpymem(b->mem.free, str, nxt_strlen(str));
-
-    last = nxt_buf_sync_alloc(mp, NXT_BUF_SYNC_LAST);
-
-    if (nxt_slow_path(last == NULL)) {
-        nxt_mp_free(mp, b);
-        return NULL;
-    }
-
-    nxt_buf_chain_add(&b, last);
-
-    return b;
-}
-
-
-
-static void
-nxt_router_gen_error(nxt_task_t *task, nxt_conn_t *c, int code,
-    const char* str)
-{
-    nxt_mp_t   *mp;
-    nxt_buf_t  *b;
-
-    /* TODO: fix when called in the middle of response */
-
-    nxt_log_alert(task->log, "error %d: %s", code, str);
-
-    if (c->socket.fd == -1) {
-        return;
-    }
-
-    mp = c->mem_pool;
-
-    b = nxt_router_get_error_buf(task, mp, code, str);
-    if (nxt_slow_path(b == NULL)) {
-        return;
-    }
-
-    if (c->write == NULL) {
-        c->write = b;
-        c->write_state = &nxt_router_conn_write_state;
-
-        nxt_conn_write(task->thread->engine, c);
-
-    } else {
-        nxt_debug(task, "router data attach out bufs to existing chain");
-
-        nxt_buf_chain_add(&c->write, b);
-    }
-}
-
-
 static void
 nxt_router_app_port_ready(nxt_task_t *task, nxt_port_recv_msg_t *msg,
     void *data)
@@ -3228,283 +3115,22 @@ nxt_router_app_port(nxt_task_t *task, nxt_app_t *app, nxt_req_app_link_t *ra)
 }
 
 
-static void
-nxt_router_conn_http_header_parse(nxt_task_t *task, void *obj, void *data)
-{
-    size_t                    size;
-    nxt_int_t                 ret;
-    nxt_buf_t                 *buf;
-    nxt_conn_t                *c;
-    nxt_sockaddr_t            *local;
-    nxt_app_parse_ctx_t       *ap;
-    nxt_app_request_body_t    *b;
-    nxt_socket_conf_joint_t   *joint;
-    nxt_app_request_header_t  *h;
-
-    c = obj;
-    ap = data;
-    buf = c->read;
-    joint = c->joint;
-
-    nxt_debug(task, "router conn http header parse");
-
-    if (ap == NULL) {
-        ap = nxt_app_http_req_init(task);
-        if (nxt_slow_path(ap == NULL)) {
-            nxt_router_gen_error(task, c, 500,
-                                 "Failed to allocate parse context");
-            return;
-        }
-
-        c->socket.data = ap;
-
-        ap->r.remote.start = nxt_sockaddr_address(c->remote);
-        ap->r.remote.length = c->remote->address_length;
-
-        /*
-         * TODO: need an application flag to get local address
-         * required by "SERVER_ADDR" in Pyhton and PHP. Not used in Go.
-         */
-        local = nxt_router_local_addr(task, c);
-
-        if (nxt_fast_path(local != NULL)) {
-            ap->r.local.start = nxt_sockaddr_address(local);
-            ap->r.local.length = local->address_length;
-        }
-
-        ap->r.header.buf = buf;
-    }
-
-    h = &ap->r.header;
-    b = &ap->r.body;
-
-    ret = nxt_app_http_req_header_parse(task, ap, buf);
-
-    nxt_debug(task, "http parse request header: %d", ret);
-
-    switch (nxt_expect(NXT_DONE, ret)) {
-
-    case NXT_DONE:
-        nxt_debug(task, "router request header parsing complete, "
-                  "content length: %O, preread: %uz",
-                  h->parsed_content_length, nxt_buf_mem_used_size(&buf->mem));
-
-        if (b->done) {
-            nxt_router_process_http_request(task, c, ap);
-
-            return;
-        }
-
-        if (joint->socket_conf->max_body_size > 0
-            && (size_t) h->parsed_content_length
-               > joint->socket_conf->max_body_size)
-        {
-            nxt_router_gen_error(task, c, 413, "Content-Length too big");
-            return;
-        }
-
-        if (nxt_buf_mem_free_size(&buf->mem) == 0) {
-            size = nxt_min(joint->socket_conf->body_buffer_size,
-                           (size_t) h->parsed_content_length);
-
-            buf->next = nxt_buf_mem_alloc(c->mem_pool, size, 0);
-            if (nxt_slow_path(buf->next == NULL)) {
-                nxt_router_gen_error(task, c, 500, "Failed to allocate "
-                                     "buffer for request body");
-                return;
-            }
-
-            c->read = buf->next;
-
-            b->preread_size += nxt_buf_mem_used_size(&buf->mem);
-        }
-
-        if (b->buf == NULL) {
-            b->buf = c->read;
-        }
-
-        c->read_state = &nxt_router_conn_read_body_state;
-        break;
-
-    case NXT_ERROR:
-        nxt_router_gen_error(task, c, 400, "Request header parse error");
-        return;
-
-    default:  /* NXT_AGAIN */
-
-        if (c->read->mem.free == c->read->mem.end) {
-            size = joint->socket_conf->large_header_buffer_size;
-
-            if (size <= (size_t) nxt_buf_mem_used_size(&buf->mem)
-                || ap->r.header.bufs
-                   >= joint->socket_conf->large_header_buffers)
-            {
-                nxt_router_gen_error(task, c, 413,
-                                     "Too long request headers");
-                return;
-            }
-
-            buf->next = nxt_buf_mem_alloc(c->mem_pool, size, 0);
-            if (nxt_slow_path(buf->next == NULL)) {
-                nxt_router_gen_error(task, c, 500,
-                                     "Failed to allocate large header "
-                                     "buffer");
-                return;
-            }
-
-            ap->r.header.bufs++;
-
-            size = c->read->mem.free - c->read->mem.pos;
-
-            c->read = nxt_buf_cpy(buf->next, c->read->mem.pos, size);
-        }
-
-    }
-
-    nxt_conn_read(task->thread->engine, c);
-}
-
-
-static nxt_sockaddr_t *
-nxt_router_local_addr(nxt_task_t *task, nxt_conn_t *c)
-{
-    int             ret;
-    size_t          size, length;
-    socklen_t       socklen;
-    nxt_sockaddr_t  *sa;
-
-    if (c->local != NULL) {
-        return c->local;
-    }
-
-    /* AF_UNIX should not get in here. */
-
-    switch (c->remote->u.sockaddr.sa_family) {
-#if (NXT_INET6)
-    case AF_INET6:
-        socklen = sizeof(struct sockaddr_in6);
-        length = NXT_INET6_ADDR_STR_LEN;
-        size = offsetof(nxt_sockaddr_t, u) + socklen + length;
-        break;
-#endif
-    case AF_INET:
-    default:
-        socklen = sizeof(struct sockaddr_in);
-        length = NXT_INET_ADDR_STR_LEN;
-        size = offsetof(nxt_sockaddr_t, u) + socklen + length;
-        break;
-    }
-
-    sa = nxt_mp_get(c->mem_pool, size);
-    if (nxt_slow_path(sa == NULL)) {
-        return NULL;
-    }
-
-    sa->socklen = socklen;
-    sa->length = length;
-
-    ret = getsockname(c->socket.fd, &sa->u.sockaddr, &socklen);
-    if (nxt_slow_path(ret != 0)) {
-        nxt_log(task, NXT_LOG_CRIT, "getsockname(%d) failed", c->socket.fd);
-        return NULL;
-    }
-
-    c->local = sa;
-
-    nxt_sockaddr_text(sa);
-
-    /*
-     * TODO: here we can adjust the end of non-freeable block
-     * in c->mem_pool to the end of actual sockaddr length.
-     */
-
-    return sa;
-}
-
-
-static void
-nxt_router_conn_http_body_read(nxt_task_t *task, void *obj, void *data)
-{
-    size_t                    size;
-    nxt_int_t                 ret;
-    nxt_buf_t                 *buf;
-    nxt_conn_t                *c;
-    nxt_app_parse_ctx_t       *ap;
-    nxt_app_request_body_t    *b;
-    nxt_socket_conf_joint_t   *joint;
-    nxt_app_request_header_t  *h;
-
-    c = obj;
-    ap = data;
-    buf = c->read;
-
-    nxt_debug(task, "router conn http body read");
-
-    nxt_assert(ap != NULL);
-
-    b = &ap->r.body;
-    h = &ap->r.header;
-
-    ret = nxt_app_http_req_body_read(task, ap, buf);
-
-    nxt_debug(task, "http read request body: %d", ret);
-
-    switch (nxt_expect(NXT_DONE, ret)) {
-
-    case NXT_DONE:
-        nxt_router_process_http_request(task, c, ap);
-        return;
-
-    case NXT_ERROR:
-        nxt_router_gen_error(task, c, 500, "Read body error");
-        return;
-
-    default:  /* NXT_AGAIN */
-
-        if (nxt_buf_mem_free_size(&buf->mem) == 0) {
-            joint = c->joint;
-
-            b->preread_size += nxt_buf_mem_used_size(&buf->mem);
-
-            size = nxt_min(joint->socket_conf->body_buffer_size,
-                           (size_t) h->parsed_content_length - b->preread_size);
-
-            buf->next = nxt_buf_mem_alloc(c->mem_pool, size, 0);
-            if (nxt_slow_path(buf->next == NULL)) {
-                nxt_router_gen_error(task, c, 500, "Failed to allocate "
-                                     "buffer for request body");
-                return;
-            }
-
-            c->read = buf->next;
-        }
-
-        nxt_debug(task, "router request body read again, rest: %uz",
-                  h->parsed_content_length - b->preread_size);
-    }
-
-    nxt_conn_read(task->thread->engine, c);
-}
-
-
-static void
-nxt_router_process_http_request(nxt_task_t *task, nxt_conn_t *c,
-    nxt_app_parse_ctx_t *ap)
+void
+nxt_router_process_http_request(nxt_task_t *task, nxt_app_parse_ctx_t *ar)
 {
-    nxt_int_t                res;
-    nxt_app_t                *app;
-    nxt_port_t               *port;
-    nxt_event_engine_t       *engine;
-    nxt_req_app_link_t       ra_local, *ra;
-    nxt_req_conn_link_t      *rc;
-    nxt_socket_conf_joint_t  *joint;
+    nxt_int_t            res;
+    nxt_app_t            *app;
+    nxt_port_t           *port;
+    nxt_event_engine_t   *engine;
+    nxt_http_request_t   *r;
+    nxt_req_app_link_t   ra_local, *ra;
+    nxt_req_conn_link_t  *rc;
 
-    joint = c->joint;
-    app = joint->socket_conf->application;
+    r = ar->request;
+    app = r->socket_conf->application;
 
     if (app == NULL) {
-        nxt_router_gen_error(task, c, 500,
-                             "Application is NULL in socket_conf");
+        nxt_http_request_error(task, r, NXT_HTTP_INTERNAL_SERVER_ERROR);
         return;
     }
 
@@ -3516,27 +3142,16 @@ nxt_router_process_http_request(nxt_task_t *task, nxt_conn_t *c,
                                           sizeof(nxt_req_conn_link_t));
 
     if (nxt_slow_path(rc == NULL)) {
-        nxt_router_gen_error(task, c, 500, "Failed to allocate "
-                             "req<->conn link");
-
+        nxt_http_request_error(task, r, NXT_HTTP_INTERNAL_SERVER_ERROR);
         return;
     }
 
     rc->stream = nxt_port_rpc_ex_stream(rc);
-    rc->conn = c;
     rc->app = app;
 
     nxt_router_app_use(task, app, 1);
 
-    nxt_timer_disable(engine, &c->read_timer);
-
-    nxt_queue_insert_tail(&c->requests, &rc->link);
-
-    nxt_debug(task, "stream #%uD linked to conn %p at engine %p",
-              rc->stream, c, engine);
-
-    rc->ap = ap;
-    c->socket.data = NULL;
+    rc->ap = ar;
 
     ra = &ra_local;
     nxt_router_ra_init(task, ra, rc);
@@ -3912,7 +3527,7 @@ fail:
 }
 
 
-static const nxt_conn_state_t  nxt_router_conn_close_state
+const nxt_conn_state_t  nxt_router_conn_close_state
     nxt_aligned(64) =
 {
     .ready_handler = nxt_router_conn_free,
@@ -3920,75 +3535,6 @@ static const nxt_conn_state_t  nxt_router_conn_close_state
 
 
 static void
-nxt_router_conn_ready(nxt_task_t *task, void *obj, void *data)
-{
-    nxt_buf_t         *b;
-    nxt_bool_t        last;
-    nxt_conn_t        *c;
-    nxt_work_queue_t  *wq;
-
-    nxt_debug(task, "router conn ready %p", obj);
-
-    c = obj;
-    b = c->write;
-
-    wq = &task->thread->engine->fast_work_queue;
-
-    last = 0;
-
-    while (b != NULL) {
-        if (!nxt_buf_is_sync(b)) {
-            if (nxt_buf_used_size(b) > 0) {
-                break;
-            }
-        }
-
-        if (nxt_buf_is_last(b)) {
-            last = 1;
-        }
-
-        nxt_work_queue_add(wq, b->completion_handler, task, b, b->parent);
-
-        b = b->next;
-    }
-
-    c->write = b;
-
-    if (b != NULL) {
-        nxt_debug(task, "router conn %p has more data to write", obj);
-
-        nxt_conn_write(task->thread->engine, c);
-
-    } else {
-        nxt_debug(task, "router conn %p no more data to write, last = %d", obj,
-                  last);
-
-        if (last != 0) {
-            nxt_debug(task, "enqueue router conn close %p (ready handler)", c);
-
-            nxt_work_queue_add(wq, nxt_router_conn_close, task, c,
-                               c->socket.data);
-        }
-    }
-}
-
-
-static void
-nxt_router_conn_close(nxt_task_t *task, void *obj, void *data)
-{
-    nxt_conn_t  *c;
-
-    c = obj;
-
-    nxt_debug(task, "router conn close");
-
-    c->write_state = &nxt_router_conn_close_state;
-
-    nxt_conn_close(task->thread->engine, c);
-}
-
-
-static void
 nxt_router_conn_mp_cleanup(nxt_task_t *task, void *obj, void *data)
 {
     nxt_socket_conf_joint_t  *joint;
@@ -4004,31 +3550,12 @@ nxt_router_conn_free(nxt_task_t *task, void *obj, void *data)
 {
     nxt_conn_t               *c;
     nxt_event_engine_t       *engine;
-    nxt_req_conn_link_t      *rc;
-    nxt_app_parse_ctx_t      *ap;
     nxt_socket_conf_joint_t  *joint;
 
     c = obj;
-    ap = data;
 
     nxt_debug(task, "router conn close done");
 
-    if (ap != NULL) {
-        nxt_app_http_req_done(task, ap);
-
-        c->socket.data = NULL;
-    }
-
-    nxt_queue_each(rc, &c->requests, nxt_req_conn_link_t, link) {
-
-        nxt_debug(task, "conn %p close, stream #%uD", c, rc->stream);
-
-        nxt_router_rc_unlink(task, rc);
-
-        nxt_port_rpc_cancel(task, task->thread->engine->port, rc->stream);
-
-    } nxt_queue_loop;
-
     nxt_queue_remove(&c->link);
 
     engine = task->thread->engine;
@@ -4045,65 +3572,18 @@ nxt_router_conn_free(nxt_task_t *task, void *obj, void *data)
 
 
 static void
-nxt_router_conn_error(nxt_task_t *task, void *obj, void *data)
-{
-    nxt_conn_t  *c;
-
-    c = obj;
-
-    nxt_debug(task, "router conn error");
-
-    if (c->socket.fd != -1) {
-        c->write_state = &nxt_router_conn_close_state;
-
-        nxt_conn_close(task->thread->engine, c);
-    }
-}
-
-
-static void
-nxt_router_conn_timeout(nxt_task_t *task, void *obj, void *data)
-{
-    nxt_conn_t   *c;
-    nxt_timer_t  *timer;
-
-    timer = obj;
-
-    nxt_debug(task, "router conn timeout");
-
-    c = nxt_read_timer_conn(timer);
-
-    if (c->read_state == &nxt_router_conn_read_header_state) {
-        nxt_router_gen_error(task, c, 408, "Read header timeout");
-
-    } else {
-        nxt_router_gen_error(task, c, 408, "Read body timeout");
-    }
-}
-
-
-static void
 nxt_router_app_timeout(nxt_task_t *task, void *obj, void *data)
 {
-    nxt_conn_t   *c;
-    nxt_timer_t  *timer;
+    nxt_timer_t          *timer;
+    nxt_app_parse_ctx_t  *ar;
 
     timer = obj;
 
     nxt_debug(task, "router app timeout");
 
-    c = nxt_read_timer_conn(timer);
-
-    nxt_router_gen_error(task, c, 408, "Application timeout");
-}
-
+    ar = nxt_timer_data(timer, nxt_app_parse_ctx_t, timer);
 
-static nxt_msec_t
-nxt_router_conn_timeout_value(nxt_conn_t *c, uintptr_t data)
-{
-    nxt_socket_conf_joint_t  *joint;
-
-    joint = c->joint;
-
-    return nxt_value_at(nxt_msec_t, joint->socket_conf, data);
+    if (!ar->request->header_sent) {
+        nxt_http_request_error(task, ar->request, NXT_HTTP_SERVICE_UNAVAILABLE);
+    }
 }