technocloud-public/nginx-unit
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Initial proxy support.

Browse Source
This commit is contained in:
Igor Sysoev
2019-11-14 16:39:54 +03:00
parent 57e326b411
commit ddde9c23cf
16 changed files with 1264 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

787
src/nxt_h1proto.c
View File

@@ -45,7 +45,7 @@ static void nxt_h1p_conn_request_body_read(nxt_task_t *task, void *obj,
void *data);
static void nxt_h1p_request_local_addr(nxt_task_t *task, nxt_http_request_t *r);
static void nxt_h1p_request_header_send(nxt_task_t *task,
nxt_http_request_t *r, nxt_work_handler_t body_handler);
nxt_http_request_t *r, nxt_work_handler_t body_handler, void *data);
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,
@@ -78,11 +78,32 @@ static void nxt_h1p_idle_response_timeout(nxt_task_t *task, void *obj,
static nxt_msec_t nxt_h1p_idle_response_timer_value(nxt_conn_t *c,
uintptr_t data);
static void nxt_h1p_shutdown(nxt_task_t *task, nxt_conn_t *c);
static void nxt_h1p_shutdown_(nxt_task_t *task, nxt_conn_t *c);
static void nxt_h1p_closing(nxt_task_t *task, nxt_conn_t *c);
static void nxt_h1p_conn_ws_shutdown(nxt_task_t *task, void *obj, void *data);
static void nxt_h1p_conn_closing(nxt_task_t *task, void *obj, void *data);
static void nxt_h1p_conn_free(nxt_task_t *task, void *obj, void *data);
static void nxt_h1p_peer_connect(nxt_task_t *task, nxt_http_peer_t *peer);
static void nxt_h1p_peer_connected(nxt_task_t *task, void *obj, void *data);
static void nxt_h1p_peer_refused(nxt_task_t *task, void *obj, void *data);
static void nxt_h1p_peer_header_send(nxt_task_t *task, nxt_http_peer_t *peer);
static void nxt_h1p_peer_header_sent(nxt_task_t *task, void *obj, void *data);
static void nxt_h1p_peer_header_read(nxt_task_t *task, nxt_http_peer_t *peer);
static ssize_t nxt_h1p_peer_io_read_handler(nxt_task_t *task, nxt_conn_t *c);
static void nxt_h1p_peer_header_read_done(nxt_task_t *task, void *obj,
void *data);
static nxt_int_t nxt_h1p_peer_header_parse(nxt_http_peer_t *peer,
nxt_buf_mem_t *bm);
static void nxt_h1p_peer_read(nxt_task_t *task, nxt_http_peer_t *peer);
static void nxt_h1p_peer_read_done(nxt_task_t *task, void *obj, void *data);
static void nxt_h1p_peer_closed(nxt_task_t *task, void *obj, void *data);
static void nxt_h1p_peer_error(nxt_task_t *task, void *obj, void *data);
static void nxt_h1p_peer_send_timeout(nxt_task_t *task, void *obj, void *data);
static void nxt_h1p_peer_read_timeout(nxt_task_t *task, void *obj, void *data);
static nxt_msec_t nxt_h1p_peer_timer_value(nxt_conn_t *c, uintptr_t data);
static void nxt_h1p_peer_close(nxt_task_t *task, nxt_http_peer_t *peer);
static void nxt_h1p_peer_free(nxt_task_t *task, void *obj, void *data);
#if (NXT_TLS)
static const nxt_conn_state_t nxt_http_idle_state;
static const nxt_conn_state_t nxt_h1p_shutdown_state;
@@ -94,6 +115,13 @@ static const nxt_conn_state_t nxt_h1p_request_send_state;
static const nxt_conn_state_t nxt_h1p_timeout_response_state;
static const nxt_conn_state_t nxt_h1p_keepalive_state;
static const nxt_conn_state_t nxt_h1p_close_state;
static const nxt_conn_state_t nxt_h1p_peer_connect_state;
static const nxt_conn_state_t nxt_h1p_peer_header_send_state;
static const nxt_conn_state_t nxt_h1p_peer_header_body_send_state;
static const nxt_conn_state_t nxt_h1p_peer_header_read_state;
static const nxt_conn_state_t nxt_h1p_peer_header_read_timer_state;
static const nxt_conn_state_t nxt_h1p_peer_read_state;
static const nxt_conn_state_t nxt_h1p_peer_close_state;
const nxt_http_proto_table_t nxt_http_proto[3] = {
@@ -106,6 +134,13 @@ const nxt_http_proto_table_t nxt_http_proto[3] = {
.body_bytes_sent = nxt_h1p_request_body_bytes_sent,
.discard = nxt_h1p_request_discard,
.close = nxt_h1p_request_close,
.peer_connect = nxt_h1p_peer_connect,
.peer_header_send = nxt_h1p_peer_header_send,
.peer_header_read = nxt_h1p_peer_header_read,
.peer_read = nxt_h1p_peer_read,
.peer_close = nxt_h1p_peer_close,
.ws_frame_start = nxt_h1p_websocket_frame_start,
},
/* NXT_HTTP_PROTO_H2 */
@@ -113,9 +148,9 @@ const nxt_http_proto_table_t nxt_http_proto[3] = {
};
static nxt_lvlhsh_t nxt_h1p_fields_hash;
static nxt_lvlhsh_t nxt_h1p_fields_hash;
static nxt_http_field_proc_t nxt_h1p_fields[] = {
static nxt_http_field_proc_t nxt_h1p_fields[] = {
{ nxt_string("Connection"), &nxt_h1p_connection, 0 },
{ nxt_string("Upgrade"), &nxt_h1p_upgrade, 0 },
{ nxt_string("Sec-WebSocket-Key"), &nxt_h1p_websocket_key, 0 },
@@ -136,11 +171,32 @@ static nxt_http_field_proc_t nxt_h1p_fields[] = {
};
static nxt_lvlhsh_t nxt_h1p_peer_fields_hash;
static nxt_http_field_proc_t nxt_h1p_peer_fields[] = {
{ nxt_string("Connection"), &nxt_http_proxy_skip, 0 },
{ nxt_string("Transfer-Encoding"), &nxt_http_proxy_skip, 0 },
{ nxt_string("Server"), &nxt_http_proxy_skip, 0 },
{ nxt_string("Date"), &nxt_http_proxy_date, 0 },
{ nxt_string("Content-Length"), &nxt_http_proxy_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));
nxt_int_t ret;
ret = nxt_http_fields_hash(&nxt_h1p_fields_hash, rt->mem_pool,
nxt_h1p_fields, nxt_nitems(nxt_h1p_fields));
if (nxt_fast_path(ret == NXT_OK)) {
ret = nxt_http_fields_hash(&nxt_h1p_peer_fields_hash,
rt->mem_pool, nxt_h1p_peer_fields,
nxt_nitems(nxt_h1p_peer_fields));
}
return ret;
}
@@ -303,7 +359,7 @@ nxt_http_conn_test(nxt_task_t *task, void *obj, void *data)
* Listening socket had been closed while
* connection was in keep-alive state.
*/
nxt_h1p_shutdown(task, c);
nxt_h1p_closing(task, c);
return;
}
@@ -388,7 +444,7 @@ nxt_h1p_conn_proto_init(nxt_task_t *task, void *obj, void *data)
h1p = nxt_mp_zget(c->mem_pool, sizeof(nxt_h1proto_t));
if (nxt_slow_path(h1p == NULL)) {
nxt_h1p_shutdown(task, c);
nxt_h1p_closing(task, c);
return;
}
@@ -452,7 +508,7 @@ nxt_h1p_conn_request_init(nxt_task_t *task, void *obj, void *data)
nxt_mp_release(r->mem_pool);
}
nxt_h1p_shutdown(task, c);
nxt_h1p_closing(task, c);
}
@@ -668,6 +724,7 @@ nxt_h1p_connection(void *ctx, nxt_http_field_t *field, uintptr_t data)
nxt_http_request_t *r;
r = ctx;
field->hopbyhop = 1;
if (field->value_length == 5 && nxt_memcmp(field->value, "close", 5) == 0) {
r->proto.h1->keepalive = 0;
@@ -738,6 +795,8 @@ nxt_h1p_transfer_encoding(void *ctx, nxt_http_field_t *field, uintptr_t data)
nxt_http_request_t *r;
r = ctx;
field->skip = 1;
field->hopbyhop = 1;
if (field->value_length == 7
&& nxt_memcmp(field->value, "chunked", 7) == 0)
@@ -1003,7 +1062,7 @@ static const nxt_str_t nxt_http_server_error[] = {
static void
nxt_h1p_request_header_send(nxt_task_t *task, nxt_http_request_t *r,
nxt_work_handler_t body_handler)
nxt_work_handler_t body_handler, void *data)
{
u_char *p;
size_t size;
@@ -1180,7 +1239,7 @@ nxt_h1p_request_header_send(nxt_task_t *task, nxt_http_request_t *r,
* in engine->write_work_queue.
*/
nxt_work_queue_add(&task->thread->engine->fast_work_queue,
body_handler, task, r, NULL);
body_handler, task, r, data);
} else {
header->next = nxt_http_buf_last(r);
@@ -1784,14 +1843,31 @@ nxt_h1p_shutdown(nxt_task_t *task, nxt_conn_t *c)
}
} else {
nxt_h1p_shutdown_(task, c);
nxt_h1p_closing(task, c);
}
}
static void
nxt_h1p_shutdown_(nxt_task_t *task, nxt_conn_t *c)
nxt_h1p_conn_ws_shutdown(nxt_task_t *task, void *obj, void *data)
{
nxt_timer_t *timer;
nxt_h1p_websocket_timer_t *ws_timer;
nxt_debug(task, "h1p conn ws shutdown");
timer = obj;
ws_timer = nxt_timer_data(timer, nxt_h1p_websocket_timer_t, timer);
nxt_h1p_closing(task, ws_timer->h1p->conn);
}
static void
nxt_h1p_closing(nxt_task_t *task, nxt_conn_t *c)
{
nxt_debug(task, "h1p closing");
c->socket.data = NULL;
#if (NXT_TLS)
@@ -1822,21 +1898,6 @@ static const nxt_conn_state_t nxt_h1p_shutdown_state
#endif
static void
nxt_h1p_conn_ws_shutdown(nxt_task_t *task, void *obj, void *data)
{
nxt_timer_t *timer;
nxt_h1p_websocket_timer_t *ws_timer;
nxt_debug(task, "h1p conn ws shutdown");
timer = obj;
ws_timer = nxt_timer_data(timer, nxt_h1p_websocket_timer_t, timer);
nxt_h1p_shutdown_(task, ws_timer->h1p->conn);
}
static void
nxt_h1p_conn_closing(nxt_task_t *task, void *obj, void *data)
{
@@ -1882,3 +1943,673 @@ nxt_h1p_conn_free(nxt_task_t *task, void *obj, void *data)
nxt_router_listen_event_release(&engine->task, lev, NULL);
}
static void
nxt_h1p_peer_connect(nxt_task_t *task, nxt_http_peer_t *peer)
{
nxt_mp_t *mp;
nxt_int_t ret;
nxt_conn_t *c, *client;
nxt_h1proto_t *h1p;
nxt_fd_event_t *socket;
nxt_work_queue_t *wq;
nxt_http_request_t *r;
nxt_debug(task, "h1p peer connect");
peer->status = NXT_HTTP_UNSET;
r = peer->request;
mp = nxt_mp_create(1024, 128, 256, 32);
if (nxt_slow_path(mp == NULL)) {
goto fail;
}
h1p = nxt_mp_zalloc(mp, sizeof(nxt_h1proto_t));
if (nxt_slow_path(h1p == NULL)) {
goto fail;
}
ret = nxt_http_parse_request_init(&h1p->parser, r->mem_pool);
if (nxt_slow_path(ret != NXT_OK)) {
goto fail;
}
c = nxt_conn_create(mp, task);
if (nxt_slow_path(c == NULL)) {
goto fail;
}
c->mem_pool = mp;
h1p->conn = c;
peer->proto.h1 = h1p;
h1p->request = r;
c->socket.task = task;
c->read_timer.task = task;
c->write_timer.task = task;
c->socket.data = peer;
c->remote = peer->sockaddr;
c->socket.write_ready = 1;
c->write_state = &nxt_h1p_peer_connect_state;
/*
* TODO: queues should be implemented via client proto interface.
*/
client = r->proto.h1->conn;
socket = &client->socket;
wq = socket->read_work_queue;
c->read_work_queue = wq;
c->socket.read_work_queue = wq;
c->read_timer.work_queue = wq;
wq = socket->write_work_queue;
c->write_work_queue = wq;
c->socket.write_work_queue = wq;
c->write_timer.work_queue = wq;
/* TODO END */
nxt_conn_connect(task->thread->engine, c);
return;
fail:
peer->status = NXT_HTTP_INTERNAL_SERVER_ERROR;
r->state->error_handler(task, r, peer);
}
static const nxt_conn_state_t nxt_h1p_peer_connect_state
nxt_aligned(64) =
{
.ready_handler = nxt_h1p_peer_connected,
.close_handler = nxt_h1p_peer_refused,
.error_handler = nxt_h1p_peer_error,
.timer_handler = nxt_h1p_peer_send_timeout,
.timer_value = nxt_h1p_peer_timer_value,
.timer_data = offsetof(nxt_socket_conf_t, proxy_timeout),
};
static void
nxt_h1p_peer_connected(nxt_task_t *task, void *obj, void *data)
{
nxt_http_peer_t *peer;
nxt_http_request_t *r;
peer = data;
nxt_debug(task, "h1p peer connected");
r = peer->request;
r->state->ready_handler(task, r, peer);
}
static void
nxt_h1p_peer_refused(nxt_task_t *task, void *obj, void *data)
{
nxt_http_peer_t *peer;
nxt_http_request_t *r;
peer = data;
nxt_debug(task, "h1p peer refused");
//peer->status = NXT_HTTP_SERVICE_UNAVAILABLE;
peer->status = NXT_HTTP_BAD_GATEWAY;
r = peer->request;
r->state->error_handler(task, r, peer);
}
static void
nxt_h1p_peer_header_send(nxt_task_t *task, nxt_http_peer_t *peer)
{
u_char *p;
size_t size;
nxt_buf_t *header, *body;
nxt_conn_t *c;
nxt_http_field_t *field;
nxt_http_request_t *r;
nxt_debug(task, "h1p peer header send");
r = peer->request;
size = r->method->length + sizeof(" ") + r->target.length
+ sizeof(" HTTP/1.0\r\n")
+ sizeof("\r\n");
nxt_list_each(field, r->fields) {
if (!field->hopbyhop) {
size += field->name_length + field->value_length;
size += nxt_length(": \r\n");
}
} nxt_list_loop;
header = nxt_http_buf_mem(task, r, size);
if (nxt_slow_path(header == NULL)) {
r->state->error_handler(task, r, peer);
return;
}
p = header->mem.free;
p = nxt_cpymem(p, r->method->start, r->method->length);
*p++ = ' ';
p = nxt_cpymem(p, r->target.start, r->target.length);
p = nxt_cpymem(p, " HTTP/1.0\r\n", 11);
nxt_list_each(field, r->fields) {
if (!field->hopbyhop) {
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;
*p++ = '\r'; *p++ = '\n';
header->mem.free = p;
size = p - header->mem.pos;
c = peer->proto.h1->conn;
c->write = header;
c->write_state = &nxt_h1p_peer_header_send_state;
if (r->body != NULL) {
body = nxt_buf_mem_alloc(r->mem_pool, 0, 0);
if (nxt_slow_path(body == NULL)) {
r->state->error_handler(task, r, peer);
return;
}
header->next = body;
body->mem = r->body->mem;
size += nxt_buf_mem_used_size(&body->mem);
// nxt_mp_retain(r->mem_pool);
}
if (size > 16384) {
/* Use proxy_send_timeout instead of proxy_timeout. */
c->write_state = &nxt_h1p_peer_header_body_send_state;
}
nxt_conn_write(task->thread->engine, c);
}
static const nxt_conn_state_t nxt_h1p_peer_header_send_state
nxt_aligned(64) =
{
.ready_handler = nxt_h1p_peer_header_sent,
.error_handler = nxt_h1p_peer_error,
.timer_handler = nxt_h1p_peer_send_timeout,
.timer_value = nxt_h1p_peer_timer_value,
.timer_data = offsetof(nxt_socket_conf_t, proxy_timeout),
};
static const nxt_conn_state_t nxt_h1p_peer_header_body_send_state
nxt_aligned(64) =
{
.ready_handler = nxt_h1p_peer_header_sent,
.error_handler = nxt_h1p_peer_error,
.timer_handler = nxt_h1p_peer_send_timeout,
.timer_value = nxt_h1p_peer_timer_value,
.timer_data = offsetof(nxt_socket_conf_t, proxy_send_timeout),
.timer_autoreset = 1,
};
static void
nxt_h1p_peer_header_sent(nxt_task_t *task, void *obj, void *data)
{
nxt_conn_t *c;
nxt_http_peer_t *peer;
nxt_http_request_t *r;
nxt_event_engine_t *engine;
c = obj;
peer = data;
nxt_debug(task, "h1p peer header sent");
engine = task->thread->engine;
c->write = nxt_sendbuf_completion(task, &engine->fast_work_queue, c->write);
if (c->write == NULL) {
r = peer->request;
r->state->ready_handler(task, r, peer);
return;
}
nxt_conn_write(engine, c);
}
static void
nxt_h1p_peer_header_read(nxt_task_t *task, nxt_http_peer_t *peer)
{
nxt_conn_t *c;
nxt_debug(task, "h1p peer header read");
c = peer->proto.h1->conn;
if (c->write_timer.enabled) {
c->read_state = &nxt_h1p_peer_header_read_state;
} else {
c->read_state = &nxt_h1p_peer_header_read_timer_state;
}
nxt_conn_read(task->thread->engine, c);
}
static const nxt_conn_state_t nxt_h1p_peer_header_read_state
nxt_aligned(64) =
{
.ready_handler = nxt_h1p_peer_header_read_done,
.close_handler = nxt_h1p_peer_closed,
.error_handler = nxt_h1p_peer_error,
.io_read_handler = nxt_h1p_peer_io_read_handler,
};
static const nxt_conn_state_t nxt_h1p_peer_header_read_timer_state
nxt_aligned(64) =
{
.ready_handler = nxt_h1p_peer_header_read_done,
.close_handler = nxt_h1p_peer_closed,
.error_handler = nxt_h1p_peer_error,
.io_read_handler = nxt_h1p_peer_io_read_handler,
.timer_handler = nxt_h1p_peer_read_timeout,
.timer_value = nxt_h1p_peer_timer_value,
.timer_data = offsetof(nxt_socket_conf_t, proxy_timeout),
};
static ssize_t
nxt_h1p_peer_io_read_handler(nxt_task_t *task, nxt_conn_t *c)
{
size_t size;
ssize_t n;
nxt_buf_t *b;
nxt_http_peer_t *peer;
nxt_socket_conf_t *skcf;
nxt_http_request_t *r;
peer = c->socket.data;
r = peer->request;
b = c->read;
if (b == NULL) {
skcf = r->conf->socket_conf;
size = (peer->header_received) ? skcf->proxy_buffer_size
: skcf->proxy_header_buffer_size;
nxt_debug(task, "h1p peer io read: %z", size);
b = nxt_http_proxy_buf_mem_alloc(task, r, size);
if (nxt_slow_path(b == NULL)) {
c->socket.error = NXT_ENOMEM;
return NXT_ERROR;
}
}
n = c->io->recvbuf(c, b);
if (n > 0) {
c->read = b;
} else {
c->read = NULL;
nxt_http_proxy_buf_mem_free(task, r, b);
}
return n;
}
static void
nxt_h1p_peer_header_read_done(nxt_task_t *task, void *obj, void *data)
{
nxt_int_t ret;
nxt_buf_t *b;
nxt_conn_t *c;
nxt_http_peer_t *peer;
nxt_http_request_t *r;
nxt_event_engine_t *engine;
c = obj;
peer = data;
nxt_debug(task, "h1p peer header read done");
b = c->read;
ret = nxt_h1p_peer_header_parse(peer, &b->mem);
r = peer->request;
ret = nxt_expect(NXT_DONE, ret);
if (ret != NXT_AGAIN) {
engine = task->thread->engine;
nxt_timer_disable(engine, &c->write_timer);
nxt_timer_disable(engine, &c->read_timer);
}
switch (ret) {
case NXT_DONE:
peer->fields = peer->proto.h1->parser.fields;
ret = nxt_http_fields_process(peer->fields,
&nxt_h1p_peer_fields_hash, r);
if (nxt_slow_path(ret != NXT_OK)) {
peer->status = NXT_HTTP_INTERNAL_SERVER_ERROR;
break;
}
c->read = NULL;
if (nxt_buf_mem_used_size(&b->mem) != 0) {
peer->body = b;
}
peer->header_received = 1;
r->state->ready_handler(task, r, peer);
return;
case NXT_AGAIN:
if (nxt_buf_mem_free_size(&b->mem) != 0) {
nxt_conn_read(task->thread->engine, c);
return;
}
/* Fall through. */
default:
case NXT_ERROR:
case NXT_HTTP_PARSE_INVALID:
case NXT_HTTP_PARSE_UNSUPPORTED_VERSION:
case NXT_HTTP_PARSE_TOO_LARGE_FIELD:
peer->status = NXT_HTTP_BAD_GATEWAY;
break;
}
nxt_http_proxy_buf_mem_free(task, r, b);
r->state->error_handler(task, r, peer);
}
static nxt_int_t
nxt_h1p_peer_header_parse(nxt_http_peer_t *peer, nxt_buf_mem_t *bm)
{
u_char *p;
size_t length;
nxt_int_t status;
if (peer->status < 0) {
length = nxt_buf_mem_used_size(bm);
if (nxt_slow_path(length < 12)) {
return NXT_AGAIN;
}
p = bm->pos;
if (nxt_slow_path(nxt_memcmp(p, "HTTP/1.", 7) != 0
|| (p[7] != '0' && p[7] != '1')))
{
return NXT_ERROR;
}
status = nxt_int_parse(&p[9], 3);
if (nxt_slow_path(status < 0)) {
return NXT_ERROR;
}
p += 12;
length -= 12;
p = nxt_memchr(p, '\n', length);
if (nxt_slow_path(p == NULL)) {
return NXT_AGAIN;
}
bm->pos = p + 1;
peer->status = status;
}
return nxt_http_parse_fields(&peer->proto.h1->parser, bm);
}
static void
nxt_h1p_peer_read(nxt_task_t *task, nxt_http_peer_t *peer)
{
nxt_conn_t *c;
nxt_debug(task, "h1p peer read");
c = peer->proto.h1->conn;
c->read_state = &nxt_h1p_peer_read_state;
nxt_conn_read(task->thread->engine, c);
}
static const nxt_conn_state_t nxt_h1p_peer_read_state
nxt_aligned(64) =
{
.ready_handler = nxt_h1p_peer_read_done,
.close_handler = nxt_h1p_peer_closed,
.error_handler = nxt_h1p_peer_error,
.io_read_handler = nxt_h1p_peer_io_read_handler,
.timer_handler = nxt_h1p_peer_read_timeout,
.timer_value = nxt_h1p_peer_timer_value,
.timer_data = offsetof(nxt_socket_conf_t, proxy_read_timeout),
.timer_autoreset = 1,
};
static void
nxt_h1p_peer_read_done(nxt_task_t *task, void *obj, void *data)
{
nxt_conn_t *c;
nxt_http_peer_t *peer;
nxt_http_request_t *r;
c = obj;
peer = data;
nxt_debug(task, "h1p peer read done");
peer->body = c->read;
c->read = NULL;
r = peer->request;
r->state->ready_handler(task, r, peer);
}
static void
nxt_h1p_peer_closed(nxt_task_t *task, void *obj, void *data)
{
nxt_http_peer_t *peer;
nxt_http_request_t *r;
peer = data;
nxt_debug(task, "h1p peer closed");
r = peer->request;
if (peer->header_received) {
peer->body = nxt_http_buf_last(r);
peer->closed = 1;
r->state->ready_handler(task, r, peer);
} else {
peer->status = NXT_HTTP_BAD_GATEWAY;
r->state->error_handler(task, r, peer);
}
}
static void
nxt_h1p_peer_error(nxt_task_t *task, void *obj, void *data)
{
nxt_http_peer_t *peer;
nxt_http_request_t *r;
peer = data;
nxt_debug(task, "h1p peer error");
peer->status = NXT_HTTP_BAD_GATEWAY;
r = peer->request;
r->state->error_handler(task, r, peer);
}
static void
nxt_h1p_peer_send_timeout(nxt_task_t *task, void *obj, void *data)
{
nxt_conn_t *c;
nxt_timer_t *timer;
nxt_http_peer_t *peer;
nxt_http_request_t *r;
timer = obj;
nxt_debug(task, "h1p peer send timeout");
c = nxt_write_timer_conn(timer);
c->block_write = 1;
c->block_read = 1;
peer = c->socket.data;
peer->status = NXT_HTTP_GATEWAY_TIMEOUT;
r = peer->request;
r->state->error_handler(task, r, peer);
}
static void
nxt_h1p_peer_read_timeout(nxt_task_t *task, void *obj, void *data)
{
nxt_conn_t *c;
nxt_timer_t *timer;
nxt_http_peer_t *peer;
nxt_http_request_t *r;
timer = obj;
nxt_debug(task, "h1p peer read timeout");
c = nxt_read_timer_conn(timer);
c->block_write = 1;
c->block_read = 1;
peer = c->socket.data;
peer->status = NXT_HTTP_GATEWAY_TIMEOUT;
r = peer->request;
r->state->error_handler(task, r, peer);
}
static nxt_msec_t
nxt_h1p_peer_timer_value(nxt_conn_t *c, uintptr_t data)
{
nxt_http_peer_t *peer;
peer = c->socket.data;
return nxt_value_at(nxt_msec_t, peer->request->conf->socket_conf, data);
}
static void
nxt_h1p_peer_close(nxt_task_t *task, nxt_http_peer_t *peer)
{
nxt_conn_t *c;
nxt_debug(task, "h1p peer close");
peer->closed = 1;
c = peer->proto.h1->conn;
task = &c->task;
c->socket.task = task;
c->read_timer.task = task;
c->write_timer.task = task;
if (c->socket.fd != -1) {
c->write_state = &nxt_h1p_peer_close_state;
nxt_conn_close(task->thread->engine, c);
} else {
nxt_h1p_peer_free(task, c, NULL);
}
}
static const nxt_conn_state_t nxt_h1p_peer_close_state
nxt_aligned(64) =
{
.ready_handler = nxt_h1p_peer_free,
};
static void
nxt_h1p_peer_free(nxt_task_t *task, void *obj, void *data)
{
nxt_conn_t *c;
c = obj;
nxt_debug(task, "h1p peer free");
nxt_conn_free(task, c);
}