HTTP keep-alive connections support.

17 files changed, 1993 insertions, 891 deletions

diff --git a/auto/sources b/auto/sources
index 48052d9a..51dc0ba7 100644
--- a/auto/sources
+++ b/auto/sources
@@ -80,6 +80,10 @@ NXT_LIB_SRCS=" \
     src/nxt_worker_process.c \
     src/nxt_controller.c \
     src/nxt_router.c \
+    src/nxt_h1proto.c \
+    src/nxt_http_request.c \
+    src/nxt_http_response.c \
+    src/nxt_http_error.c \
     src/nxt_application.c \
     src/nxt_go.c \
     src/nxt_port_hash.c \
diff --git a/src/go/unit/response.go b/src/go/unit/response.go
index 258a82c9..44694011 100644
--- a/src/go/unit/response.go
+++ b/src/go/unit/response.go
@@ -54,7 +54,7 @@ func (r *response) WriteHeader(code int) {
 		return
 	}
 	r.headerSent = true
-	fmt.Fprintf(r, "%s %d %s\r\n", r.req.Proto, code, http.StatusText(code))
+	fmt.Fprintf(r, "Status: %d\r\n", code)
 
 	// Set a default Content-Type
 	if _, hasType := r.header["Content-Type"]; !hasType {
diff --git a/src/nxt_application.c b/src/nxt_application.c
index 29b7a1e6..e77a9796 100644
--- a/src/nxt_application.c
+++ b/src/nxt_application.c
@@ -10,6 +10,8 @@
 #include <nxt_runtime.h>
 #include <nxt_application.h>
 #include <nxt_main_process.h>
+#include <nxt_router.h>
+#include <nxt_http.h>
 
 #include <glob.h>
 
@@ -28,22 +30,8 @@ static nxt_app_module_t *nxt_app_module_load(nxt_task_t *task,
     const char *name);
 static nxt_app_type_t nxt_app_parse_type(u_char *p, size_t length);
 
-static nxt_int_t nxt_app_request_content_length(void *ctx,
-    nxt_http_field_t *field, uintptr_t data);
-static nxt_int_t nxt_app_request_content_type(void *ctx,
-    nxt_http_field_t *field, uintptr_t data);
-static nxt_int_t nxt_app_request_cookie(void *ctx, nxt_http_field_t *field,
-    uintptr_t data);
-static nxt_int_t nxt_app_request_host(void *ctx, nxt_http_field_t *field,
-    uintptr_t data);
-
-
-static nxt_http_field_proc_t  nxt_app_request_fields[] = {
-    { nxt_string("Content-Length"), &nxt_app_request_content_length, 0 },
-    { nxt_string("Content-Type"), &nxt_app_request_content_type, 0 },
-    { nxt_string("Cookie"), &nxt_app_request_cookie, 0 },
-    { nxt_string("Host"), &nxt_app_request_host, 0 },
-};
+
+static void nxt_app_http_release(nxt_task_t *task, void *obj, void *data);
 
 
 static uint32_t  compat[] = {
@@ -51,8 +39,6 @@ static uint32_t  compat[] = {
 };
 
 
-static nxt_lvlhsh_t              nxt_app_request_fields_hash;
-
 static nxt_thread_mutex_t        nxt_app_mutex;
 static nxt_thread_cond_t         nxt_app_cond;
 
@@ -372,15 +358,6 @@ nxt_app_module_load(nxt_task_t *task, const char *name)
 }
 
 
-nxt_int_t
-nxt_app_http_init(nxt_task_t *task, nxt_runtime_t *rt)
-{
-    return nxt_http_fields_hash(&nxt_app_request_fields_hash, rt->mem_pool,
-                                nxt_app_request_fields,
-                                nxt_nitems(nxt_app_request_fields));
-}
-
-
 void
 nxt_app_quit_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
 {
@@ -704,197 +681,29 @@ nxt_app_msg_read_size(nxt_task_t *task, nxt_app_rmsg_t *msg, size_t *size)
 }
 
 
-static nxt_int_t
-nxt_app_request_content_length(void *ctx, nxt_http_field_t *field,
-    uintptr_t data)
-{
-    nxt_app_parse_ctx_t       *c;
-    nxt_app_request_header_t  *h;
-
-    c = ctx;
-    h = &c->r.header;
-
-    h->content_length.length = field->value_length;
-    h->content_length.start = field->value;
-
-    h->parsed_content_length = nxt_off_t_parse(field->value,
-                                               field->value_length);
-
-    return NXT_OK;
-}
-
-
-static nxt_int_t
-nxt_app_request_content_type(void *ctx, nxt_http_field_t *field,
-    uintptr_t data)
-{
-    nxt_app_parse_ctx_t       *c;
-    nxt_app_request_header_t  *h;
-
-    c = ctx;
-    h = &c->r.header;
-
-    h->content_type.length = field->value_length;
-    h->content_type.start = field->value;
-
-    return NXT_OK;
-}
-
-
-static nxt_int_t
-nxt_app_request_cookie(void *ctx, nxt_http_field_t *field, uintptr_t data)
-{
-    nxt_app_parse_ctx_t       *c;
-    nxt_app_request_header_t  *h;
-
-    c = ctx;
-    h = &c->r.header;
-
-    h->cookie.length = field->value_length;
-    h->cookie.start = field->value;
-
-    return NXT_OK;
-}
-
-
-static nxt_int_t
-nxt_app_request_host(void *ctx, nxt_http_field_t *field, uintptr_t data)
-{
-    nxt_app_parse_ctx_t       *c;
-    nxt_app_request_header_t  *h;
-
-    c = ctx;
-    h = &c->r.header;
-
-    h->host.length = field->value_length;
-    h->host.start = field->value;
-
-    return NXT_OK;
-}
-
-
-nxt_app_parse_ctx_t *
-nxt_app_http_req_init(nxt_task_t *task)
-{
-    nxt_mp_t             *mp;
-    nxt_int_t            rc;
-    nxt_app_parse_ctx_t  *ctx;
-
-    mp = nxt_mp_create(1024, 128, 256, 32);
-    if (nxt_slow_path(mp == NULL)) {
-        return NULL;
-    }
-
-    ctx = nxt_mp_zget(mp, sizeof(nxt_app_parse_ctx_t));
-    if (nxt_slow_path(ctx == NULL)) {
-        nxt_mp_destroy(mp);
-        return NULL;
-    }
-
-    ctx->mem_pool = mp;
-
-    rc = nxt_http_parse_request_init(&ctx->parser, mp);
-    if (nxt_slow_path(rc != NXT_OK)) {
-        nxt_mp_destroy(mp);
-        return NULL;
-    }
-
-    return ctx;
-}
-
-
 nxt_int_t
-nxt_app_http_req_header_parse(nxt_task_t *task, nxt_app_parse_ctx_t *ctx,
-    nxt_buf_t *buf)
+nxt_app_http_req_done(nxt_task_t *task, nxt_app_parse_ctx_t *ar)
 {
-    nxt_int_t                 rc;
-    nxt_app_request_body_t    *b;
-    nxt_http_request_parse_t  *p;
-    nxt_app_request_header_t  *h;
-
-    p = &ctx->parser;
-    b = &ctx->r.body;
-    h = &ctx->r.header;
-
-    nxt_assert(h->done == 0);
-
-    rc = nxt_http_parse_request(p, &buf->mem);
-
-    if (nxt_slow_path(rc != NXT_DONE)) {
-        return rc;
-    }
-
-    rc = nxt_http_fields_process(p->fields, &nxt_app_request_fields_hash, ctx);
-
-    if (nxt_slow_path(rc != NXT_OK)) {
-        return rc;
-    }
-
-    h->fields = p->fields;
-    h->done = 1;
+    ar->timer.handler = nxt_app_http_release;
+    nxt_timer_add(task->thread->engine, &ar->timer, 0);
 
-    h->version.start = p->version.str;
-    h->version.length = sizeof(p->version.str);
-
-    h->method = p->method;
-
-    h->target.start = p->target_start;
-    h->target.length = p->target_end - p->target_start;
-
-    h->path = p->path;
-    h->query = p->args;
-
-    if (h->parsed_content_length == 0) {
-        b->done = 1;
-
-    }
-
-    if (buf->mem.free == buf->mem.pos) {
-        return NXT_DONE;
-    }
-
-    b->buf = buf;
-    b->done = nxt_buf_mem_used_size(&buf->mem) >=
-              h->parsed_content_length;
-
-    if (b->done == 1) {
-        b->preread_size = nxt_buf_mem_used_size(&buf->mem);
-    }
-
-    return NXT_DONE;
+    return NXT_OK;
 }
 
 
-nxt_int_t
-nxt_app_http_req_body_read(nxt_task_t *task, nxt_app_parse_ctx_t *ctx,
-    nxt_buf_t *buf)
+static void
+nxt_app_http_release(nxt_task_t *task, void *obj, void *data)
 {
-    nxt_app_request_body_t    *b;
-    nxt_app_request_header_t  *h;
-
-    b = &ctx->r.body;
-    h = &ctx->r.header;
-
-    nxt_assert(h->done == 1);
-    nxt_assert(b->done == 0);
-
-    b->done = nxt_buf_mem_used_size(&buf->mem) + b->preread_size >=
-              (size_t) h->parsed_content_length;
+    nxt_timer_t          *timer;
+    nxt_app_parse_ctx_t  *ar;
 
-    if (b->done == 1) {
-        b->preread_size += nxt_buf_mem_used_size(&buf->mem);
-    }
-
-    return b->done == 1 ? NXT_DONE : NXT_AGAIN;
-}
+    timer = obj;
 
+    nxt_debug(task, "http app release");
 
-nxt_int_t
-nxt_app_http_req_done(nxt_task_t *task, nxt_app_parse_ctx_t *ctx)
-{
-    nxt_mp_release(ctx->mem_pool);
+    ar = nxt_timer_data(timer, nxt_app_parse_ctx_t, timer);
 
-    return NXT_OK;
+    nxt_mp_release(ar->request->mem_pool);
 }
 
 
diff --git a/src/nxt_application.h b/src/nxt_application.h
index b42d77ff..a9a1d5d3 100644
--- a/src/nxt_application.h
+++ b/src/nxt_application.h
@@ -115,28 +115,22 @@ typedef struct {
 } nxt_app_request_t;
 
 
-typedef struct nxt_app_parse_ctx_s nxt_app_parse_ctx_t;
+typedef struct nxt_http_request_s   nxt_http_request_t;
+
+typedef struct nxt_app_parse_ctx_s  nxt_app_parse_ctx_t;
 
 struct nxt_app_parse_ctx_s {
     nxt_app_request_t         r;
+    nxt_http_request_t        *request;
+    nxt_timer_t               timer;
     nxt_http_request_parse_t  parser;
+    nxt_http_request_parse_t  resp_parser;
     nxt_mp_t                  *mem_pool;
 };
 
 
-nxt_app_parse_ctx_t *nxt_app_http_req_init(nxt_task_t *task);
-
-nxt_int_t nxt_app_http_req_header_parse(nxt_task_t *task,
-    nxt_app_parse_ctx_t *ctx, nxt_buf_t *buf);
-
-nxt_int_t nxt_app_http_req_body_read(nxt_task_t *task,
-    nxt_app_parse_ctx_t *ctx, nxt_buf_t *buf);
-
-
 nxt_int_t nxt_app_http_req_done(nxt_task_t *task, nxt_app_parse_ctx_t *ctx);
 
-nxt_int_t nxt_app_http_init(nxt_task_t *task, nxt_runtime_t *rt);
-
 
 typedef struct nxt_app_wmsg_s  nxt_app_wmsg_t;
 typedef struct nxt_app_rmsg_s  nxt_app_rmsg_t;
diff --git a/src/nxt_conn.c b/src/nxt_conn.c
index 854181ca..6d0182e3 100644
--- a/src/nxt_conn.c
+++ b/src/nxt_conn.c
@@ -85,8 +85,6 @@ nxt_conn_create(nxt_mp_t *mp, nxt_task_t *task)
     nxt_conn_timer_init(&c->read_timer, c, c->socket.read_work_queue);
     nxt_conn_timer_init(&c->write_timer, c, c->socket.write_work_queue);
 
-    nxt_queue_init(&c->requests);
-
     nxt_log_debug(&c->log, "connections: %uD", thr->engine->connections);
 
     return c;
@@ -165,3 +163,60 @@ nxt_conn_work_queue_set(nxt_conn_t *c, nxt_work_queue_t *wq)
     c->read_timer.work_queue = wq;
     c->write_timer.work_queue = wq;
 }
+
+
+nxt_sockaddr_t *
+nxt_conn_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;
+}
diff --git a/src/nxt_conn.h b/src/nxt_conn.h
index d2f3db3b..110b1634 100644
--- a/src/nxt_conn.h
+++ b/src/nxt_conn.h
@@ -141,8 +141,6 @@ struct nxt_conn_s {
 
     nxt_conn_io_t                 *io;
 
-    nxt_queue_t                   requests; /* of nxt_req_conn_link_t */
-
     union {
 #if (NXT_SSLTLS)
         void                      *ssltls;
@@ -231,6 +229,8 @@ NXT_EXPORT void nxt_conn_close(nxt_event_engine_t *engine, nxt_conn_t *c);
 NXT_EXPORT void nxt_conn_timer(nxt_event_engine_t *engine, nxt_conn_t *c,
     const nxt_conn_state_t *state, nxt_timer_t *tev);
 NXT_EXPORT void nxt_conn_work_queue_set(nxt_conn_t *c, nxt_work_queue_t *wq);
+NXT_EXPORT nxt_sockaddr_t *nxt_conn_local_addr(nxt_task_t *task,
+    nxt_conn_t *c);
 
 void nxt_conn_sys_socket(nxt_task_t *task, void *obj, void *data);
 void nxt_conn_io_connect(nxt_task_t *task, void *obj, void *data);
@@ -289,13 +289,13 @@ NXT_EXPORT void nxt_event_conn_job_sendfile(nxt_task_t *task,
     } while (0)
 
 
-#define nxt_conn_write(e, c)                                                  \
+#define nxt_conn_write(engine, c)                                             \
     do {                                                                      \
-        nxt_event_engine_t  *engine = e;                                      \
+        nxt_event_engine_t  *e = engine;                                      \
                                                                               \
-        c->socket.write_work_queue = &engine->write_work_queue;               \
+        c->socket.write_work_queue = &e->write_work_queue;                    \
                                                                               \
-        nxt_work_queue_add(&engine->write_work_queue, c->io->write,           \
+        nxt_work_queue_add(&e->write_work_queue, c->io->write,                \
                            c->socket.task, c, c->socket.data);                \
     } while (0)
 
diff --git a/src/nxt_h1proto.c b/src/nxt_h1proto.c
new file mode 100644
index 00000000..3d86b5dd
--- /dev/null
+++ b/src/nxt_h1proto.c
@@ -0,0 +1,989 @@
+
+/*
+ * Copyright (C) Igor Sysoev
+ * Copyright (C) NGINX, Inc.
+ */
+
+#include <nxt_router.h>
+#include <nxt_http.h>
+
+
+static void nxt_h1p_read_header(nxt_task_t *task, void *obj, void *data);
+static void nxt_h1p_header_parse(nxt_task_t *task, void *obj, void *data);
+static nxt_int_t nxt_h1p_connection(void *ctx, nxt_http_field_t *field,
+    uintptr_t data);
+static nxt_int_t nxt_h1p_transfer_encoding(void *ctx, nxt_http_field_t *field,
+    uintptr_t data);
+static void nxt_h1p_request_body_read(nxt_task_t *task, nxt_http_request_t *r);
+static void nxt_h1p_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);
+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,
+    nxt_buf_t *out);
+static void nxt_h1p_sent(nxt_task_t *task, void *obj, void *data);
+static void nxt_h1p_request_close(nxt_task_t *task, nxt_http_proto_t proto);
+static void nxt_h1p_keepalive(nxt_task_t *task, nxt_h1proto_t *h1p,
+    nxt_conn_t *c);
+static void nxt_h1p_close(nxt_task_t *task, nxt_conn_t *c);
+static void nxt_h1p_conn_close(nxt_task_t *task, void *obj, void *data);
+static void nxt_h1p_conn_error(nxt_task_t *task, void *obj, void *data);
+static void nxt_h1p_conn_timeout(nxt_task_t *task, void *obj, void *data);
+static nxt_msec_t nxt_h1p_timeout_value(nxt_conn_t *c, uintptr_t data);
+
+
+static const nxt_conn_state_t  nxt_h1p_idle_state;
+static const nxt_conn_state_t  nxt_h1p_read_header_state;
+static const nxt_conn_state_t  nxt_h1p_read_body_state;
+static const nxt_conn_state_t  nxt_h1p_send_state;
+
+
+const nxt_http_proto_body_read_t  nxt_http_proto_body_read[3] = {
+    nxt_h1p_request_body_read,
+    NULL,
+    NULL,
+};
+
+
+const nxt_http_proto_local_addr_t  nxt_http_proto_local_addr[3] = {
+    nxt_h1p_request_local_addr,
+    NULL,
+    NULL,
+};
+
+
+const nxt_http_proto_header_send_t  nxt_http_proto_header_send[3] = {
+    nxt_h1p_request_header_send,
+    NULL,
+    NULL,
+};
+
+
+const nxt_http_proto_send_t  nxt_http_proto_send[3] = {
+    nxt_h1p_request_send,
+    NULL,
+    NULL,
+};
+
+
+const nxt_http_proto_close_t  nxt_http_proto_close[3] = {
+    nxt_h1p_request_close,
+    NULL,
+    NULL,
+};
+
+
+static nxt_lvlhsh_t            nxt_h1p_fields_hash;
+
+static nxt_http_field_proc_t   nxt_h1p_fields[] = {
+    { nxt_string("Connection"),        &nxt_h1p_connection, 0 },
+    { nxt_string("Transfer-Encoding"), &nxt_h1p_transfer_encoding, 0 },
+
+    { nxt_string("Host"),              &nxt_http_request_host, 0 },
+    { nxt_string("Cookie"),            &nxt_http_request_field,
+        offsetof(nxt_http_request_t, cookie) },
+    { nxt_string("Content-Type"),      &nxt_http_request_field,
+        offsetof(nxt_http_request_t, content_type) },
+    { nxt_string("Content-Length"),    &nxt_http_request_content_length, 0 },
+};
+
+
+nxt_int_t
+nxt_h1p_init(nxt_task_t *task, nxt_runtime_t *rt)
+{
+    return nxt_http_fields_hash(&nxt_h1p_fields_hash, rt->mem_pool,
+                                nxt_h1p_fields, nxt_nitems(nxt_h1p_fields));
+}
+
+
+void
+nxt_http_conn_init(nxt_task_t *task, void *obj, void *data)
+{
+    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, "http conn init");
+
+    c->joint = joint;
+    joint->count++;
+
+    skcf = joint->socket_conf;
+    c->local = skcf->sockaddr;
+    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_h1p_idle_state;
+
+    nxt_conn_wait(c);
+}
+
+
+static const nxt_conn_state_t  nxt_h1p_idle_state
+    nxt_aligned(64) =
+{
+    .ready_handler = nxt_h1p_read_header,
+    .close_handler = nxt_h1p_conn_close,
+    .error_handler = nxt_h1p_conn_error,
+
+    .timer_handler = nxt_h1p_conn_timeout,
+    .timer_value = nxt_h1p_timeout_value,
+    .timer_data = offsetof(nxt_socket_conf_t, idle_timeout),
+};
+
+
+static void
+nxt_h1p_read_header(nxt_task_t *task, void *obj, void *data)
+{
+    size_t                   size;
+    nxt_conn_t               *c;
+    nxt_socket_conf_joint_t  *joint;
+
+    c = obj;
+
+    nxt_debug(task, "h1p read header");
+
+    if (c->read == NULL) {
+        joint = c->joint;
+        size = joint->socket_conf->header_buffer_size;
+
+        c->read = nxt_buf_mem_alloc(c->mem_pool, size, 0);
+        if (nxt_slow_path(c->read == NULL)) {
+            nxt_h1p_conn_error(task, c, c->socket.data);
+            return;
+        }
+    }
+
+    c->read_state = &nxt_h1p_read_header_state;
+
+    nxt_conn_read(task->thread->engine, c);
+}
+
+
+static const nxt_conn_state_t  nxt_h1p_read_header_state
+    nxt_aligned(64) =
+{
+    .ready_handler = nxt_h1p_header_parse,
+    .close_handler = nxt_h1p_conn_close,
+    .error_handler = nxt_h1p_conn_error,
+
+    .timer_handler = nxt_h1p_conn_timeout,
+    .timer_value = nxt_h1p_timeout_value,
+    .timer_data = offsetof(nxt_socket_conf_t, header_read_timeout),
+};
+
+
+static void
+nxt_h1p_header_parse(nxt_task_t *task, void *obj, void *data)
+{
+    size_t                   size;
+    nxt_int_t                ret;
+    nxt_buf_t                *in, *b;
+    nxt_conn_t               *c;
+    nxt_h1proto_t            *h1p;
+    nxt_http_request_t       *r;
+    nxt_socket_conf_joint_t  *joint;
+
+    c = obj;
+    h1p = data;
+
+    nxt_debug(task, "h1p header parse");
+
+    if (h1p == NULL) {
+        h1p = nxt_mp_zget(c->mem_pool, sizeof(nxt_h1proto_t));
+        if (nxt_slow_path(h1p == NULL)) {
+            goto fail;
+        }
+
+        c->socket.data = h1p;
+        h1p->conn = c;
+    }
+
+    r = h1p->request;
+
+    if (r == NULL) {
+        r = nxt_http_request_create(task);
+        if (nxt_slow_path(r == NULL)) {
+            goto fail;
+        }
+
+        h1p->request = r;
+        r->proto.h1 = h1p;
+        joint = c->joint;
+        r->socket_conf = joint->socket_conf;
+
+        r->remote = c->remote;
+
+        ret = nxt_http_parse_request_init(&h1p->parser, r->mem_pool);
+        if (nxt_slow_path(ret != NXT_OK)) {
+            /*
+             * The request is very uncomplete here,
+             * so "internal server error" useless here.
+             */
+            nxt_mp_release(r->mem_pool);
+            h1p->request = NULL;
+            goto fail;
+        }
+    }
+
+    ret = nxt_http_parse_request(&h1p->parser, &c->read->mem);
+
+    if (nxt_fast_path(ret == NXT_DONE)) {
+        r->target.start = h1p->parser.target_start;
+        r->target.length = h1p->parser.target_end - h1p->parser.target_start;
+
+        r->version.start = h1p->parser.version.str;
+        r->version.length = sizeof(h1p->parser.version.str);
+
+        r->method = &h1p->parser.method;
+        r->path = &h1p->parser.path;
+        r->args = &h1p->parser.args;
+
+        /*
+         * By default the keepalive mode is disabled in HTTP/1.0 and
+         * enabled in HTTP/1.1.  The mode can be overridden later by
+         * the "Connection" field processed in nxt_h1p_connection().
+         */
+        h1p->keepalive = (h1p->parser.version.str[7] != '0');
+
+        r->fields = h1p->parser.fields;
+
+        ret = nxt_http_fields_process(r->fields, &nxt_h1p_fields_hash, r);
+
+        if (nxt_fast_path(ret == NXT_OK)) {
+            r->state->ready_handler(task, r, NULL);
+            return;
+        }
+
+    } else if (ret == NXT_AGAIN) {
+        in = c->read;
+
+        if (nxt_buf_mem_free_size(&in->mem) == 0) {
+            size = r->socket_conf->large_header_buffer_size;
+
+            if (size <= (size_t) nxt_buf_mem_used_size(&in->mem)
+                || h1p->nbuffers >= r->socket_conf->large_header_buffers)
+            {
+                nxt_http_request_error(task, r,
+                                      NXT_HTTP_REQUEST_HEADER_FIELDS_TOO_LARGE);
+                return;
+            }
+
+            b = nxt_buf_mem_alloc(c->mem_pool, size, 0);
+            if (nxt_slow_path(b == NULL)) {
+                nxt_http_request_error(task, r, NXT_HTTP_INTERNAL_SERVER_ERROR);
+                return;
+            }
+
+            h1p->nbuffers++;
+
+            size = nxt_buf_mem_used_size(&in->mem);
+            b->mem.free = nxt_cpymem(b->mem.pos, in->mem.pos, size);
+
+            in->next = b;
+            nxt_buf_chain_add(&h1p->buffers, in);
+            c->read = b;
+        }
+
+        nxt_conn_read(task->thread->engine, c);
+        return;
+    }
+
+    /* ret == NXT_ERROR */
+
+    nxt_http_request_error(task, r, NXT_HTTP_BAD_REQUEST);
+    return;
+
+fail:
+
+    nxt_h1p_conn_close(task, c, h1p);
+}
+
+
+static nxt_int_t
+nxt_h1p_connection(void *ctx, nxt_http_field_t *field, uintptr_t data)
+{
+    nxt_http_request_t  *r;
+
+    r = ctx;
+
+    if (field->value_length == 5 && nxt_memcmp(field->value, "close", 5) == 0) {
+        r->proto.h1->keepalive = 0;
+    }
+
+    return NXT_OK;
+}
+
+
+static nxt_int_t
+nxt_h1p_transfer_encoding(void *ctx, nxt_http_field_t *field, uintptr_t data)
+{
+    nxt_http_te_t       te;
+    nxt_http_request_t  *r;
+
+    r = ctx;
+
+    if (field->value_length == 7
+        && nxt_memcmp(field->value, "chunked", 7) == 0)
+    {
+        te = NXT_HTTP_TE_CHUNKED;
+
+    } else {
+        te = NXT_HTTP_TE_UNSUPPORTED;
+    }
+
+    r->proto.h1->transfer_encoding = te;
+
+    return NXT_OK;
+}
+
+
+static void
+nxt_h1p_request_body_read(nxt_task_t *task, nxt_http_request_t *r)
+{
+    size_t             size, rest_length;
+    nxt_buf_t          *b;
+    nxt_conn_t         *c;
+    nxt_http_status_t  status;
+
+    nxt_debug(task, "h1p body read %O te:%d",
+              r->content_length_n, r->proto.h1->transfer_encoding);
+
+    switch (r->proto.h1->transfer_encoding) {
+
+    case NXT_HTTP_TE_CHUNKED:
+        status = NXT_HTTP_LENGTH_REQUIRED;
+        goto error;
+
+    case NXT_HTTP_TE_UNSUPPORTED:
+        status = NXT_HTTP_NOT_IMPLEMENTED;
+        goto error;
+
+    default:
+    case NXT_HTTP_TE_NONE:
+        break;
+    }
+
+    if (r->content_length_n == -1 || r->content_length_n == 0) {
+        goto ready;
+    }
+
+    if (r->content_length_n > (nxt_off_t) r->socket_conf->max_body_size) {
+        status = NXT_HTTP_PAYLOAD_TOO_LARGE;
+        goto error;
+    }
+
+    rest_length = (size_t) r->content_length_n;
+
+    b = r->body;
+
+    if (b == NULL) {
+        b = nxt_buf_mem_alloc(r->mem_pool, rest_length, 0);
+        if (nxt_slow_path(b == NULL)) {
+            status = NXT_HTTP_INTERNAL_SERVER_ERROR;
+            goto error;
+        }
+
+        r->body = b;
+    }
+
+    c = r->proto.h1->conn;
+
+    size = nxt_buf_mem_used_size(&c->read->mem);
+
+    if (size != 0) {
+        if (size >= rest_length) {
+            size = rest_length;
+            rest_length = 0;
+
+        } else {
+            rest_length -= size;
+        }
+
+        b->mem.free = nxt_cpymem(b->mem.free, c->read->mem.pos, size);
+        c->read->mem.pos += size;
+    }
+
+    nxt_debug(task, "h1p body rest: %O", rest_length);
+
+    r->rest_length = rest_length;
+
+    if (rest_length != 0) {
+        c->read = b;
+        c->read_state = &nxt_h1p_read_body_state;
+
+        nxt_conn_read(task->thread->engine, c);
+        return;
+    }
+
+ready:
+
+    nxt_work_queue_add(&task->thread->engine->fast_work_queue,
+                       r->state->ready_handler, task, r, NULL);
+
+    return;
+
+error:
+
+    r->proto.h1->keepalive = 0;
+
+    nxt_http_request_error(task, r, status);
+}
+
+
+static const nxt_conn_state_t  nxt_h1p_read_body_state
+    nxt_aligned(64) =
+{
+    .ready_handler = nxt_h1p_body_read,
+    .close_handler = nxt_h1p_conn_close,
+    .error_handler = nxt_h1p_conn_error,
+
+    .timer_handler = nxt_h1p_conn_timeout,
+    .timer_value = nxt_h1p_timeout_value,
+    .timer_data = offsetof(nxt_socket_conf_t, body_read_timeout),
+    .timer_autoreset = 1,
+};
+
+
+static void
+nxt_h1p_body_read(nxt_task_t *task, void *obj, void *data)
+{
+    size_t              size;
+    nxt_conn_t          *c;
+    nxt_h1proto_t       *h1p;
+    nxt_http_request_t  *r;
+
+    c = obj;
+    h1p = data;
+
+    nxt_debug(task, "h1p body read");
+
+    r = h1p->request;
+    size = nxt_buf_mem_used_size(&c->read->mem);
+
+    c->read->mem.pos += size;
+    r->rest_length -= size;
+
+    nxt_debug(task, "h1p body rest: %O", r->rest_length);
+
+    if (r->rest_length != 0) {
+        nxt_conn_read(task->thread->engine, c);
+
+    } else {
+        nxt_work_queue_add(&task->thread->engine->fast_work_queue,
+                           r->state->ready_handler, task, r, NULL);
+    }
+}
+
+
+static void
+nxt_h1p_request_local_addr(nxt_task_t *task, nxt_http_request_t *r)
+{
+    r->local = nxt_conn_local_addr(task, r->proto.h1->conn);
+}
+
+
+#define NXT_HTTP_LAST_SUCCESS                                                 \
+    (NXT_HTTP_OK + nxt_nitems(nxt_http_success) - 1)
+
+static const nxt_str_t  nxt_http_success[] = {
+    nxt_string("HTTP/1.1 200 OK\r\n"),
+    nxt_string("HTTP/1.1 201 Created\r\n"),
+    nxt_string("HTTP/1.1 202 Accepted\r\n"),
+    nxt_string("HTTP/1.1 203 Non-Authoritative Information\r\n"),
+    nxt_string("HTTP/1.1 204 No Content\r\n"),
+    nxt_string("HTTP/1.1 205 Reset Content\r\n"),
+    nxt_string("HTTP/1.1 206 Partial Content\r\n"),
+};
+
+
+#define NXT_HTTP_LAST_REDIRECTION                                             \
+    (NXT_HTTP_MULTIPLE_CHOICES + nxt_nitems(nxt_http_redirection) - 1)
+
+static const nxt_str_t  nxt_http_redirection[] = {
+    nxt_string("HTTP/1.1 300 Multiple Choices\r\n"),
+    nxt_string("HTTP/1.1 301 Moved Permanently\r\n"),
+    nxt_string("HTTP/1.1 302 Found\r\n"),
+    nxt_string("HTTP/1.1 303 See Other\r\n"),
+    nxt_string("HTTP/1.1 304 Not Modified\r\n"),
+};
+
+
+#define NXT_HTTP_LAST_CLIENT_ERROR                                            \
+    (NXT_HTTP_BAD_REQUEST + nxt_nitems(nxt_http_client_error) - 1)
+
+static const nxt_str_t  nxt_http_client_error[] = {
+    nxt_string("HTTP/1.1 400 Bad Request\r\n"),
+    nxt_string("HTTP/1.1 401 Unauthorized\r\n"),
+    nxt_string("HTTP/1.1 402 Payment Required\r\n"),
+    nxt_string("HTTP/1.1 403 Forbidden\r\n"),
+    nxt_string("HTTP/1.1 404 Not Found\r\n"),
+    nxt_string("HTTP/1.1 405 Method Not Allowed\r\n"),
+    nxt_string("HTTP/1.1 406 Not Acceptable\r\n"),
+    nxt_string("HTTP/1.1 407 Proxy Authentication Required\r\n"),
+    nxt_string("HTTP/1.1 408 Request Timeout\r\n"),
+    nxt_string("HTTP/1.1 409 Conflict\r\n"),
+    nxt_string("HTTP/1.1 410 Gone\r\n"),
+    nxt_string("HTTP/1.1 411 Length Required\r\n"),
+    nxt_string("HTTP/1.1 412 Precondition Failed\r\n"),
+    nxt_string("HTTP/1.1 413 Payload Too Large\r\n"),
+    nxt_string("HTTP/1.1 414 URI Too Long\r\n"),
+    nxt_string("HTTP/1.1 415 Unsupported Media Type\r\n"),
+    nxt_string("HTTP/1.1 416 Range Not Satisfiable\r\n"),
+    nxt_string("HTTP/1.1 417 Expectation Failed\r\n"),
+    nxt_string("HTTP/1.1 418\r\n"),
+    nxt_string("HTTP/1.1 419\r\n"),
+    nxt_string("HTTP/1.1 420\r\n"),
+    nxt_string("HTTP/1.1 421\r\n"),
+    nxt_string("HTTP/1.1 422\r\n"),
+    nxt_string("HTTP/1.1 423\r\n"),
+    nxt_string("HTTP/1.1 424\r\n"),
+    nxt_string("HTTP/1.1 425\r\n"),
+    nxt_string("HTTP/1.1 426\r\n"),
+    nxt_string("HTTP/1.1 427\r\n"),
+    nxt_string("HTTP/1.1 428\r\n"),
+    nxt_string("HTTP/1.1 429\r\n"),
+    nxt_string("HTTP/1.1 430\r\n"),
+    nxt_string("HTTP/1.1 431 Request Header Fields Too Large\r\n"),
+};
+
+
+#define NXT_HTTP_LAST_SERVER_ERROR                                            \
+    (NXT_HTTP_INTERNAL_SERVER_ERROR + nxt_nitems(nxt_http_server_error) - 1)
+
+static const nxt_str_t  nxt_http_server_error[] = {
+    nxt_string("HTTP/1.1 500 Internal Server Error\r\n"),
+    nxt_string("HTTP/1.1 501 Not Implemented\r\n"),
+    nxt_string("HTTP/1.1 502 Bad Gateway\r\n"),
+    nxt_string("HTTP/1.1 503 Service Unavailable\r\n"),
+    nxt_string("HTTP/1.1 504 Gateway Timeout\r\n"),
+};
+
+
+#define UNKNOWN_STATUS_LENGTH  (sizeof("HTTP/1.1 65536\r\n") - 1)
+
+static void
+nxt_h1p_request_header_send(nxt_task_t *task, nxt_http_request_t *r)
+{
+    u_char              *p;
+    size_t              size;
+    nxt_buf_t           *header;
+    nxt_str_t           unknown_status;
+    nxt_int_t           conn;
+    nxt_uint_t          n;
+    nxt_bool_t          http11;
+    nxt_conn_t          *c;
+    nxt_h1proto_t       *h1p;
+    const nxt_str_t     *status;
+    nxt_http_field_t    *field;
+    nxt_event_engine_t  *engine;
+    u_char              buf[UNKNOWN_STATUS_LENGTH];
+
+    static const char   chunked[] = "Transfer-Encoding: chunked\r\n";
+
+    static const nxt_str_t  connection[2] = {
+        nxt_string("Connection: close\r\n"),
+        nxt_string("Connection: keep-alive\r\n"),
+    };
+
+    nxt_debug(task, "h1p request header send");
+
+    r->header_sent = 1;
+    h1p = r->proto.h1;
+    n = r->status;
+
+    if (n >= NXT_HTTP_OK && n <= NXT_HTTP_LAST_SUCCESS) {
+        status = &nxt_http_success[n - NXT_HTTP_OK];
+
+    } else if (n >= NXT_HTTP_MULTIPLE_CHOICES
+               && n <= NXT_HTTP_LAST_REDIRECTION)
+    {
+        status = &nxt_http_redirection[n - NXT_HTTP_MULTIPLE_CHOICES];
+
+    } else if (n >= NXT_HTTP_BAD_REQUEST && n <= NXT_HTTP_LAST_CLIENT_ERROR) {
+        status = &nxt_http_client_error[n - NXT_HTTP_BAD_REQUEST];
+
+    } else if (n >= NXT_HTTP_INTERNAL_SERVER_ERROR
+               && n <= NXT_HTTP_LAST_SERVER_ERROR)
+    {
+        status = &nxt_http_server_error[n - NXT_HTTP_INTERNAL_SERVER_ERROR];
+
+    } else {
+        p = nxt_sprintf(buf, buf + UNKNOWN_STATUS_LENGTH,
+                        "HTTP/1.1 %03d\r\n", n);
+
+        unknown_status.length = p - buf;
+        unknown_status.start = buf;
+        status = &unknown_status;
+    }
+
+    size = status->length + sizeof("\r\n");
+    size += sizeof("\r\n");  /* Trailing CRLF. */
+
+    http11 = (h1p->parser.version.str[7] != '0');
+
+    if (r->resp.content_length == NULL || r->resp.content_length->skip) {
+        if (http11) {
+            h1p->chunked = 1;
+            size += sizeof(chunked) - 1;
+
+        } else {
+            h1p->keepalive = 0;
+        }
+    }
+
+    conn = -1;
+
+    if (http11 ^ h1p->keepalive) {
+        conn = h1p->keepalive;
+        size += connection[conn].length;
+    }
+
+    nxt_list_each(field, r->resp.fields) {
+
+        if (!field->skip) {
+            size += field->name_length + field->value_length;
+            size += sizeof(": \r\n") - 1;
+        }
+
+    } nxt_list_loop;
+
+    header = nxt_buf_mem_alloc(r->mem_pool, size, 0);
+    if (nxt_slow_path(header == NULL)) {
+        /* The internal server error is set just for logging. */
+        r->status = NXT_HTTP_INTERNAL_SERVER_ERROR;
+        nxt_h1p_conn_close(task, h1p->conn, h1p);
+        return;
+    }
+
+    p = header->mem.free;
+
+    p = nxt_cpymem(p, status->start, status->length);
+
+    nxt_list_each(field, r->resp.fields) {
+
+        if (!field->skip) {
+            p = nxt_cpymem(p, field->name, field->name_length);
+            *p++ = ':'; *p++ = ' ';
+            p = nxt_cpymem(p, field->value, field->value_length);
+            *p++ = '\r'; *p++ = '\n';
+        }
+
+    } nxt_list_loop;
+
+    if (conn >= 0) {
+        p = nxt_cpymem(p, connection[conn].start, connection[conn].length);
+    }
+
+    if (h1p->chunked) {
+        p = nxt_cpymem(p, chunked, sizeof(chunked) - 1);
+
+    } else {
+        *p++ = '\r'; *p++ = '\n';
+    }
+
+    header->mem.free = p;
+
+    c = h1p->conn;
+
+    c->write = header;
+    c->write_state = &nxt_h1p_send_state;
+
+    engine = task->thread->engine;
+
+    nxt_work_queue_add(&engine->fast_work_queue, r->state->ready_handler,
+                       task, r, NULL);
+
+    nxt_conn_write(engine, c);
+}
+
+
+static const nxt_conn_state_t  nxt_h1p_send_state
+    nxt_aligned(64) =
+{
+    .ready_handler = nxt_h1p_sent,
+    .close_handler = nxt_h1p_conn_close,
+    .error_handler = nxt_h1p_conn_error,
+
+    .timer_handler = nxt_h1p_conn_timeout,
+    .timer_value = nxt_h1p_timeout_value,
+    .timer_data = offsetof(nxt_socket_conf_t, send_timeout),
+    .timer_autoreset = 1,
+};
+
+
+static void
+nxt_h1p_request_send(nxt_task_t *task, nxt_http_request_t *r, nxt_buf_t *out)
+{
+    nxt_conn_t  *c;
+
+    nxt_debug(task, "h1p request send");
+
+    c = r->proto.h1->conn;
+
+    if (r->proto.h1->chunked) {
+        out = nxt_h1p_chunk_create(task, r, out);
+        if (nxt_slow_path(out == NULL)) {
+            nxt_h1p_conn_error(task, c, c->socket.data);
+            return;
+        }
+    }
+
+    if (c->write == NULL) {
+        c->write = out;
+        c->write_state = &nxt_h1p_send_state;
+
+        nxt_conn_write(task->thread->engine, c);
+
+    } else {
+        nxt_buf_chain_add(&c->write, out);
+    }
+}
+
+
+static nxt_buf_t *
+nxt_h1p_chunk_create(nxt_task_t *task, nxt_http_request_t *r, nxt_buf_t *out)
+{
+    size_t             size;
+    nxt_buf_t          *b, **prev, *header, *tail;
+
+    const size_t       chunk_size = 2 * (sizeof("\r\n") - 1) + NXT_OFF_T_HEXLEN;
+    static const char  tail_chunk[] = "\r\n0\r\n\r\n";
+
+    size = 0;
+    prev = &out;
+
+    for (b = out; b != NULL; b = b->next) {
+
+        if (nxt_buf_is_last(b)) {
+            tail = nxt_buf_mem_alloc(r->mem_pool, chunk_size, 0);
+            if (nxt_slow_path(tail == NULL)) {
+                return NULL;
+            }
+
+            *prev = tail;
+            tail->next = b;
+            /*
+             * The tail_chunk size with trailing zero is 8 bytes, so
+             * memcpy may be inlined with just single 8 byte move operation.
+             */
+            nxt_memcpy(tail->mem.free, tail_chunk, sizeof(tail_chunk));
+            tail->mem.free += sizeof(tail_chunk) - 1;
+
+            break;
+        }
+
+        size += nxt_buf_used_size(b);
+        prev = &b->next;
+    }
+
+    if (size == 0) {
+        return out;
+    }
+
+    header = nxt_buf_mem_alloc(r->mem_pool, chunk_size, 0);
+    if (nxt_slow_path(header == NULL)) {
+        return NULL;
+    }
+
+    header->next = out;
+    header->mem.free = nxt_sprintf(header->mem.free, header->mem.end,
+                                   "\r\n%xO\r\n", size);
+    return header;
+}
+
+
+static void
+nxt_h1p_sent(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_conn_t          *c;
+    nxt_event_engine_t  *engine;
+
+    c = obj;
+
+    nxt_debug(task, "h1p sent");
+
+    engine = task->thread->engine;
+
+    c->write = nxt_sendbuf_completion0(task, &engine->fast_work_queue,
+                                       c->write);
+    if (c->write != NULL) {
+        nxt_conn_write(engine, c);
+    }
+}
+
+
+static void
+nxt_h1p_request_close(nxt_task_t *task, nxt_http_proto_t proto)
+{
+    nxt_conn_t     *c;
+    nxt_h1proto_t  *h1p;
+
+    nxt_debug(task, "h1p request close");
+
+    h1p = proto.h1;
+    h1p->request = NULL;
+
+    c = h1p->conn;
+
+    if (h1p->keepalive) {
+        nxt_h1p_keepalive(task, h1p, c);
+
+    } else {
+        nxt_h1p_close(task, c);
+    }
+}
+
+
+static void
+nxt_h1p_keepalive(nxt_task_t *task, nxt_h1proto_t *h1p, nxt_conn_t *c)
+{
+    size_t     size;
+    nxt_buf_t  *in, *b, *next;
+
+    nxt_debug(task, "h1p keepalive");
+
+    b = h1p->buffers;
+
+    nxt_memzero(&h1p->parser, sizeof(nxt_h1proto_t));
+
+    h1p->conn = c;
+    in = c->read;
+
+    size = nxt_buf_mem_used_size(&in->mem);
+
+    if (size == 0) {
+        if (b != NULL) {
+            in = b;
+            c->read = in;
+
+            for (b = b->next; b != NULL; b = next) {
+                next = b->next;
+                nxt_mp_free(c->mem_pool, b);
+            }
+
+            in->next = NULL;
+        }
+
+        in->mem.pos = in->mem.start;
+        in->mem.free = in->mem.start;
+
+        if (c->socket.read_ready) {
+            c->read_state = &nxt_h1p_read_header_state;
+            nxt_conn_read(task->thread->engine, c);
+
+        } else {
+            c->read_state = &nxt_h1p_idle_state;
+            nxt_conn_wait(c);
+        }
+
+    } else {
+        nxt_debug(task, "h1p pipelining");
+
+        if (b != NULL) {
+            do {
+                next = b->next;
+                nxt_mp_free(c->mem_pool, b);
+                b = next;
+            } while (b != in);
+        }
+
+        nxt_memmove(in->mem.start, in->mem.pos, size);
+
+        in->mem.pos = in->mem.start;
+        in->mem.free = in->mem.start + size;
+
+        nxt_h1p_header_parse(task, c, c->socket.data);
+    }
+}
+
+
+static void
+nxt_h1p_close(nxt_task_t *task, nxt_conn_t *c)
+{
+    nxt_debug(task, "h1p close");
+
+    c->socket.data = NULL;
+
+    if (c->socket.fd != -1) {
+        c->write_state = &nxt_router_conn_close_state;
+
+        nxt_conn_close(task->thread->engine, c);
+    }
+}
+
+
+static void
+nxt_h1p_conn_close(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_conn_t          *c;
+    nxt_h1proto_t       *h1p;
+    nxt_http_request_t  *r;
+
+    c = obj;
+    h1p = data;
+
+    nxt_debug(task, "h1p conn close");
+
+    if (h1p != NULL) {
+        r = h1p->request;
+
+        if (r != NULL) {
+            r->state->error_handler(task, r, r->proto.h1);
+            return;
+        }
+    }
+
+    nxt_h1p_close(task, c);
+}
+
+
+static void
+nxt_h1p_conn_error(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_conn_t     *c;
+    nxt_h1proto_t  *h1p;
+
+    c = obj;
+    h1p = data;
+
+    nxt_debug(task, "h1p conn error");
+
+    nxt_h1p_conn_close(task, c, h1p);
+}
+
+
+static void
+nxt_h1p_conn_timeout(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_conn_t   *c;
+    nxt_timer_t  *timer;
+
+    timer = obj;
+
+    nxt_debug(task, "h1p conn timeout");
+
+    c = nxt_read_timer_conn(timer);
+
+    nxt_h1p_conn_close(task, c, c->socket.data);
+}
+
+
+static nxt_msec_t
+nxt_h1p_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);
+}
diff --git a/src/nxt_http.h b/src/nxt_http.h
new file mode 100644
index 00000000..c6704d68
--- /dev/null
+++ b/src/nxt_http.h
@@ -0,0 +1,181 @@
+
+/*
+ * Copyright (C) Igor Sysoev
+ * Copyright (C) NGINX, Inc.
+ */
+
+#ifndef _NXT_HTTP_H_INCLUDED_
+#define _NXT_HTTP_H_INCLUDED_
+
+
+typedef enum {
+    NXT_HTTP_INVALID = 0,
+
+    NXT_HTTP_OK = 200,
+
+    NXT_HTTP_MULTIPLE_CHOICES = 300,
+    NXT_HTTP_MOVED_PERMANENTLY = 301,
+    NXT_HTTP_FOUND = 302,
+    NXT_HTTP_SEE_OTHER = 303,
+    NXT_HTTP_NOT_MODIFIED = 304,
+
+    NXT_HTTP_BAD_REQUEST = 400,
+    NXT_HTTP_LENGTH_REQUIRED = 411,
+    NXT_HTTP_PAYLOAD_TOO_LARGE = 413,
+    NXT_HTTP_URI_TOO_LONG = 414,
+    NXT_HTTP_REQUEST_HEADER_FIELDS_TOO_LARGE = 431,
+
+    NXT_HTTP_INTERNAL_SERVER_ERROR = 500,
+    NXT_HTTP_NOT_IMPLEMENTED = 501,
+    NXT_HTTP_BAD_GATEWAY = 502,
+    NXT_HTTP_SERVICE_UNAVAILABLE = 503,
+    NXT_HTTP_GATEWAY_TIMEOUT = 504,
+} nxt_http_status_t;
+
+
+typedef enum {
+    NXT_HTTP_TE_NONE = 0,
+    NXT_HTTP_TE_CHUNKED = 1,
+    NXT_HTTP_TE_UNSUPPORTED = 2,
+} nxt_http_te_t;
+
+
+typedef struct {
+    nxt_work_handler_t              ready_handler;
+    nxt_work_handler_t              error_handler;
+} nxt_http_request_state_t;
+
+
+typedef struct nxt_http_request_s   nxt_http_request_t;
+
+
+typedef struct {
+    nxt_http_request_parse_t        parser;
+
+    uint8_t                         nbuffers;
+    uint8_t                         keepalive;            /* 1 bit  */
+    uint8_t                         chunked;              /* 1 bit  */
+    nxt_http_te_t                   transfer_encoding:8;  /* 2 bits */
+
+    nxt_http_request_t              *request;
+    nxt_conn_t                      *conn;
+    nxt_buf_t                       *buffers;
+} nxt_h1proto_t;
+
+
+typedef union {
+    void                            *any;
+    nxt_h1proto_t                   *h1;
+} nxt_http_proto_t;
+
+
+#define nxt_http_field_name_set(_field, _name)                                \
+    do {                                                                      \
+         (_field)->name_length = sizeof(_name) - 1;                           \
+         (_field)->name = (u_char *) _name;                                   \
+    } while (0)
+
+
+#define nxt_http_field_set(_field, _name, _value)                             \
+    do {                                                                      \
+         (_field)->name_length = sizeof(_name) - 1;                           \
+         (_field)->value_length = sizeof(_value) - 1;                         \
+         (_field)->name = (u_char *) _name;                                   \
+         (_field)->value = (u_char *) _value;                                 \
+    } while (0)
+
+
+typedef struct {
+    nxt_list_t                      *fields;
+    nxt_http_field_t                *content_type;
+    nxt_http_field_t                *content_length;
+    nxt_off_t                       content_length_n;
+} nxt_http_response_t;
+
+
+struct nxt_http_request_s {
+    nxt_http_proto_t                proto;
+    nxt_socket_conf_t               *socket_conf;
+
+    nxt_mp_t                        *mem_pool;
+
+    nxt_buf_t                       *body;
+    nxt_buf_t                       *out;
+    const nxt_http_request_state_t  *state;
+
+    nxt_str_t                       target;
+    nxt_str_t                       version;
+    nxt_str_t                       *method;
+    nxt_str_t                       *path;
+    nxt_str_t                       *args;
+
+    nxt_list_t                      *fields;
+    nxt_http_field_t                *host;
+    nxt_http_field_t                *content_type;
+    nxt_http_field_t                *content_length;
+    nxt_http_field_t                *cookie;
+    nxt_off_t                       content_length_n;
+    nxt_off_t                       rest_length;
+
+    nxt_sockaddr_t                  *remote;
+    nxt_sockaddr_t                  *local;
+
+    nxt_http_response_t             resp;
+
+    nxt_http_status_t               status:16;
+
+    uint8_t                         protocol;     /* 2 bits */
+    uint8_t                         logged;       /* 1 bit  */
+    uint8_t                         header_sent;  /* 1 bit  */
+};
+
+
+typedef void (*nxt_http_proto_body_read_t)(nxt_task_t *task,
+    nxt_http_request_t *r);
+typedef void (*nxt_http_proto_local_addr_t)(nxt_task_t *task,
+    nxt_http_request_t *r);
+typedef void (*nxt_http_proto_header_send_t)(nxt_task_t *task,
+    nxt_http_request_t *r);
+typedef void (*nxt_http_proto_send_t)(nxt_task_t *task, nxt_http_request_t *r,
+    nxt_buf_t *out);
+typedef void (*nxt_http_proto_close_t)(nxt_task_t *task,
+    nxt_http_proto_t proto);
+
+
+nxt_int_t nxt_http_init(nxt_task_t *task, nxt_runtime_t *rt);
+nxt_int_t nxt_h1p_init(nxt_task_t *task, nxt_runtime_t *rt);
+nxt_int_t nxt_http_response_hash_init(nxt_task_t *task, nxt_runtime_t *rt);
+
+void nxt_http_conn_init(nxt_task_t *task, void *obj, void *data);
+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_local_addr(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_send(nxt_task_t *task, nxt_http_request_t *r,
+    nxt_buf_t *out);
+void nxt_http_request_release(nxt_task_t *task, nxt_http_request_t *r);
+nxt_buf_t *nxt_http_request_last_buffer(nxt_task_t *task,
+    nxt_http_request_t *r);
+void nxt_http_request_close_handler(nxt_task_t *task, void *obj, void *data);
+
+nxt_int_t nxt_http_request_host(void *ctx, nxt_http_field_t *field,
+    uintptr_t data);
+nxt_int_t nxt_http_request_field(void *ctx, nxt_http_field_t *field,
+    uintptr_t offset);
+nxt_int_t nxt_http_request_content_length(void *ctx, nxt_http_field_t *field,
+    uintptr_t data);
+
+
+extern nxt_lvlhsh_t                        nxt_response_fields_hash;
+extern const nxt_conn_state_t              nxt_router_conn_close_state;
+
+extern const nxt_http_proto_body_read_t    nxt_http_proto_body_read[];
+extern const nxt_http_proto_local_addr_t   nxt_http_proto_local_addr[];
+extern const nxt_http_proto_header_send_t  nxt_http_proto_header_send[];
+extern const nxt_http_proto_send_t         nxt_http_proto_send[];
+extern const nxt_http_proto_close_t        nxt_http_proto_close[];
+
+
+#endif  /* _NXT_HTTP_H_INCLUDED_ */
diff --git a/src/nxt_http_error.c b/src/nxt_http_error.c
new file mode 100644
index 00000000..5030264b
--- /dev/null
+++ b/src/nxt_http_error.c
@@ -0,0 +1,105 @@
+
+/*
+ * Copyright (C) Igor Sysoev
+ * Copyright (C) NGINX, Inc.
+ */
+
+#include <nxt_router.h>
+#include <nxt_http.h>
+
+
+static void nxt_http_request_send_error_body(nxt_task_t *task, void *r,
+    void *data);
+
+
+static const nxt_http_request_state_t  nxt_http_request_send_state;
+
+
+static const char  error[] =
+    "<html><head><title>Error</title><head>"
+    "<body>Error.</body></html>\r\n";
+
+
+void
+nxt_http_request_error(nxt_task_t *task, nxt_http_request_t *r,
+    nxt_http_status_t status)
+{
+    nxt_http_field_t  *content_type;
+
+    nxt_debug(task, "http request error: %d", status);
+
+    if (r->header_sent) {
+        goto fail;
+    }
+
+    r->status = status;
+
+    r->resp.fields = nxt_list_create(r->mem_pool, 8, sizeof(nxt_http_field_t));
+    if (nxt_slow_path(r == NULL)) {
+        goto fail;
+    }
+
+    content_type = nxt_list_zero_add(r->resp.fields);
+    if (nxt_slow_path(content_type == NULL)) {
+        goto fail;
+    }
+
+    nxt_http_field_set(content_type, "Content-Type", "text/html");
+
+    r->resp.content_length = NULL;
+    r->resp.content_length_n = sizeof(error) - 1;
+
+    r->state = &nxt_http_request_send_state;
+
+    nxt_http_request_header_send(task, r);
+    return;
+
+fail:
+
+    nxt_http_request_release(task, r);
+}
+
+
+static const nxt_http_request_state_t  nxt_http_request_send_state
+    nxt_aligned(64) =
+{
+    .ready_handler = nxt_http_request_send_error_body,
+    .error_handler = nxt_http_request_close_handler,
+};
+
+
+static void
+nxt_http_request_send_error_body(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_buf_t           *out, *last;
+    nxt_http_request_t  *r;
+
+    r = obj;
+
+    nxt_debug(task, "http request send error body");
+
+    out = nxt_buf_mem_alloc(r->mem_pool, 0, 0);
+    if (nxt_slow_path(out == NULL)) {
+        goto fail;
+    }
+
+    out->mem.start = (u_char *) error;
+    out->mem.pos = out->mem.start;
+    out->mem.free = out->mem.start + sizeof(error) - 1;
+    out->mem.end = out->mem.free;
+
+    last = nxt_http_request_last_buffer(task, r);
+    if (nxt_slow_path(last == NULL)) {
+        goto fail;
+    }
+
+    out->next = last;
+
+    nxt_http_request_send(task, r, out);
+
+    return;
+
+fail:
+    // TODO
+    nxt_http_request_release(task, r);
+}
diff --git a/src/nxt_http_request.c b/src/nxt_http_request.c
new file mode 100644
index 00000000..eab61814
--- /dev/null
+++ b/src/nxt_http_request.c
@@ -0,0 +1,384 @@
+
+/*
+ * Copyright (C) Igor Sysoev
+ * Copyright (C) NGINX, Inc.
+ */
+
+#include <nxt_router.h>
+#include <nxt_http.h>
+
+
+static void nxt_http_request_start(nxt_task_t *task, void *obj, void *data);
+static void nxt_http_app_request(nxt_task_t *task, void *obj, void *data);
+static void nxt_http_request_done(nxt_task_t *task, void *obj, void *data);
+
+
+static const nxt_http_request_state_t  nxt_http_request_init_state;
+static const nxt_http_request_state_t  nxt_http_request_body_state;
+
+
+nxt_int_t
+nxt_http_init(nxt_task_t *task, nxt_runtime_t *rt)
+{
+    nxt_int_t  ret;
+
+    ret = nxt_h1p_init(task, rt);
+
+    if (ret != NXT_OK) {
+        return ret;
+    }
+
+    return nxt_http_response_hash_init(task, rt);
+}
+
+
+nxt_int_t
+nxt_http_request_host(void *ctx, nxt_http_field_t *field, uintptr_t data)
+{
+    nxt_http_request_t  *r;
+
+    r = ctx;
+
+    /* TODO: validate host. */
+
+    r->host = field;
+
+    return NXT_OK;
+}
+
+
+nxt_int_t
+nxt_http_request_field(void *ctx, nxt_http_field_t *field, uintptr_t offset)
+{
+    nxt_http_request_t  *r;
+
+    r = ctx;
+
+    nxt_value_at(nxt_http_field_t *, r, offset) = field;
+
+    return NXT_OK;
+}
+
+
+nxt_int_t
+nxt_http_request_content_length(void *ctx, nxt_http_field_t *field,
+    uintptr_t data)
+{
+    nxt_http_request_t  *r;
+
+    r = ctx;
+
+    r->content_length = field;
+    r->content_length_n = nxt_off_t_parse(field->value, field->value_length);
+
+    return NXT_OK;
+}
+
+
+nxt_http_request_t *
+nxt_http_request_create(nxt_task_t *task)
+{
+    nxt_mp_t            *mp;
+    nxt_http_request_t  *r;
+
+    mp = nxt_mp_create(1024, 128, 256, 32);
+    if (nxt_slow_path(mp == NULL)) {
+        return NULL;
+    }
+
+    r = nxt_mp_zget(mp, sizeof(nxt_http_request_t));
+    if (nxt_slow_path(r == NULL)) {
+        goto fail;
+    }
+
+    r->resp.fields = nxt_list_create(mp, 8, sizeof(nxt_http_field_t));
+    if (nxt_slow_path(r == NULL)) {
+        goto fail;
+    }
+
+    r->mem_pool = mp;
+    r->content_length_n = -1;
+    r->resp.content_length_n = -1;
+    r->state = &nxt_http_request_init_state;
+
+    return r;
+
+fail:
+
+    nxt_mp_release(mp);
+    return NULL;
+}
+
+
+static const nxt_http_request_state_t  nxt_http_request_init_state
+    nxt_aligned(64) =
+{
+    .ready_handler = nxt_http_request_start,
+    .error_handler = nxt_http_request_close_handler,
+};
+
+
+static void
+nxt_http_request_start(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_http_request_t  *r;
+
+    r = obj;
+
+    r->state = &nxt_http_request_body_state;
+
+    nxt_http_request_read_body(task, r);
+}
+
+
+static const nxt_http_request_state_t  nxt_http_request_body_state
+    nxt_aligned(64) =
+{
+    .ready_handler = nxt_http_app_request,
+    .error_handler = nxt_http_request_close_handler,
+};
+
+
+static void
+nxt_http_app_request(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_int_t            ret;
+    nxt_event_engine_t   *engine;
+    nxt_http_request_t   *r;
+    nxt_app_parse_ctx_t  *ar;
+
+    r = obj;
+
+    ar = nxt_mp_zget(r->mem_pool, sizeof(nxt_app_parse_ctx_t));
+    if (nxt_slow_path(ar == NULL)) {
+        nxt_http_request_error(task, r, NXT_HTTP_INTERNAL_SERVER_ERROR);
+        return;
+    }
+
+    ar->request = r;
+    ar->mem_pool = r->mem_pool;
+    nxt_mp_retain(r->mem_pool);
+
+    // STUB
+    engine = task->thread->engine;
+    ar->timer.task = &engine->task;
+    ar->timer.work_queue = &engine->fast_work_queue;
+    ar->timer.log = engine->task.log;
+    ar->timer.precision = NXT_TIMER_DEFAULT_PRECISION;
+
+    ar->r.remote.start = nxt_sockaddr_address(r->remote);
+    ar->r.remote.length = r->remote->address_length;
+
+    /*
+     * TODO: need an application flag to get local address
+     * required by "SERVER_ADDR" in Pyhton and PHP. Not used in Go.
+     */
+    nxt_http_request_local_addr(task, r);
+
+    if (nxt_fast_path(r->local != NULL)) {
+        ar->r.local.start = nxt_sockaddr_address(r->local);
+        ar->r.local.length = r->local->address_length;
+    }
+
+    ar->r.header.fields = r->fields;
+    ar->r.header.done = 1;
+    ar->r.header.version = r->version;
+
+    if (r->method != NULL) {
+        ar->r.header.method = *r->method;
+    }
+
+    ar->r.header.target = r->target;
+
+    if (r->path != NULL) {
+        ar->r.header.path = *r->path;
+    }
+
+    if (r->args != NULL) {
+        ar->r.header.query = *r->args;
+    }
+
+    if (r->host != NULL) {
+        ar->r.header.host.length = r->host->value_length;
+        ar->r.header.host.start = r->host->value;
+    }
+
+    if (r->content_type != NULL) {
+        ar->r.header.content_type.length = r->content_type->value_length;
+        ar->r.header.content_type.start = r->content_type->value;
+    }
+
+    if (r->content_length != NULL) {
+        ar->r.header.content_length.length = r->content_length->value_length;
+        ar->r.header.content_length.start = r->content_length->value;
+    }
+
+    if (r->cookie != NULL) {
+        ar->r.header.cookie.length = r->cookie->value_length;
+        ar->r.header.cookie.start = r->cookie->value;
+    }
+
+    ar->r.body.done = 1;
+
+    ret = nxt_http_parse_request_init(&ar->resp_parser, r->mem_pool);
+    if (nxt_slow_path(ret != NXT_OK)) {
+        nxt_http_request_error(task, r, NXT_HTTP_INTERNAL_SERVER_ERROR);
+        return;
+    }
+
+    nxt_router_process_http_request(task, ar);
+}
+
+
+void
+nxt_http_request_read_body(nxt_task_t *task, nxt_http_request_t *r)
+{
+    if (r->proto.any != NULL) {
+        nxt_http_proto_body_read[r->protocol](task, r);
+    }
+}
+
+
+void
+nxt_http_request_local_addr(nxt_task_t *task, nxt_http_request_t *r)
+{
+    if (r->proto.any != NULL) {
+        nxt_http_proto_local_addr[r->protocol](task, r);
+    }
+}
+
+
+void
+nxt_http_request_header_send(nxt_task_t *task, nxt_http_request_t *r)
+{
+    u_char            *p, *end;
+    nxt_http_field_t  *server, *content_length;
+
+    /*
+     * TODO: "Server" and "Content-Length" processing should be moved
+     * to the last header filter.
+     */
+
+    server = nxt_list_zero_add(r->resp.fields);
+    if (nxt_slow_path(server == NULL)) {
+        goto fail;
+    }
+
+    nxt_http_field_set(server, "Server", "unit/" NXT_VERSION);
+
+    if (r->resp.content_length_n != -1
+        && (r->resp.content_length == NULL || r->resp.content_length->skip))
+    {
+        content_length = nxt_list_zero_add(r->resp.fields);
+        if (nxt_slow_path(content_length == NULL)) {
+            goto fail;
+        }
+
+        nxt_http_field_name_set(content_length, "Content-Length");
+
+        p = nxt_mp_nget(r->mem_pool, NXT_OFF_T_LEN);
+        if (nxt_slow_path(p == NULL)) {
+            goto fail;
+        }
+
+        content_length->value = p;
+        end = nxt_sprintf(p, p + NXT_OFF_T_LEN, "%O", r->resp.content_length_n);
+        content_length->value_length = end - p;
+
+        r->resp.content_length = content_length;
+    }
+
+    if (r->proto.any != NULL) {
+        nxt_http_proto_header_send[r->protocol](task, r);
+    }
+
+    return;
+
+fail:
+
+    nxt_http_request_error(task, r, NXT_HTTP_INTERNAL_SERVER_ERROR);
+}
+
+
+void
+nxt_http_request_send(nxt_task_t *task, nxt_http_request_t *r, nxt_buf_t *out)
+{
+    if (r->proto.any != NULL) {
+        nxt_http_proto_send[r->protocol](task, r, out);
+    }
+}
+
+
+nxt_buf_t *
+nxt_http_request_last_buffer(nxt_task_t *task, nxt_http_request_t *r)
+{
+    nxt_buf_t  *b;
+
+    b = nxt_buf_mem_alloc(r->mem_pool, 0, 0);
+
+    if (nxt_fast_path(b != NULL)) {
+        nxt_buf_set_sync(b);
+        nxt_buf_set_last(b);
+        b->completion_handler = nxt_http_request_done;
+        b->parent = r;
+
+    } else {
+        nxt_http_request_release(task, r);
+    }
+
+    return b;
+}
+
+
+static void
+nxt_http_request_done(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_http_request_t  *r;
+
+    r = data;
+
+    nxt_debug(task, "http request done");
+
+    nxt_http_request_close_handler(task, r, r->proto.any);
+}
+
+
+void
+nxt_http_request_release(nxt_task_t *task, nxt_http_request_t *r)
+{
+    nxt_debug(task, "http request release");
+
+    nxt_work_queue_add(&task->thread->engine->fast_work_queue,
+                       nxt_http_request_close_handler, task, r, r->proto.any);
+}
+
+
+void
+nxt_http_request_close_handler(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_http_proto_t        proto;
+    nxt_http_request_t      *r;
+    nxt_http_proto_close_t  handler;
+
+    r = obj;
+    proto.any = data;
+
+    nxt_debug(task, "http request close handler");
+
+    if (!r->logged) {
+        r->logged = 1;
+        // STUB
+        nxt_debug(task, "http request log: \"%*s \"%V %V %V\" %d\"",
+                  r->remote->address_length, nxt_sockaddr_address(r->remote),
+                  r->method, &r->target, &r->version, r->status);
+    }
+
+    handler = nxt_http_proto_close[r->protocol];
+
+    r->proto.any = NULL;
+    nxt_mp_release(r->mem_pool);
+
+    if (proto.any != NULL) {
+        handler(task, proto);
+    }
+}
diff --git a/src/nxt_http_response.c b/src/nxt_http_response.c
new file mode 100644
index 00000000..330890c1
--- /dev/null
+++ b/src/nxt_http_response.c
@@ -0,0 +1,83 @@
+
+/*
+ * Copyright (C) Igor Sysoev
+ * Copyright (C) NGINX, Inc.
+ */
+
+#include <nxt_router.h>
+#include <nxt_http.h>
+
+
+static nxt_int_t nxt_http_response_status(void *ctx, nxt_http_field_t *field,
+    uintptr_t data);
+static nxt_int_t nxt_http_response_skip(void *ctx, nxt_http_field_t *field,
+    uintptr_t data);
+static nxt_int_t nxt_http_response_field(void *ctx, nxt_http_field_t *field,
+    uintptr_t offset);
+
+
+nxt_lvlhsh_t  nxt_response_fields_hash;
+
+static nxt_http_field_proc_t   nxt_response_fields[] = {
+    { nxt_string("Status"),         &nxt_http_response_status, 0 },
+    { nxt_string("Server"),         &nxt_http_response_skip, 0 },
+    { nxt_string("Connection"),     &nxt_http_response_skip, 0 },
+    { nxt_string("Content-Type"),   &nxt_http_response_field,
+        offsetof(nxt_http_request_t, resp.content_type) },
+    { nxt_string("Content-Length"), &nxt_http_response_field,
+        offsetof(nxt_http_request_t, resp.content_length) },
+};
+
+
+nxt_int_t
+nxt_http_response_hash_init(nxt_task_t *task, nxt_runtime_t *rt)
+{
+    return nxt_http_fields_hash(&nxt_response_fields_hash, rt->mem_pool,
+                    nxt_response_fields, nxt_nitems(nxt_response_fields));
+}
+
+
+nxt_int_t
+nxt_http_response_status(void *ctx, nxt_http_field_t *field,
+    uintptr_t data)
+{
+    nxt_int_t           status;
+    nxt_http_request_t  *r;
+
+    r = ctx;
+
+    field->skip = 1;
+
+    if (field->value_length >= 3) {
+        status = nxt_int_parse(field->value, 3);
+
+        if (status >= 100 && status <= 999) {
+            r->status = status;
+            return NXT_OK;
+        }
+    }
+
+    return NXT_ERROR;
+}
+
+
+nxt_int_t
+nxt_http_response_skip(void *ctx, nxt_http_field_t *field, uintptr_t data)
+{
+    field->skip = 1;
+
+    return NXT_OK;
+}
+
+
+nxt_int_t
+nxt_http_response_field(void *ctx, nxt_http_field_t *field, uintptr_t offset)
+{
+    nxt_http_request_t  *r;
+
+    r = ctx;
+
+    nxt_value_at(nxt_http_field_t *, r, offset) = field;
+
+    return NXT_OK;
+}
diff --git a/src/nxt_php_sapi.c b/src/nxt_php_sapi.c
index 14427b03..10b5a4e9 100644
--- a/src/nxt_php_sapi.c
+++ b/src/nxt_php_sapi.c
@@ -505,19 +505,11 @@ nxt_php_send_headers(sapi_headers_struct *sapi_headers TSRMLS_DC)
     zend_llist_position  zpos;
 
     static const u_char default_repsonse[]
-        = "HTTP/1.1 200 OK\r\n"
-          "Server: unit/" NXT_VERSION "\r\n"
-          "Content-Type: text/html; charset=UTF-8\r\n"
-          "Connection: close\r\n"
+        = "Status: 200\r\n"
           "\r\n";
 
-    static const u_char default_headers[]
-        = "Server: unit/" NXT_VERSION "\r\n"
-          "Connection: close\r\n";
-
-    static const u_char http_11[] = "HTTP/1.1 ";
+    static const u_char status_200[] = "Status: 200";
     static const u_char cr_lf[] = "\r\n";
-    static const u_char _200_ok[] = "200 OK";
 
     ctx = SG(server_context);
 
@@ -539,23 +531,26 @@ nxt_php_send_headers(sapi_headers_struct *sapi_headers TSRMLS_DC)
         status = (u_char *) SG(sapi_headers).http_status_line;
         len = nxt_strlen(status);
 
-        RC(nxt_php_write(ctx, status, len, 0, 0));
+        if (len < 12) {
+            goto fail;
+        }
+
+        RC(nxt_php_write(ctx, status_200, sizeof(status_200) - 4, 0, 0));
+        RC(nxt_php_write(ctx, status + 9, 3, 0, 0));
 
     } else if (SG(sapi_headers).http_response_code) {
         status = nxt_sprintf(buf, buf + sizeof(buf), "%03d",
                         SG(sapi_headers).http_response_code);
         len = status - buf;
 
-        RC(nxt_php_write(ctx, http_11, sizeof(http_11) - 1, 0, 0));
+        RC(nxt_php_write(ctx, status_200, sizeof(status_200) - 4, 0, 0));
         RC(nxt_php_write(ctx, buf, len, 0, 0));
 
     } else {
-        RC(nxt_php_write(ctx, http_11, sizeof(http_11) - 1, 0, 0));
-        RC(nxt_php_write(ctx, _200_ok, sizeof(_200_ok) - 1, 0, 0));
+        RC(nxt_php_write(ctx, status_200, sizeof(status_200) - 1, 0, 0));
     }
 
     RC(nxt_php_write(ctx, cr_lf, sizeof(cr_lf) - 1, 0, 0));
-    RC(nxt_php_write(ctx, default_headers, sizeof(default_headers) - 1, 0, 0));
 
     h = zend_llist_get_first_ex(&sapi_headers->headers, &zpos);
 
diff --git a/src/nxt_python_wsgi.c b/src/nxt_python_wsgi.c
index 0f56ea37..ae0a60d9 100644
--- a/src/nxt_python_wsgi.c
+++ b/src/nxt_python_wsgi.c
@@ -775,11 +775,7 @@ nxt_py_start_resp(PyObject *self, PyObject *args)
     nxt_uint_t  i, n;
     nxt_python_run_ctx_t  *ctx;
 
-    static const u_char resp[] = "HTTP/1.1 ";
-
-    static const u_char default_headers[]
-        = "Server: unit/" NXT_VERSION "\r\n"
-          "Connection: close\r\n";
+    static const u_char status[] = "Status: ";
 
     static const u_char cr_lf[] = "\r\n";
     static const u_char sc_sp[] = ": ";
@@ -794,7 +790,7 @@ nxt_py_start_resp(PyObject *self, PyObject *args)
 
     ctx = nxt_python_run_ctx;
 
-    nxt_python_write(ctx, resp, sizeof(resp) - 1, 0, 0);
+    nxt_python_write(ctx, status, sizeof(status) - 1, 0, 0);
 
     rc = nxt_python_write_py_str(ctx, string, 0, 0);
     if (nxt_slow_path(rc != NXT_OK)) {
@@ -804,8 +800,6 @@ nxt_py_start_resp(PyObject *self, PyObject *args)
 
     nxt_python_write(ctx, cr_lf, sizeof(cr_lf) - 1, 0, 0);
 
-    nxt_python_write(ctx, default_headers, sizeof(default_headers) - 1, 0, 0);
-
     headers = PyTuple_GET_ITEM(args, 1);
 
     if (!PyList_Check(headers)) {
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);
+    }
 }
diff --git a/src/nxt_router.h b/src/nxt_router.h
index 76a04d6e..c405c8f3 100644
--- a/src/nxt_router.h
+++ b/src/nxt_router.h
@@ -14,6 +14,9 @@
 #include <nxt_application.h>
 
 
+typedef struct nxt_http_request_s  nxt_http_request_t;
+
+
 typedef struct {
     nxt_thread_spinlock_t  lock;
     nxt_queue_t            engines;
@@ -127,8 +130,10 @@ typedef struct {
     size_t                 large_header_buffers;
     size_t                 body_buffer_size;
     size_t                 max_body_size;
+    nxt_msec_t             idle_timeout;
     nxt_msec_t             header_read_timeout;
     nxt_msec_t             body_read_timeout;
+    nxt_msec_t             send_timeout;
 } nxt_socket_conf_t;
 
 
@@ -146,7 +151,9 @@ void nxt_router_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
 void nxt_router_conf_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
 void nxt_router_remove_pid_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
 
+void nxt_router_process_http_request(nxt_task_t *task, nxt_app_parse_ctx_t *ar);
 void nxt_router_app_port_close(nxt_task_t *task, nxt_port_t *port);
 void nxt_router_app_use(nxt_task_t *task, nxt_app_t *app, int i);
 
+
 #endif  /* _NXT_ROUTER_H_INCLUDED_ */
diff --git a/src/nxt_sendbuf.c b/src/nxt_sendbuf.c
index 1684f67c..b36c5edb 100644
--- a/src/nxt_sendbuf.c
+++ b/src/nxt_sendbuf.c
@@ -434,3 +434,23 @@ nxt_sendbuf_completion(nxt_task_t *task, nxt_work_queue_t *wq, nxt_buf_t *b,
 
     return b;
 }
+
+
+nxt_buf_t *
+nxt_sendbuf_completion0(nxt_task_t *task, nxt_work_queue_t *wq, nxt_buf_t *b)
+{
+    while (b != NULL) {
+
+        nxt_prefetch(b->next);
+
+        if (nxt_buf_used_size(b) != 0) {
+            break;
+        }
+
+        nxt_work_queue_add(wq, b->completion_handler, task, b, b->parent);
+
+        b = b->next;
+    }
+
+    return b;
+}
diff --git a/src/nxt_sendbuf.h b/src/nxt_sendbuf.h
index 3bcbaaf4..0406faf6 100644
--- a/src/nxt_sendbuf.h
+++ b/src/nxt_sendbuf.h
@@ -124,6 +124,8 @@ ssize_t nxt_sendbuf_copy_coalesce(nxt_conn_t *c, nxt_buf_mem_t *bm,
 nxt_buf_t *nxt_sendbuf_update(nxt_buf_t *b, size_t sent);
 nxt_buf_t *nxt_sendbuf_completion(nxt_task_t *task, nxt_work_queue_t *wq,
     nxt_buf_t *b, size_t sent, nxt_bool_t mmap_mode);
+nxt_buf_t *nxt_sendbuf_completion0(nxt_task_t *task, nxt_work_queue_t *wq,
+    nxt_buf_t *b);
 
 
 #endif /* _NXT_SENDBUF_H_INCLUDED_ */