summaryrefslogtreecommitdiffhomepage
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--src/nxt_conf_validation.c25
-rw-r--r--src/nxt_h1proto.c370
-rw-r--r--src/nxt_h1proto.h48
-rw-r--r--src/nxt_h1proto_websocket.c719
-rw-r--r--src/nxt_http.h28
-rw-r--r--src/nxt_http_request.c21
-rw-r--r--src/nxt_http_response.c2
-rw-r--r--src/nxt_http_websocket.c161
-rw-r--r--src/nxt_port.c1
-rw-r--r--src/nxt_port.h15
-rw-r--r--src/nxt_router.c181
-rw-r--r--src/nxt_router.h9
-rw-r--r--src/nxt_router_request.h71
-rw-r--r--src/nxt_sha1.c295
-rw-r--r--src/nxt_sha1.h24
-rw-r--r--src/nxt_unit.c1159
-rw-r--r--src/nxt_unit.h32
-rw-r--r--src/nxt_unit_request.h1
-rw-r--r--src/nxt_unit_typedefs.h27
-rw-r--r--src/nxt_unit_websocket.h27
-rw-r--r--src/nxt_websocket.c122
-rw-r--r--src/nxt_websocket.h21
-rw-r--r--src/nxt_websocket_accept.c68
-rw-r--r--src/nxt_websocket_header.h68
-rw-r--r--src/test/nxt_unit_websocket_chat.c348
-rw-r--r--src/test/nxt_unit_websocket_echo.c105
26 files changed, 3508 insertions, 440 deletions
diff --git a/src/nxt_conf_validation.c b/src/nxt_conf_validation.c
index 45c0eb41..ca8ec62e 100644
--- a/src/nxt_conf_validation.c
+++ b/src/nxt_conf_validation.c
@@ -102,6 +102,26 @@ static nxt_int_t nxt_conf_vldt_java_option(nxt_conf_validation_t *vldt,
nxt_conf_value_t *value);
+static nxt_conf_vldt_object_t nxt_conf_vldt_websocket_members[] = {
+ { nxt_string("read_timeout"),
+ NXT_CONF_VLDT_INTEGER,
+ NULL,
+ NULL },
+
+ { nxt_string("keepalive_interval"),
+ NXT_CONF_VLDT_INTEGER,
+ NULL,
+ NULL },
+
+ { nxt_string("max_frame_size"),
+ NXT_CONF_VLDT_INTEGER,
+ NULL,
+ NULL },
+
+ NXT_CONF_VLDT_END
+};
+
+
static nxt_conf_vldt_object_t nxt_conf_vldt_http_members[] = {
{ nxt_string("header_read_timeout"),
NXT_CONF_VLDT_INTEGER,
@@ -128,6 +148,11 @@ static nxt_conf_vldt_object_t nxt_conf_vldt_http_members[] = {
NULL,
NULL },
+ { nxt_string("websocket"),
+ NXT_CONF_VLDT_OBJECT,
+ &nxt_conf_vldt_object,
+ (void *) &nxt_conf_vldt_websocket_members },
+
NXT_CONF_VLDT_END
};
diff --git a/src/nxt_h1proto.c b/src/nxt_h1proto.c
index a92dfe3b..a60bdb36 100644
--- a/src/nxt_h1proto.c
+++ b/src/nxt_h1proto.c
@@ -6,26 +6,9 @@
#include <nxt_router.h>
#include <nxt_http.h>
-
-
-struct nxt_h1proto_s {
- 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 */
-
- uint32_t header_size;
-
- nxt_http_request_t *request;
- nxt_buf_t *buffers;
- /*
- * All fields before the conn field will
- * be zeroed in a keep-alive connection.
- */
- nxt_conn_t *conn;
-};
+#include <nxt_h1proto.h>
+#include <nxt_websocket.h>
+#include <nxt_websocket_header.h>
/*
@@ -43,12 +26,18 @@ static void nxt_h1p_conn_proto_init(nxt_task_t *task, void *obj, void *data);
static void nxt_h1p_conn_request_init(nxt_task_t *task, void *obj, void *data);
static void nxt_h1p_conn_request_header_parse(nxt_task_t *task, void *obj,
void *data);
-static nxt_int_t nxt_h1p_header_process(nxt_h1proto_t *h1p,
+static nxt_int_t nxt_h1p_header_process(nxt_task_t *task, nxt_h1proto_t *h1p,
nxt_http_request_t *r);
static nxt_int_t nxt_h1p_header_buffer_test(nxt_task_t *task,
nxt_h1proto_t *h1p, nxt_conn_t *c, nxt_socket_conf_t *skcf);
static nxt_int_t nxt_h1p_connection(void *ctx, nxt_http_field_t *field,
uintptr_t data);
+static nxt_int_t nxt_h1p_upgrade(void *ctx, nxt_http_field_t *field,
+ uintptr_t data);
+static nxt_int_t nxt_h1p_websocket_key(void *ctx, nxt_http_field_t *field,
+ uintptr_t data);
+static nxt_int_t nxt_h1p_websocket_version(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);
@@ -70,8 +59,6 @@ static void nxt_h1p_conn_request_timeout(nxt_task_t *task, void *obj,
void *data);
static void nxt_h1p_conn_request_send_timeout(nxt_task_t *task, void *obj,
void *data);
-static nxt_msec_t nxt_h1p_conn_request_timer_value(nxt_conn_t *c,
- uintptr_t data);
nxt_inline void nxt_h1p_request_error(nxt_task_t *task, nxt_h1proto_t *h1p,
nxt_http_request_t *r);
static void nxt_h1p_request_close(nxt_task_t *task, nxt_http_proto_t proto,
@@ -91,16 +78,16 @@ 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_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);
-
#if (NXT_TLS)
static const nxt_conn_state_t nxt_http_idle_state;
static const nxt_conn_state_t nxt_h1p_shutdown_state;
#endif
static const nxt_conn_state_t nxt_h1p_idle_state;
-static const nxt_conn_state_t nxt_h1p_idle_close_state;
static const nxt_conn_state_t nxt_h1p_header_parse_state;
static const nxt_conn_state_t nxt_h1p_read_body_state;
static const nxt_conn_state_t nxt_h1p_request_send_state;
@@ -119,6 +106,7 @@ 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,
+ .ws_frame_start = nxt_h1p_websocket_frame_start,
},
/* NXT_HTTP_PROTO_H2 */
/* NXT_HTTP_PROTO_DEVNULL */
@@ -129,6 +117,10 @@ 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("Upgrade"), &nxt_h1p_upgrade, 0 },
+ { nxt_string("Sec-WebSocket-Key"), &nxt_h1p_websocket_key, 0 },
+ { nxt_string("Sec-WebSocket-Version"),
+ &nxt_h1p_websocket_version, 0 },
{ nxt_string("Transfer-Encoding"), &nxt_h1p_transfer_encoding, 0 },
{ nxt_string("Host"), &nxt_http_request_host, 0 },
@@ -503,7 +495,7 @@ nxt_h1p_conn_request_header_parse(nxt_task_t *task, void *obj, void *data)
*/
h1p->keepalive = (h1p->parser.version.s.minor != '0');
- ret = nxt_h1p_header_process(h1p, r);
+ ret = nxt_h1p_header_process(task, h1p, r);
if (nxt_fast_path(ret == NXT_OK)) {
@@ -551,7 +543,7 @@ nxt_h1p_conn_request_header_parse(nxt_task_t *task, void *obj, void *data)
break;
}
- (void) nxt_h1p_header_process(h1p, r);
+ (void) nxt_h1p_header_process(task, h1p, r);
error:
@@ -562,8 +554,12 @@ error:
static nxt_int_t
-nxt_h1p_header_process(nxt_h1proto_t *h1p, nxt_http_request_t *r)
+nxt_h1p_header_process(nxt_task_t *task, nxt_h1proto_t *h1p,
+ nxt_http_request_t *r)
{
+ u_char *m;
+ nxt_int_t ret;
+
r->target.start = h1p->parser.target_start;
r->target.length = h1p->parser.target_end - h1p->parser.target_start;
@@ -578,7 +574,46 @@ nxt_h1p_header_process(nxt_h1proto_t *h1p, nxt_http_request_t *r)
r->fields = h1p->parser.fields;
- return nxt_http_fields_process(r->fields, &nxt_h1p_fields_hash, r);
+ ret = nxt_http_fields_process(r->fields, &nxt_h1p_fields_hash, r);
+ if (nxt_slow_path(ret != NXT_OK)) {
+ return ret;
+ }
+
+ if (h1p->connection_upgrade && h1p->upgrade_websocket) {
+ m = h1p->parser.method.start;
+
+ if (nxt_slow_path(h1p->parser.method.length != 3
+ || m[0] != 'G'
+ || m[1] != 'E'
+ || m[2] != 'T'))
+ {
+ nxt_log(task, NXT_LOG_INFO, "h1p upgrade: bad method");
+
+ return NXT_HTTP_BAD_REQUEST;
+ }
+
+ if (nxt_slow_path(h1p->parser.version.s.minor != '1')) {
+ nxt_log(task, NXT_LOG_INFO, "h1p upgrade: bad protocol version");
+
+ return NXT_HTTP_BAD_REQUEST;
+ }
+
+ if (nxt_slow_path(h1p->websocket_key == NULL)) {
+ nxt_log(task, NXT_LOG_INFO, "h1p upgrade: bad or absent websocket key");
+
+ return NXT_HTTP_BAD_REQUEST;
+ }
+
+ if (nxt_slow_path(h1p->websocket_version_ok == 0)) {
+ nxt_log(task, NXT_LOG_INFO, "h1p upgrade: bad or absent websocket version");
+
+ return NXT_HTTP_UPGRADE_REQUIRED;
+ }
+
+ r->websocket_handshake = 1;
+ }
+
+ return ret;
}
@@ -620,12 +655,68 @@ nxt_h1p_header_buffer_test(nxt_task_t *task, nxt_h1proto_t *h1p, nxt_conn_t *c,
static nxt_int_t
nxt_h1p_connection(void *ctx, nxt_http_field_t *field, uintptr_t data)
{
- nxt_http_request_t *r;
+ nxt_http_request_t *r;
+ static const u_char *upgrade = (const u_char *) "upgrade";
r = ctx;
if (field->value_length == 5 && nxt_memcmp(field->value, "close", 5) == 0) {
r->proto.h1->keepalive = 0;
+
+ } else if (field->value_length == 7
+ && nxt_memcasecmp(field->value, upgrade, 7) == 0)
+ {
+ r->proto.h1->connection_upgrade = 1;
+ }
+
+ return NXT_OK;
+}
+
+
+static nxt_int_t
+nxt_h1p_upgrade(void *ctx, nxt_http_field_t *field, uintptr_t data)
+{
+ nxt_http_request_t *r;
+ static const u_char *websocket = (const u_char *) "websocket";
+
+ r = ctx;
+
+ if (field->value_length == 9
+ && nxt_memcasecmp(field->value, websocket, 9) == 0)
+ {
+ r->proto.h1->upgrade_websocket = 1;
+ }
+
+ return NXT_OK;
+}
+
+
+static nxt_int_t
+nxt_h1p_websocket_key(void *ctx, nxt_http_field_t *field, uintptr_t data)
+{
+ nxt_http_request_t *r;
+
+ r = ctx;
+
+ if (field->value_length == 24) {
+ r->proto.h1->websocket_key = field;
+ }
+
+ return NXT_OK;
+}
+
+
+static nxt_int_t
+nxt_h1p_websocket_version(void *ctx, nxt_http_field_t *field, uintptr_t data)
+{
+ nxt_http_request_t *r;
+
+ r = ctx;
+
+ if (field->value_length == 2
+ && field->value[0] == '1' && field->value[1] == '3')
+ {
+ r->proto.h1->websocket_version_ok = 1;
}
return NXT_OK;
@@ -727,6 +818,7 @@ nxt_h1p_request_body_read(nxt_task_t *task, nxt_http_request_t *r)
if (size != 0) {
in->next = h1p->buffers;
h1p->buffers = in;
+ h1p->nbuffers++;
c = h1p->conn;
c->read = b;
@@ -805,6 +897,15 @@ nxt_h1p_request_local_addr(nxt_task_t *task, nxt_http_request_t *r)
}
+#define NXT_HTTP_LAST_INFORMATIONAL \
+ (NXT_HTTP_CONTINUE + nxt_nitems(nxt_http_informational) - 1)
+
+static const nxt_str_t nxt_http_informational[] = {
+ nxt_string("HTTP/1.1 100 Continue\r\n"),
+ nxt_string("HTTP/1.1 101 Switching Protocols\r\n"),
+};
+
+
#define NXT_HTTP_LAST_SUCCESS \
(NXT_HTTP_OK + nxt_nitems(nxt_http_success) - 1)
@@ -861,7 +962,7 @@ static const nxt_str_t nxt_http_client_error[] = {
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 426 Upgrade Required\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"),
@@ -911,10 +1012,14 @@ nxt_h1p_request_header_send(nxt_task_t *task, nxt_http_request_t *r)
u_char buf[UNKNOWN_STATUS_LENGTH];
static const char chunked[] = "Transfer-Encoding: chunked\r\n";
+ static const char websocket_version[] = "Sec-WebSocket-Version: 13\r\n";
- static const nxt_str_t connection[2] = {
+ static const nxt_str_t connection[3] = {
nxt_string("Connection: close\r\n"),
nxt_string("Connection: keep-alive\r\n"),
+ nxt_string("Upgrade: websocket\r\n"
+ "Connection: Upgrade\r\n"
+ "Sec-WebSocket-Accept: "),
};
nxt_debug(task, "h1p request header send");
@@ -923,7 +1028,10 @@ nxt_h1p_request_header_send(nxt_task_t *task, nxt_http_request_t *r)
h1p = r->proto.h1;
n = r->status;
- if (n >= NXT_HTTP_OK && n <= NXT_HTTP_LAST_SUCCESS) {
+ if (n >= NXT_HTTP_CONTINUE && n <= NXT_HTTP_LAST_INFORMATIONAL) {
+ status = &nxt_http_informational[n - NXT_HTTP_CONTINUE];
+
+ } else 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
@@ -955,27 +1063,41 @@ nxt_h1p_request_header_send(nxt_task_t *task, nxt_http_request_t *r)
/* Trailing CRLF at the end of header. */
size += nxt_length("\r\n");
- http11 = (h1p->parser.version.s.minor != '0');
+ conn = -1;
+
+ if (r->websocket_handshake && n == NXT_HTTP_SWITCHING_PROTOCOLS) {
+ h1p->websocket = 1;
+ h1p->keepalive = 0;
+ conn = 2;
+ size += NXT_WEBSOCKET_ACCEPT_SIZE + 2;
+
+ } else {
+ http11 = (h1p->parser.version.s.minor != '0');
+
+ if (r->resp.content_length == NULL || r->resp.content_length->skip) {
- if (r->resp.content_length == NULL || r->resp.content_length->skip) {
+ if (http11) {
+ if (n != NXT_HTTP_NOT_MODIFIED
+ && n != NXT_HTTP_NO_CONTENT
+ && !h1p->websocket)
+ {
+ h1p->chunked = 1;
+ size += nxt_length(chunked);
+ /* Trailing CRLF will be added by the first chunk header. */
+ size -= nxt_length("\r\n");
+ }
- if (http11) {
- if (n != NXT_HTTP_NOT_MODIFIED && n != NXT_HTTP_NO_CONTENT) {
- h1p->chunked = 1;
- size += nxt_length(chunked);
- /* Trailing CRLF will be added by the first chunk header. */
- size -= nxt_length("\r\n");
+ } else {
+ h1p->keepalive = 0;
}
+ }
- } else {
- h1p->keepalive = 0;
+ if (http11 ^ h1p->keepalive) {
+ conn = h1p->keepalive;
}
}
- conn = -1;
-
- if (http11 ^ h1p->keepalive) {
- conn = h1p->keepalive;
+ if (conn >= 0) {
size += connection[conn].length;
}
@@ -988,15 +1110,17 @@ nxt_h1p_request_header_send(nxt_task_t *task, nxt_http_request_t *r)
} nxt_list_loop;
+ if (nxt_slow_path(n == NXT_HTTP_UPGRADE_REQUIRED)) {
+ size += nxt_length(websocket_version);
+ }
+
header = nxt_http_buf_mem(task, r, size);
if (nxt_slow_path(header == NULL)) {
nxt_h1p_request_error(task, h1p, r);
return;
}
- p = header->mem.free;
-
- p = nxt_cpymem(p, status->start, status->length);
+ p = nxt_cpymem(header->mem.free, status->start, status->length);
nxt_list_each(field, r->resp.fields) {
@@ -1013,6 +1137,17 @@ nxt_h1p_request_header_send(nxt_task_t *task, nxt_http_request_t *r)
p = nxt_cpymem(p, connection[conn].start, connection[conn].length);
}
+ if (h1p->websocket) {
+ nxt_websocket_accept(p, h1p->websocket_key->value);
+ p += NXT_WEBSOCKET_ACCEPT_SIZE;
+
+ *p++ = '\r'; *p++ = '\n';
+ }
+
+ if (nxt_slow_path(n == NXT_HTTP_UPGRADE_REQUIRED)) {
+ p = nxt_cpymem(p, websocket_version, nxt_length(websocket_version));
+ }
+
if (h1p->chunked) {
p = nxt_cpymem(p, chunked, nxt_length(chunked));
/* Trailing CRLF will be added by the first chunk header. */
@@ -1031,6 +1166,58 @@ nxt_h1p_request_header_send(nxt_task_t *task, nxt_http_request_t *r)
c->write_state = &nxt_h1p_request_send_state;
nxt_conn_write(task->thread->engine, c);
+
+ if (h1p->websocket) {
+ nxt_h1p_websocket_first_frame_start(task, r, c->read);
+ }
+}
+
+
+void
+nxt_h1p_complete_buffers(nxt_task_t *task, nxt_h1proto_t *h1p)
+{
+ size_t size;
+ nxt_buf_t *b, *in, *next;
+ nxt_conn_t *c;
+
+ nxt_debug(task, "h1p complete buffers");
+
+ b = h1p->buffers;
+ c = h1p->conn;
+ in = c->read;
+
+ if (b != NULL) {
+ if (in == NULL) {
+ /* A request with large body. */
+ in = b;
+ c->read = in;
+
+ b = in->next;
+ in->next = NULL;
+ }
+
+ while (b != NULL) {
+ next = b->next;
+
+ nxt_work_queue_add(&task->thread->engine->fast_work_queue,
+ b->completion_handler, task, b, b->parent);
+
+ b = next;
+ }
+
+ h1p->buffers = NULL;
+ h1p->nbuffers = 0;
+ }
+
+ if (in != NULL) {
+ size = nxt_buf_mem_used_size(&in->mem);
+
+ if (size == 0) {
+ nxt_mp_free(c->mem_pool, in);
+
+ c->read = NULL;
+ }
+ }
}
@@ -1181,8 +1368,13 @@ nxt_h1p_conn_request_error(nxt_task_t *task, void *obj, void *data)
r = h1p->request;
+ if (nxt_slow_path(r == NULL)) {
+ nxt_h1p_shutdown(task, h1p->conn);
+ return;
+ }
+
if (r->fields == NULL) {
- (void) nxt_h1p_header_process(h1p, r);
+ (void) nxt_h1p_header_process(task, h1p, r);
}
if (r->status == 0) {
@@ -1218,7 +1410,7 @@ nxt_h1p_conn_request_timeout(nxt_task_t *task, void *obj, void *data)
r = h1p->request;
if (r->fields == NULL) {
- (void) nxt_h1p_header_process(h1p, r);
+ (void) nxt_h1p_header_process(task, h1p, r);
}
nxt_http_request_error(task, r, NXT_HTTP_REQUEST_TIMEOUT);
@@ -1244,7 +1436,7 @@ nxt_h1p_conn_request_send_timeout(nxt_task_t *task, void *obj, void *data)
}
-static nxt_msec_t
+nxt_msec_t
nxt_h1p_conn_request_timer_value(nxt_conn_t *c, uintptr_t data)
{
nxt_h1proto_t *h1p;
@@ -1351,7 +1543,7 @@ 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_buf_t *in;
nxt_debug(task, "h1p keepalive");
@@ -1359,40 +1551,22 @@ nxt_h1p_keepalive(nxt_task_t *task, nxt_h1proto_t *h1p, nxt_conn_t *c)
nxt_conn_tcp_nodelay_on(task, c);
}
- b = h1p->buffers;
+ nxt_h1p_complete_buffers(task, h1p);
+
+ in = c->read;
nxt_memzero(h1p, offsetof(nxt_h1proto_t, conn));
c->sent = 0;
- in = c->read;
-
if (in == NULL) {
- /* A request with large body. */
- in = b;
- c->read = in;
-
- b = in->next;
- in->next = NULL;
- }
-
- while (b != NULL) {
- next = b->next;
- nxt_mp_free(c->mem_pool, b);
- b = next;
- }
-
- size = nxt_buf_mem_used_size(&in->mem);
-
- if (size == 0) {
- nxt_mp_free(c->mem_pool, in);
-
- c->read = NULL;
c->read_state = &nxt_h1p_keepalive_state;
nxt_conn_read(task->thread->engine, c);
} else {
+ size = nxt_buf_mem_used_size(&in->mem);
+
nxt_debug(task, "h1p pipelining");
nxt_memmove(in->mem.start, in->mem.pos, size);
@@ -1421,7 +1595,7 @@ static const nxt_conn_state_t nxt_h1p_keepalive_state
};
-static const nxt_conn_state_t nxt_h1p_idle_close_state
+const nxt_conn_state_t nxt_h1p_idle_close_state
nxt_aligned(64) =
{
.close_handler = nxt_h1p_idle_close,
@@ -1564,8 +1738,33 @@ 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)
{
+ nxt_timer_t *timer;
+ nxt_h1proto_t *h1p;
+
nxt_debug(task, "h1p shutdown");
+ h1p = c->socket.data;
+
+ if (nxt_slow_path(h1p != NULL && h1p->websocket_timer != NULL)) {
+ timer = &h1p->websocket_timer->timer;
+
+ if (timer->handler != nxt_h1p_conn_ws_shutdown) {
+ timer->handler = nxt_h1p_conn_ws_shutdown;
+ nxt_timer_add(task->thread->engine, timer, 0);
+
+ } else {
+ nxt_debug(task, "h1p already scheduled ws shutdown");
+ }
+
+ } else {
+ nxt_h1p_shutdown_(task, c);
+ }
+}
+
+
+static void
+nxt_h1p_shutdown_(nxt_task_t *task, nxt_conn_t *c)
+{
c->socket.data = NULL;
#if (NXT_TLS)
@@ -1597,6 +1796,21 @@ static const nxt_conn_state_t nxt_h1p_shutdown_state
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)
{
nxt_conn_t *c;
diff --git a/src/nxt_h1proto.h b/src/nxt_h1proto.h
new file mode 100644
index 00000000..c6d3bd53
--- /dev/null
+++ b/src/nxt_h1proto.h
@@ -0,0 +1,48 @@
+
+/*
+ * Copyright (C) NGINX, Inc.
+ */
+
+#ifndef _NXT_H1PROTO_H_INCLUDED_
+#define _NXT_H1PROTO_H_INCLUDED_
+
+
+#include <nxt_main.h>
+#include <nxt_http_parse.h>
+#include <nxt_http.h>
+#include <nxt_router.h>
+
+
+typedef struct nxt_h1p_websocket_timer_s nxt_h1p_websocket_timer_t;
+
+
+struct nxt_h1proto_s {
+ nxt_http_request_parse_t parser;
+
+ uint8_t nbuffers;
+ uint8_t keepalive; /* 1 bit */
+ uint8_t chunked; /* 1 bit */
+ uint8_t websocket; /* 1 bit */
+ uint8_t connection_upgrade; /* 1 bit */
+ uint8_t upgrade_websocket; /* 1 bit */
+ uint8_t websocket_version_ok; /* 1 bit */
+ nxt_http_te_t transfer_encoding:8; /* 2 bits */
+
+ uint8_t websocket_cont_expected; /* 1 bit */
+ uint8_t websocket_closed; /* 1 bit */
+
+ uint32_t header_size;
+
+ nxt_http_field_t *websocket_key;
+ nxt_h1p_websocket_timer_t *websocket_timer;
+
+ nxt_http_request_t *request;
+ nxt_buf_t *buffers;
+ /*
+ * All fields before the conn field will
+ * be zeroed in a keep-alive connection.
+ */
+ nxt_conn_t *conn;
+};
+
+#endif /* _NXT_H1PROTO_H_INCLUDED_ */
diff --git a/src/nxt_h1proto_websocket.c b/src/nxt_h1proto_websocket.c
new file mode 100644
index 00000000..dd9b6848
--- /dev/null
+++ b/src/nxt_h1proto_websocket.c
@@ -0,0 +1,719 @@
+
+/*
+ * Copyright (C) NGINX, Inc.
+ */
+
+#include <nxt_main.h>
+#include <nxt_router.h>
+#include <nxt_http.h>
+#include <nxt_h1proto.h>
+#include <nxt_websocket.h>
+#include <nxt_websocket_header.h>
+
+typedef struct {
+ uint16_t code;
+ uint8_t args;
+ nxt_str_t desc;
+} nxt_ws_error_t;
+
+static void nxt_h1p_conn_ws_keepalive(nxt_task_t *task, void *obj, void *data);
+static void nxt_h1p_conn_ws_frame_header_read(nxt_task_t *task, void *obj,
+ void *data);
+static void nxt_h1p_conn_ws_keepalive_disable(nxt_task_t *task,
+ nxt_h1proto_t *h1p);
+static void nxt_h1p_conn_ws_keepalive_enable(nxt_task_t *task,
+ nxt_h1proto_t *h1p);
+static void nxt_h1p_conn_ws_frame_process(nxt_task_t *task, nxt_conn_t *c,
+ nxt_h1proto_t *h1p, nxt_websocket_header_t *wsh);
+static void nxt_h1p_conn_ws_error(nxt_task_t *task, void *obj, void *data);
+static ssize_t nxt_h1p_ws_io_read_handler(nxt_conn_t *c);
+static void nxt_h1p_conn_ws_timeout(nxt_task_t *task, void *obj, void *data);
+static void nxt_h1p_conn_ws_frame_payload_read(nxt_task_t *task, void *obj,
+ void *data);
+static void hxt_h1p_send_ws_error(nxt_task_t *task, nxt_http_request_t *r,
+ const nxt_ws_error_t *err, ...);
+static void nxt_h1p_conn_ws_error_sent(nxt_task_t *task, void *obj, void *data);
+static void nxt_h1p_conn_ws_pong(nxt_task_t *task, void *obj, void *data);
+
+static const nxt_conn_state_t nxt_h1p_read_ws_frame_header_state;
+static const nxt_conn_state_t nxt_h1p_read_ws_frame_payload_state;
+
+static const nxt_ws_error_t nxt_ws_err_out_of_memory = {
+ NXT_WEBSOCKET_CR_INTERNAL_SERVER_ERROR,
+ 0, nxt_string("Out of memory") };
+static const nxt_ws_error_t nxt_ws_err_too_big = {
+ NXT_WEBSOCKET_CR_MESSAGE_TOO_BIG,
+ 1, nxt_string("Message too big: %uL bytes") };
+static const nxt_ws_error_t nxt_ws_err_invalid_close_code = {
+ NXT_WEBSOCKET_CR_PROTOCOL_ERROR,
+ 1, nxt_string("Close code %ud is not valid") };
+static const nxt_ws_error_t nxt_ws_err_going_away = {
+ NXT_WEBSOCKET_CR_GOING_AWAY,
+ 0, nxt_string("Remote peer is going away") };
+static const nxt_ws_error_t nxt_ws_err_not_masked = {
+ NXT_WEBSOCKET_CR_PROTOCOL_ERROR,
+ 0, nxt_string("Not masked client frame") };
+static const nxt_ws_error_t nxt_ws_err_ctrl_fragmented = {
+ NXT_WEBSOCKET_CR_PROTOCOL_ERROR,
+ 0, nxt_string("Fragmented control frame") };
+static const nxt_ws_error_t nxt_ws_err_ctrl_too_big = {
+ NXT_WEBSOCKET_CR_PROTOCOL_ERROR,
+ 1, nxt_string("Control frame too big: %uL bytes") };
+static const nxt_ws_error_t nxt_ws_err_invalid_close_len = {
+ NXT_WEBSOCKET_CR_PROTOCOL_ERROR,
+ 0, nxt_string("Close frame payload length cannot be 1") };
+static const nxt_ws_error_t nxt_ws_err_invalid_opcode = {
+ NXT_WEBSOCKET_CR_PROTOCOL_ERROR,
+ 1, nxt_string("Unrecognized opcode %ud") };
+static const nxt_ws_error_t nxt_ws_err_cont_expected = {
+ NXT_WEBSOCKET_CR_PROTOCOL_ERROR,
+ 1, nxt_string("Continuation expected, but %ud opcode received") };
+
+void
+nxt_h1p_websocket_first_frame_start(nxt_task_t *task, nxt_http_request_t *r,
+ nxt_buf_t *ws_frame)
+{
+ nxt_conn_t *c;
+ nxt_timer_t *timer;
+ nxt_h1proto_t *h1p;
+ nxt_socket_conf_joint_t *joint;
+
+ nxt_debug(task, "h1p ws first frame start");
+
+ h1p = r->proto.h1;
+ c = h1p->conn;
+
+ if (!c->tcp_nodelay) {
+ nxt_conn_tcp_nodelay_on(task, c);
+ }
+
+ joint = c->listen->socket.data;
+
+ if (nxt_slow_path(joint != NULL
+ && joint->socket_conf->websocket_conf.keepalive_interval != 0))
+ {
+ h1p->websocket_timer = nxt_mp_zget(c->mem_pool,
+ sizeof(nxt_h1p_websocket_timer_t));
+ if (nxt_slow_path(h1p->websocket_timer == NULL)) {
+ hxt_h1p_send_ws_error(task, r, &nxt_ws_err_out_of_memory);
+ return;
+ }
+
+ h1p->websocket_timer->keepalive_interval =
+ joint->socket_conf->websocket_conf.keepalive_interval;
+ h1p->websocket_timer->h1p = h1p;
+
+ timer = &h1p->websocket_timer->timer;
+ timer->task = &c->task;
+ timer->work_queue = &task->thread->engine->fast_work_queue;
+ timer->log = &c->log;
+ timer->bias = NXT_TIMER_DEFAULT_BIAS;
+ timer->handler = nxt_h1p_conn_ws_keepalive;
+ }
+
+ nxt_h1p_websocket_frame_start(task, r, ws_frame);
+}
+
+
+void
+nxt_h1p_websocket_frame_start(nxt_task_t *task, nxt_http_request_t *r,
+ nxt_buf_t *ws_frame)
+{
+ size_t size;
+ nxt_buf_t *in;
+ nxt_conn_t *c;
+ nxt_h1proto_t *h1p;
+
+ nxt_debug(task, "h1p ws frame start");
+
+ h1p = r->proto.h1;
+
+ if (nxt_slow_path(h1p->websocket_closed)) {
+ return;
+ }
+
+ c = h1p->conn;
+ c->read = ws_frame;
+
+ nxt_h1p_complete_buffers(task, h1p);
+
+ in = c->read;
+ c->read_state = &nxt_h1p_read_ws_frame_header_state;
+
+ if (in == NULL) {
+ nxt_conn_read(task->thread->engine, c);
+ nxt_h1p_conn_ws_keepalive_enable(task, h1p);
+
+ } else {
+ size = nxt_buf_mem_used_size(&in->mem);
+
+ nxt_debug(task, "h1p read client ws frame");
+
+ nxt_memmove(in->mem.start, in->mem.pos, size);
+
+ in->mem.pos = in->mem.start;
+ in->mem.free = in->mem.start + size;
+
+ nxt_h1p_conn_ws_frame_header_read(task, c, h1p);
+ }
+}
+
+
+static void
+nxt_h1p_conn_ws_keepalive(nxt_task_t *task, void *obj, void *data)
+{
+ nxt_buf_t *out;
+ nxt_timer_t *timer;
+ nxt_h1proto_t *h1p;
+ nxt_http_request_t *r;
+ nxt_websocket_header_t *wsh;
+ nxt_h1p_websocket_timer_t *ws_timer;
+
+ nxt_debug(task, "h1p conn ws keepalive");
+
+ timer = obj;
+ ws_timer = nxt_timer_data(timer, nxt_h1p_websocket_timer_t, timer);
+ h1p = ws_timer->h1p;
+
+ r = h1p->request;
+ if (nxt_slow_path(r == NULL)) {
+ return;
+ }
+
+ out = nxt_http_buf_mem(task, r, 2);
+ if (nxt_slow_path(out == NULL)) {
+ nxt_http_request_error_handler(task, r, r->proto.any);
+ return;
+ }
+
+ out->mem.start[0] = 0;
+ out->mem.start[1] = 0;
+
+ wsh = (nxt_websocket_header_t *) out->mem.start;
+ out->mem.free = nxt_websocket_frame_init(wsh, 0);
+
+ wsh->fin = 1;
+ wsh->opcode = NXT_WEBSOCKET_OP_PING;
+
+ nxt_http_request_send(task, r, out);
+}
+
+
+static const nxt_conn_state_t nxt_h1p_read_ws_frame_header_state
+ nxt_aligned(64) =
+{
+ .ready_handler = nxt_h1p_conn_ws_frame_header_read,
+ .close_handler = nxt_h1p_conn_ws_error,
+ .error_handler = nxt_h1p_conn_ws_error,
+
+ .io_read_handler = nxt_h1p_ws_io_read_handler,
+
+ .timer_handler = nxt_h1p_conn_ws_timeout,
+ .timer_value = nxt_h1p_conn_request_timer_value,
+ .timer_data = offsetof(nxt_socket_conf_t, websocket_conf.read_timeout),
+ .timer_autoreset = 1,
+};
+
+
+static void
+nxt_h1p_conn_ws_frame_header_read(nxt_task_t *task, void *obj, void *data)
+{
+ size_t size, hsize, frame_size, max_frame_size;
+ uint64_t payload_len;
+ nxt_conn_t *c;
+ nxt_h1proto_t *h1p;
+ nxt_http_request_t *r;
+ nxt_event_engine_t *engine;
+ nxt_websocket_header_t *wsh;
+ nxt_socket_conf_joint_t *joint;
+
+ c = obj;
+ h1p = data;
+
+ nxt_h1p_conn_ws_keepalive_disable(task, h1p);
+
+ size = nxt_buf_mem_used_size(&c->read->mem);
+
+ engine = task->thread->engine;
+
+ if (size < 2) {
+ nxt_debug(task, "h1p conn ws frame header read %z", size);
+
+ nxt_conn_read(engine, c);
+ nxt_h1p_conn_ws_keepalive_enable(task, h1p);
+
+ return;
+ }
+
+ wsh = (nxt_websocket_header_t *) c->read->mem.pos;
+
+ hsize = nxt_websocket_frame_header_size(wsh);
+
+ if (size < hsize) {
+ nxt_debug(task, "h1p conn ws frame header read %z < %z", size, hsize);
+
+ nxt_conn_read(engine, c);
+ nxt_h1p_conn_ws_keepalive_enable(task, h1p);
+
+ return;
+ }
+
+ r = h1p->request;
+ if (nxt_slow_path(r == NULL)) {
+ return;
+ }
+
+ r->ws_frame = c->read;
+
+ joint = c->listen->socket.data;
+
+ if (nxt_slow_path(joint == NULL)) {
+ /*
+ * Listening socket had been closed while
+ * connection was in keep-alive state.
+ */
+ c->read_state = &nxt_h1p_idle_close_state;
+ return;
+ }
+
+ if (nxt_slow_path(wsh->mask == 0)) {
+ hxt_h1p_send_ws_error(task, r, &nxt_ws_err_not_masked);
+ return;
+ }
+
+ if ((wsh->opcode & NXT_WEBSOCKET_OP_CTRL) != 0) {
+ if (nxt_slow_path(wsh->fin == 0)) {
+ hxt_h1p_send_ws_error(task, r, &nxt_ws_err_ctrl_fragmented);
+ return;
+ }
+
+ if (nxt_slow_path(wsh->opcode != NXT_WEBSOCKET_OP_PING
+ && wsh->opcode != NXT_WEBSOCKET_OP_PONG
+ && wsh->opcode != NXT_WEBSOCKET_OP_CLOSE))
+ {
+ hxt_h1p_send_ws_error(task, r, &nxt_ws_err_invalid_opcode,
+ wsh->opcode);
+ return;
+ }
+
+ if (nxt_slow_path(wsh->payload_len > 125)) {
+ hxt_h1p_send_ws_error(task, r, &nxt_ws_err_ctrl_too_big,
+ nxt_websocket_frame_payload_len(wsh));
+ return;
+ }
+
+ if (nxt_slow_path(wsh->opcode == NXT_WEBSOCKET_OP_CLOSE
+ && wsh->payload_len == 1))
+ {
+ hxt_h1p_send_ws_error(task, r, &nxt_ws_err_invalid_close_len);
+ return;
+ }
+
+ } else {
+ if (h1p->websocket_cont_expected) {
+ if (nxt_slow_path(wsh->opcode != NXT_WEBSOCKET_OP_CONT)) {
+ hxt_h1p_send_ws_error(task, r, &nxt_ws_err_cont_expected,
+ wsh->opcode);
+ return;
+ }
+
+ } else {
+ if (nxt_slow_path(wsh->opcode != NXT_WEBSOCKET_OP_BINARY
+ && wsh->opcode != NXT_WEBSOCKET_OP_TEXT))
+ {
+ hxt_h1p_send_ws_error(task, r, &nxt_ws_err_invalid_opcode,
+ wsh->opcode);
+ return;
+ }
+ }
+
+ h1p->websocket_cont_expected = !wsh->fin;
+ }
+
+ max_frame_size = joint->socket_conf->websocket_conf.max_frame_size;
+
+ payload_len = nxt_websocket_frame_payload_len(wsh);
+
+ if (nxt_slow_path(hsize > max_frame_size
+ || payload_len > (max_frame_size - hsize)))
+ {
+ hxt_h1p_send_ws_error(task, r, &nxt_ws_err_too_big, payload_len);
+ return;
+ }
+
+ c->read_state = &nxt_h1p_read_ws_frame_payload_state;
+
+ frame_size = payload_len + hsize;
+
+ nxt_debug(task, "h1p conn ws frame header read: %z, %z", size, frame_size);
+
+ if (frame_size <= size) {
+ nxt_h1p_conn_ws_frame_process(task, c, h1p, wsh);
+
+ return;
+ }
+
+ if (frame_size < (size_t) nxt_buf_mem_size(&c->read->mem)) {
+ c->read->mem.end = c->read->mem.start + frame_size;
+
+ } else {
+ nxt_buf_t *b = nxt_buf_mem_alloc(c->mem_pool, frame_size - size, 0);
+
+ c->read->next = b;
+ c->read = b;
+ }
+
+ nxt_conn_read(engine, c);
+ nxt_h1p_conn_ws_keepalive_enable(task, h1p);
+}
+
+
+static void
+nxt_h1p_conn_ws_keepalive_disable(nxt_task_t *task, nxt_h1proto_t *h1p)
+{
+ nxt_timer_t *timer;
+
+ if (h1p->websocket_timer == NULL) {
+ return;
+ }
+
+ timer = &h1p->websocket_timer->timer;
+
+ if (nxt_slow_path(timer->handler != nxt_h1p_conn_ws_keepalive)) {
+ nxt_debug(task, "h1p ws keepalive disable: scheduled ws shutdown");
+ return;
+ }
+
+ nxt_timer_disable(task->thread->engine, timer);
+}
+
+
+static void
+nxt_h1p_conn_ws_keepalive_enable(nxt_task_t *task, nxt_h1proto_t *h1p)
+{
+ nxt_timer_t *timer;
+
+ if (h1p->websocket_timer == NULL) {
+ return;
+ }
+
+ timer = &h1p->websocket_timer->timer;
+
+ if (nxt_slow_path(timer->handler != nxt_h1p_conn_ws_keepalive)) {
+ nxt_debug(task, "h1p ws keepalive enable: scheduled ws shutdown");
+ return;
+ }
+
+ nxt_timer_add(task->thread->engine, timer,
+ h1p->websocket_timer->keepalive_interval);
+}
+
+
+static void
+nxt_h1p_conn_ws_frame_process(nxt_task_t *task, nxt_conn_t *c,
+ nxt_h1proto_t *h1p, nxt_websocket_header_t *wsh)
+{
+ size_t hsize;
+ uint8_t *p, *mask;
+ uint16_t code;
+ nxt_http_request_t *r;
+ nxt_event_engine_t *engine;
+
+ engine = task->thread->engine;
+ r = h1p->request;
+
+ c->read = NULL;
+
+ if (nxt_slow_path(wsh->opcode == NXT_WEBSOCKET_OP_PING)) {
+ nxt_work_queue_add(&engine->fast_work_queue, nxt_h1p_conn_ws_pong,
+ task, r, NULL);
+ return;
+ }
+
+ if (nxt_slow_path(wsh->opcode == NXT_WEBSOCKET_OP_CLOSE)) {
+ if (wsh->payload_len >= 2) {
+ hsize = nxt_websocket_frame_header_size(wsh);
+ mask = nxt_pointer_to(wsh, hsize - 4);
+ p = nxt_pointer_to(wsh, hsize);
+
+ code = ((p[0] ^ mask[0]) << 8) + (p[1] ^ mask[1]);
+
+ if (nxt_slow_path(code < 1000 || code >= 5000
+ || (code > 1003 && code < 1007)
+ || (code > 1014 && code < 3000)))
+ {
+ hxt_h1p_send_ws_error(task, r, &nxt_ws_err_invalid_close_code,
+ code);
+ return;
+ }
+ }
+
+ h1p->websocket_closed = 1;
+ }
+
+ nxt_work_queue_add(&engine->fast_work_queue, r->state->ready_handler,
+ task, r, NULL);
+}
+
+
+static void
+nxt_h1p_conn_ws_error(nxt_task_t *task, void *obj, void *data)
+{
+ nxt_h1proto_t *h1p;
+ nxt_http_request_t *r;
+
+ h1p = data;
+
+ nxt_debug(task, "h1p conn ws error");
+
+ r = h1p->request;
+
+ h1p->keepalive = 0;
+
+ if (nxt_fast_path(r != NULL)) {
+ r->state->error_handler(task, r, h1p);
+ }
+}
+
+
+static ssize_t
+nxt_h1p_ws_io_read_handler(nxt_conn_t *c)
+{
+ size_t size;
+ ssize_t n;
+ nxt_buf_t *b;
+
+ b = c->read;
+
+ if (b == NULL) {
+ /* Enough for control frame. */
+ size = 10 + 125;
+
+ b = nxt_buf_mem_alloc(c->mem_pool, size, 0);
+ 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_mp_free(c->mem_pool, b);
+ }
+
+ return n;
+}
+
+
+static void
+nxt_h1p_conn_ws_timeout(nxt_task_t *task, void *obj, void *data)
+{
+ nxt_conn_t *c;
+ nxt_timer_t *timer;
+ nxt_h1proto_t *h1p;
+ nxt_http_request_t *r;
+
+ timer = obj;
+
+ nxt_debug(task, "h1p conn ws timeout");
+
+ c = nxt_read_timer_conn(timer);
+ c->block_read = 1;
+ /*
+ * Disable SO_LINGER off during socket closing
+ * to send "408 Request Timeout" error response.
+ */
+ c->socket.timedout = 0;
+
+ h1p = c->socket.data;
+ h1p->keepalive = 0;
+
+ r = h1p->request;
+ if (nxt_slow_path(r == NULL)) {
+ return;
+ }
+
+ hxt_h1p_send_ws_error(task, r, &nxt_ws_err_going_away);
+}
+
+
+static const nxt_conn_state_t nxt_h1p_read_ws_frame_payload_state
+ nxt_aligned(64) =
+{
+ .ready_handler = nxt_h1p_conn_ws_frame_payload_read,
+ .close_handler = nxt_h1p_conn_ws_error,
+ .error_handler = nxt_h1p_conn_ws_error,
+
+ .timer_handler = nxt_h1p_conn_ws_timeout,
+ .timer_value = nxt_h1p_conn_request_timer_value,
+ .timer_data = offsetof(nxt_socket_conf_t, websocket_conf.read_timeout),
+ .timer_autoreset = 1,
+};
+
+
+static void
+nxt_h1p_conn_ws_frame_payload_read(nxt_task_t *task, void *obj, void *data)
+{
+ nxt_conn_t *c;
+ nxt_h1proto_t *h1p;
+ nxt_http_request_t *r;
+ nxt_event_engine_t *engine;
+ nxt_websocket_header_t *wsh;
+
+ c = obj;
+ h1p = data;
+
+ nxt_h1p_conn_ws_keepalive_disable(task, h1p);
+
+ nxt_debug(task, "h1p conn ws frame read");
+
+ if (nxt_buf_mem_free_size(&c->read->mem) == 0) {
+ r = h1p->request;
+ if (nxt_slow_path(r == NULL)) {
+ return;
+ }
+
+ wsh = (nxt_websocket_header_t *) r->ws_frame->mem.pos;
+
+ nxt_h1p_conn_ws_frame_process(task, c, h1p, wsh);
+
+ return;
+ }
+
+ engine = task->thread->engine;
+
+ nxt_conn_read(engine, c);
+ nxt_h1p_conn_ws_keepalive_enable(task, h1p);
+}
+
+
+static void
+hxt_h1p_send_ws_error(nxt_task_t *task, nxt_http_request_t *r,
+ const nxt_ws_error_t *err, ...)
+{
+ u_char *p;
+ va_list args;
+ nxt_buf_t *out;
+ nxt_str_t desc;
+ nxt_websocket_header_t *wsh;
+ u_char buf[125];
+
+ if (nxt_slow_path(err->args)) {
+ va_start(args, err);
+ p = nxt_vsprintf(buf, buf + sizeof(buf), (char *) err->desc.start,
+ args);
+ va_end(args);
+
+ desc.start = buf;
+ desc.length = p - buf;
+
+ } else {
+ desc = err->desc;
+ }
+
+ nxt_log(task, NXT_LOG_INFO, "websocket error %d: %V", err->code, &desc);
+
+ out = nxt_http_buf_mem(task, r, 2 + sizeof(err->code) + desc.length);
+ if (nxt_slow_path(out == NULL)) {
+ nxt_http_request_error_handler(task, r, r->proto.any);
+ return;
+ }
+
+ out->mem.start[0] = 0;
+ out->mem.start[1] = 0;
+
+ wsh = (nxt_websocket_header_t *) out->mem.start;
+ p = nxt_websocket_frame_init(wsh, sizeof(err->code) + desc.length);
+
+ wsh->fin = 1;
+ wsh->opcode = NXT_WEBSOCKET_OP_CLOSE;
+
+ *p++ = (err->code >> 8) & 0xFF;
+ *p++ = err->code & 0xFF;
+
+ out->mem.free = nxt_cpymem(p, desc.start, desc.length);
+ out->next = nxt_http_buf_last(r);
+
+ if (out->next != NULL) {
+ out->next->completion_handler = nxt_h1p_conn_ws_error_sent;
+ }
+
+ nxt_http_request_send(task, r, out);
+}
+
+
+static void
+nxt_h1p_conn_ws_error_sent(nxt_task_t *task, void *obj, void *data)
+{
+ nxt_http_request_t *r;
+
+ r = data;
+
+ nxt_debug(task, "h1p conn ws error sent");
+
+ r->state->error_handler(task, r, r->proto.any);
+}
+
+
+static void
+nxt_h1p_conn_ws_pong(nxt_task_t *task, void *obj, void *data)
+{
+ uint8_t payload_len, i;
+ nxt_buf_t *b, *out, *next;
+ nxt_http_request_t *r;
+ nxt_websocket_header_t *wsh;
+ uint8_t mask[4];
+
+ nxt_debug(task, "h1p conn ws pong");
+
+ r = obj;
+ b = r->ws_frame;
+
+ wsh = (nxt_websocket_header_t *) b->mem.pos;
+ payload_len = wsh->payload_len;
+
+ b->mem.pos += 2;
+
+ nxt_memcpy(mask, b->mem.pos, 4);
+
+ b->mem.pos += 4;
+
+ out = nxt_http_buf_mem(task, r, 2 + payload_len);
+ if (nxt_slow_path(out == NULL)) {
+ nxt_http_request_error_handler(task, r, r->proto.any);
+ return;
+ }
+
+ out->mem.start[0] = 0;
+ out->mem.start[1] = 0;
+
+ wsh = (nxt_websocket_header_t *) out->mem.start;
+ out->mem.free = nxt_websocket_frame_init(wsh, payload_len);
+
+ wsh->fin = 1;
+ wsh->opcode = NXT_WEBSOCKET_OP_PONG;
+
+ for (i = 0; i < payload_len; i++) {
+ while (nxt_buf_mem_used_size(&b->mem) == 0) {
+ next = b->next;
+
+ nxt_work_queue_add(&task->thread->engine->fast_work_queue,
+ b->completion_handler, task, b, b->parent);
+
+ b = next;
+ }
+
+ *out->mem.free++ = *b->mem.pos++ ^ mask[i % 4];
+ }
+
+ r->ws_frame = b;
+
+ nxt_http_request_send(task, r, out);
+
+ nxt_http_request_ws_frame_start(task, r, r->ws_frame);
+}
diff --git a/src/nxt_http.h b/src/nxt_http.h
index c1a230ec..ac1eedcf 100644
--- a/src/nxt_http.h
+++ b/src/nxt_http.h
@@ -11,6 +11,9 @@
typedef enum {
NXT_HTTP_INVALID = 0,
+ NXT_HTTP_CONTINUE = 100,
+ NXT_HTTP_SWITCHING_PROTOCOLS = 101,
+
NXT_HTTP_OK = 200,
NXT_HTTP_NO_CONTENT = 204,
@@ -26,6 +29,7 @@ typedef enum {
NXT_HTTP_LENGTH_REQUIRED = 411,
NXT_HTTP_PAYLOAD_TOO_LARGE = 413,
NXT_HTTP_URI_TOO_LONG = 414,
+ NXT_HTTP_UPGRADE_REQUIRED = 426,
NXT_HTTP_REQUEST_HEADER_FIELDS_TOO_LARGE = 431,
NXT_HTTP_TO_HTTPS = 497,
@@ -61,6 +65,13 @@ typedef struct {
typedef struct nxt_h1proto_s nxt_h1proto_t;
+struct nxt_h1p_websocket_timer_s {
+ nxt_timer_t timer;
+ nxt_h1proto_t *h1p;
+ nxt_msec_t keepalive_interval;
+};
+
+
typedef union {
void *any;
nxt_h1proto_t *h1;
@@ -99,6 +110,7 @@ struct nxt_http_request_s {
nxt_mp_t *mem_pool;
nxt_buf_t *body;
+ nxt_buf_t *ws_frame;
nxt_buf_t *out;
const nxt_http_request_state_t *state;
@@ -127,6 +139,8 @@ struct nxt_http_request_s {
nxt_timer_t timer;
void *timer_data;
+ void *req_rpc_data;
+
nxt_buf_t *last;
nxt_http_response_t resp;
@@ -138,6 +152,7 @@ struct nxt_http_request_s {
uint8_t logged; /* 1 bit */
uint8_t header_sent; /* 1 bit */
uint8_t error; /* 1 bit */
+ uint8_t websocket_handshake; /* 1 bit */
};
@@ -166,6 +181,8 @@ typedef struct {
void (*discard)(nxt_task_t *task, nxt_http_request_t *r, nxt_buf_t *last);
void (*close)(nxt_task_t *task, nxt_http_proto_t proto,
nxt_socket_conf_joint_t *joint);
+ void (*ws_frame_start)(nxt_task_t *task, nxt_http_request_t *r,
+ nxt_buf_t *ws_frame);
} nxt_http_proto_table_t;
@@ -179,12 +196,15 @@ void nxt_http_request_error(nxt_task_t *task, nxt_http_request_t *r,
nxt_http_status_t status);
void nxt_http_request_read_body(nxt_task_t *task, nxt_http_request_t *r);
void nxt_http_request_header_send(nxt_task_t *task, nxt_http_request_t *r);
+void nxt_http_request_ws_frame_start(nxt_task_t *task, nxt_http_request_t *r,
+ nxt_buf_t *ws_frame);
void nxt_http_request_send(nxt_task_t *task, nxt_http_request_t *r,
nxt_buf_t *out);
nxt_buf_t *nxt_http_buf_mem(nxt_task_t *task, nxt_http_request_t *r,
size_t size);
nxt_buf_t *nxt_http_buf_last(nxt_http_request_t *r);
void nxt_http_request_error_handler(nxt_task_t *task, void *obj, void *data);
+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);
@@ -212,5 +232,13 @@ extern nxt_lvlhsh_t nxt_response_fields_hash;
extern const nxt_http_proto_table_t nxt_http_proto[];
+void nxt_h1p_websocket_first_frame_start(nxt_task_t *task,
+ nxt_http_request_t *r, nxt_buf_t *ws_frame);
+void nxt_h1p_websocket_frame_start(nxt_task_t *task, nxt_http_request_t *r,
+ nxt_buf_t *ws_frame);
+void nxt_h1p_complete_buffers(nxt_task_t *task, nxt_h1proto_t *h1p);
+nxt_msec_t nxt_h1p_conn_request_timer_value(nxt_conn_t *c, uintptr_t data);
+
+extern const nxt_conn_state_t nxt_h1p_idle_close_state;
#endif /* _NXT_HTTP_H_INCLUDED_ */
diff --git a/src/nxt_http_request.c b/src/nxt_http_request.c
index 1ab22223..916004d2 100644
--- a/src/nxt_http_request.c
+++ b/src/nxt_http_request.c
@@ -16,8 +16,6 @@ static void nxt_http_request_proto_info(nxt_task_t *task,
static void nxt_http_request_mem_buf_completion(nxt_task_t *task, void *obj,
void *data);
static void nxt_http_request_done(nxt_task_t *task, void *obj, void *data);
-static void nxt_http_request_close_handler(nxt_task_t *task, void *obj,
- void *data);
static u_char *nxt_http_date(u_char *buf, nxt_realtime_t *now, struct tm *tm,
size_t size, const char *format);
@@ -444,6 +442,16 @@ fail:
void
+nxt_http_request_ws_frame_start(nxt_task_t *task, nxt_http_request_t *r,
+ nxt_buf_t *ws_frame)
+{
+ if (r->proto.any != NULL) {
+ nxt_http_proto[r->protocol].ws_frame_start(task, r, ws_frame);
+ }
+}
+
+
+void
nxt_http_request_send(nxt_task_t *task, nxt_http_request_t *r, nxt_buf_t *out)
{
if (nxt_fast_path(r->proto.any != NULL)) {
@@ -530,7 +538,7 @@ nxt_http_request_error_handler(nxt_task_t *task, void *obj, void *data)
}
-static void
+void
nxt_http_request_close_handler(nxt_task_t *task, void *obj, void *data)
{
nxt_http_proto_t proto;
@@ -556,12 +564,13 @@ nxt_http_request_close_handler(nxt_task_t *task, void *obj, void *data)
}
}
- protocol = r->protocol;
-
r->proto.any = NULL;
- nxt_mp_release(r->mem_pool);
if (nxt_fast_path(proto.any != NULL)) {
+ protocol = r->protocol;
+
+ nxt_mp_release(r->mem_pool);
+
nxt_http_proto[protocol].close(task, proto, conf);
}
}
diff --git a/src/nxt_http_response.c b/src/nxt_http_response.c
index 755182db..00ecff00 100644
--- a/src/nxt_http_response.c
+++ b/src/nxt_http_response.c
@@ -28,6 +28,8 @@ static nxt_http_field_proc_t nxt_response_fields[] = {
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_string("Upgrade"), &nxt_http_response_skip, 0 },
+ { nxt_string("Sec-WebSocket-Accept"), &nxt_http_response_skip, 0 },
};
diff --git a/src/nxt_http_websocket.c b/src/nxt_http_websocket.c
new file mode 100644
index 00000000..d58d615c
--- /dev/null
+++ b/src/nxt_http_websocket.c
@@ -0,0 +1,161 @@
+
+/*
+ * Copyright (C) NGINX, Inc.
+ */
+
+#include <nxt_main.h>
+#include <nxt_router.h>
+#include <nxt_http.h>
+#include <nxt_router_request.h>
+#include <nxt_port_memory_int.h>
+#include <nxt_websocket.h>
+#include <nxt_websocket_header.h>
+
+
+static void nxt_http_websocket_client(nxt_task_t *task, void *obj, void *data);
+static void nxt_http_websocket_error_handler(nxt_task_t *task, void *obj,
+ void *data);
+
+
+const nxt_http_request_state_t nxt_http_websocket
+ nxt_aligned(64) =
+{
+ .ready_handler = nxt_http_websocket_client,
+ .error_handler = nxt_http_websocket_error_handler,
+};
+
+
+static void
+nxt_http_websocket_client(nxt_task_t *task, void *obj, void *data)
+{
+ size_t frame_size, used_size, copy_size, buf_free_size;
+ size_t chunk_copy_size;
+ nxt_buf_t *out, *buf, **out_tail, *b, *next;
+ nxt_int_t res;
+ nxt_http_request_t *r;
+ nxt_request_app_link_t *req_app_link;
+ nxt_request_rpc_data_t *req_rpc_data;
+ nxt_websocket_header_t *wsh;
+
+ r = obj;
+
+ if (nxt_slow_path((req_rpc_data = r->req_rpc_data) == NULL
+ || (req_app_link = req_rpc_data->req_app_link) == NULL))
+ {
+ nxt_debug(task, "websocket client frame for destroyed request");
+
+ return;
+ }
+
+ nxt_debug(task, "http websocket client frame");
+
+ wsh = (nxt_websocket_header_t *) r->ws_frame->mem.pos;
+
+ frame_size = nxt_websocket_frame_header_size(wsh)
+ + nxt_websocket_frame_payload_len(wsh);
+
+ buf = NULL;
+ buf_free_size = 0;
+ out = NULL;
+ out_tail = &out;
+
+ b = r->ws_frame;
+
+ while (b != NULL && frame_size > 0) {
+ used_size = nxt_buf_mem_used_size(&b->mem);
+ copy_size = nxt_min(used_size, frame_size);
+
+ while (copy_size > 0) {
+ if (buf == NULL || buf_free_size == 0) {
+ buf_free_size = nxt_min(frame_size, PORT_MMAP_DATA_SIZE);
+
+ buf = nxt_port_mmap_get_buf(task, req_app_link->app_port,
+ buf_free_size);
+
+ *out_tail = buf;
+ out_tail = &buf->next;
+ }
+
+ chunk_copy_size = nxt_min(buf_free_size, copy_size);
+
+ buf->mem.free = nxt_cpymem(buf->mem.free, b->mem.pos,
+ chunk_copy_size);
+
+ copy_size -= chunk_copy_size;
+ b->mem.pos += chunk_copy_size;
+ buf_free_size -= chunk_copy_size;
+ }
+
+ frame_size -= copy_size;
+ next = b->next;
+
+ if (nxt_buf_mem_used_size(&b->mem) == 0) {
+ nxt_work_queue_add(&task->thread->engine->fast_work_queue,
+ b->completion_handler, task, b, b->parent);
+
+ r->ws_frame = next;
+ }
+
+ b = next;
+ }
+
+ res = nxt_port_socket_twrite(task, req_app_link->app_port,
+ NXT_PORT_MSG_WEBSOCKET, -1,
+ req_app_link->stream,
+ req_app_link->reply_port->id, out, NULL);
+ if (nxt_slow_path(res != NXT_OK)) {
+ // TODO: handle
+ }
+
+ b = r->ws_frame;
+
+ if (b != NULL) {
+ used_size = nxt_buf_mem_used_size(&b->mem);
+
+ if (used_size > 0) {
+ nxt_memmove(b->mem.start, b->mem.pos, used_size);
+
+ b->mem.pos = b->mem.start;
+ b->mem.free = b->mem.start + used_size;
+ }
+ }
+
+ nxt_http_request_ws_frame_start(task, r, r->ws_frame);
+}
+
+
+static void
+nxt_http_websocket_error_handler(nxt_task_t *task, void *obj, void *data)
+{
+ nxt_http_request_t *r;
+ nxt_request_app_link_t *req_app_link;
+ nxt_request_rpc_data_t *req_rpc_data;
+
+ nxt_debug(task, "http websocket error handler");
+
+ r = obj;
+
+ if ((req_rpc_data = r->req_rpc_data) == NULL) {
+ nxt_debug(task, " req_rpc_data is NULL");
+ goto close_handler;
+ }
+
+ if ((req_app_link = req_rpc_data->req_app_link) == NULL) {
+ nxt_debug(task, " req_app_link is NULL");
+ goto close_handler;
+ }
+
+ if (req_app_link->app_port == NULL) {
+ nxt_debug(task, " app_port is NULL");
+ goto close_handler;
+ }
+
+ (void) nxt_port_socket_twrite(task, req_app_link->app_port,
+ NXT_PORT_MSG_WEBSOCKET_LAST,
+ -1, req_app_link->stream,
+ req_app_link->reply_port->id, NULL, NULL);
+
+close_handler:
+
+ nxt_http_request_close_handler(task, obj, data);
+}
diff --git a/src/nxt_port.c b/src/nxt_port.c
index aff46666..cef65cab 100644
--- a/src/nxt_port.c
+++ b/src/nxt_port.c
@@ -68,6 +68,7 @@ nxt_port_new(nxt_task_t *task, nxt_port_id_t id, nxt_pid_t pid,
nxt_queue_init(&port->messages);
nxt_thread_mutex_create(&port->write_mutex);
nxt_queue_init(&port->pending_requests);
+ nxt_queue_init(&port->active_websockets);
} else {
nxt_mp_destroy(mp);
diff --git a/src/nxt_port.h b/src/nxt_port.h
index e4e76693..eeb6caa5 100644
--- a/src/nxt_port.h
+++ b/src/nxt_port.h
@@ -36,6 +36,12 @@ struct nxt_port_handlers_s {
/* Stop process command. */
nxt_port_handler_t quit;
+ /* Request headers. */
+ nxt_port_handler_t req_headers;
+
+ /* Websocket frame. */
+ nxt_port_handler_t websocket_frame;
+
/* Various data. */
nxt_port_handler_t data;
};
@@ -71,6 +77,9 @@ typedef enum {
_NXT_PORT_MSG_REMOVE_PID = nxt_port_handler_idx(remove_pid),
_NXT_PORT_MSG_QUIT = nxt_port_handler_idx(quit),
+ _NXT_PORT_MSG_REQ_HEADERS = nxt_port_handler_idx(req_headers),
+ _NXT_PORT_MSG_WEBSOCKET = nxt_port_handler_idx(websocket_frame),
+
_NXT_PORT_MSG_DATA = nxt_port_handler_idx(data),
NXT_PORT_MSG_MAX = sizeof(nxt_port_handlers_t)
@@ -99,6 +108,10 @@ typedef enum {
NXT_PORT_MSG_QUIT = _NXT_PORT_MSG_QUIT | NXT_PORT_MSG_LAST,
NXT_PORT_MSG_REMOVE_PID = _NXT_PORT_MSG_REMOVE_PID | NXT_PORT_MSG_LAST,
+ NXT_PORT_MSG_REQ_HEADERS = _NXT_PORT_MSG_REQ_HEADERS,
+ NXT_PORT_MSG_WEBSOCKET = _NXT_PORT_MSG_WEBSOCKET,
+ NXT_PORT_MSG_WEBSOCKET_LAST = _NXT_PORT_MSG_WEBSOCKET | NXT_PORT_MSG_LAST,
+
NXT_PORT_MSG_DATA = _NXT_PORT_MSG_DATA,
NXT_PORT_MSG_DATA_LAST = _NXT_PORT_MSG_DATA | NXT_PORT_MSG_LAST,
} nxt_port_msg_type_t;
@@ -181,6 +194,8 @@ struct nxt_port_s {
uint32_t app_responses;
nxt_queue_t pending_requests;
+ nxt_queue_t active_websockets;
+
nxt_port_handler_t handler;
nxt_port_handler_t *data;
diff --git a/src/nxt_router.c b/src/nxt_router.c
index 566e0c65..b87f588f 100644
--- a/src/nxt_router.c
+++ b/src/nxt_router.c
@@ -14,7 +14,7 @@
#include <nxt_port_memory_int.h>
#include <nxt_unit_request.h>
#include <nxt_unit_response.h>
-
+#include <nxt_router_request.h>
typedef struct {
nxt_str_t type;
@@ -48,64 +48,6 @@ typedef struct {
#endif
-typedef struct nxt_msg_info_s {
- nxt_buf_t *buf;
- nxt_port_mmap_tracking_t tracking;
- nxt_work_handler_t completion_handler;
-} nxt_msg_info_t;
-
-
-typedef struct nxt_request_app_link_s nxt_request_app_link_t;
-
-
-typedef enum {
- NXT_APR_NEW_PORT,
- NXT_APR_REQUEST_FAILED,
- NXT_APR_GOT_RESPONSE,
- NXT_APR_CLOSE,
-} nxt_apr_action_t;
-
-
-typedef struct {
- uint32_t stream;
- nxt_app_t *app;
-
- nxt_port_t *app_port;
- nxt_apr_action_t apr_action;
-
- nxt_http_request_t *request;
- nxt_msg_info_t msg_info;
- nxt_request_app_link_t *req_app_link;
-} nxt_request_rpc_data_t;
-
-
-struct nxt_request_app_link_s {
- uint32_t stream;
- nxt_atomic_t use_count;
-
- nxt_port_t *app_port;
- nxt_apr_action_t apr_action;
-
- nxt_port_t *reply_port;
- nxt_http_request_t *request;
- nxt_msg_info_t msg_info;
- nxt_request_rpc_data_t *req_rpc_data;
-
- nxt_nsec_t res_time;
-
- nxt_queue_link_t link_app_requests; /* for nxt_app_t.requests */
- /* for nxt_port_t.pending_requests */
- nxt_queue_link_t link_port_pending;
- nxt_queue_link_t link_app_pending; /* for nxt_app_t.pending */
-
- nxt_mp_t *mem_pool;
- nxt_work_t work;
-
- int err_code;
- const char *err_str;
-};
-
-
typedef struct {
nxt_socket_conf_t *socket_conf;
nxt_router_temp_conf_t *temp_conf;
@@ -305,6 +247,8 @@ static nxt_int_t nxt_router_http_request_done(nxt_task_t *task,
static void nxt_router_http_request_release(nxt_task_t *task, void *obj,
void *data);
+const nxt_http_request_state_t nxt_http_websocket;
+
static nxt_router_t *nxt_router;
static const nxt_str_t http_prefix = nxt_string("HTTP_");
@@ -663,6 +607,7 @@ nxt_request_app_link_release(nxt_task_t *task,
nxt_request_app_link_t *req_app_link)
{
nxt_mp_t *mp;
+ nxt_http_request_t *r;
nxt_request_rpc_data_t *req_rpc_data;
nxt_assert(task->thread->engine == req_app_link->work.data);
@@ -683,10 +628,11 @@ nxt_request_app_link_release(nxt_task_t *task,
req_rpc_data->msg_info = req_app_link->msg_info;
if (req_rpc_data->app->timeout != 0) {
- req_rpc_data->request->timer.handler = nxt_router_app_timeout;
- req_rpc_data->request->timer_data = req_rpc_data;
- nxt_timer_add(task->thread->engine,
- &req_rpc_data->request->timer,
+ r = req_rpc_data->request;
+
+ r->timer.handler = nxt_router_app_timeout;
+ r->timer_data = req_rpc_data;
+ nxt_timer_add(task->thread->engine, &r->timer,
req_rpc_data->app->timeout);
}
@@ -833,14 +779,16 @@ nxt_request_rpc_data_unlink(nxt_task_t *task,
if (req_app_link->link_app_requests.next == NULL
&& req_app_link->link_port_pending.next == NULL
- && req_app_link->link_app_pending.next == NULL)
+ && req_app_link->link_app_pending.next == NULL
+ && req_app_link->link_port_websockets.next == NULL)
{
req_app_link = NULL;
} else {
ra_use_delta -=
nxt_queue_chk_remove(&req_app_link->link_app_requests)
- + nxt_queue_chk_remove(&req_app_link->link_port_pending);
+ + nxt_queue_chk_remove(&req_app_link->link_port_pending)
+ + nxt_queue_chk_remove(&req_app_link->link_port_websockets);
nxt_queue_chk_remove(&req_app_link->link_app_pending);
}
@@ -863,6 +811,7 @@ nxt_request_rpc_data_unlink(nxt_task_t *task,
nxt_router_http_request_done(task, req_rpc_data->request);
+ req_rpc_data->request->req_rpc_data = NULL;
req_rpc_data->request = NULL;
}
}
@@ -1412,6 +1361,28 @@ static nxt_conf_map_t nxt_router_http_conf[] = {
};
+static nxt_conf_map_t nxt_router_websocket_conf[] = {
+ {
+ nxt_string("max_frame_size"),
+ NXT_CONF_MAP_SIZE,
+ offsetof(nxt_websocket_conf_t, max_frame_size),
+ },
+
+ {
+ nxt_string("read_timeout"),
+ NXT_CONF_MAP_MSEC,
+ offsetof(nxt_websocket_conf_t, read_timeout),
+ },
+
+ {
+ nxt_string("keepalive_interval"),
+ NXT_CONF_MAP_MSEC,
+ offsetof(nxt_websocket_conf_t, keepalive_interval),
+ },
+
+};
+
+
static nxt_int_t
nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
u_char *start, u_char *end)
@@ -1425,7 +1396,7 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
nxt_app_t *app, *prev;
nxt_router_t *router;
nxt_app_joint_t *app_joint;
- nxt_conf_value_t *conf, *http, *value;
+ nxt_conf_value_t *conf, *http, *value, *websocket;
nxt_conf_value_t *applications, *application;
nxt_conf_value_t *listeners, *listener;
nxt_conf_value_t *routes_conf;
@@ -1448,6 +1419,7 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
#if (NXT_TLS)
static nxt_str_t certificate_path = nxt_string("/tls/certificate");
#endif
+ static nxt_str_t websocket_path = nxt_string("/settings/http/websocket");
conf = nxt_conf_json_parse(tmcf->mem_pool, start, end, NULL);
if (conf == NULL) {
@@ -1658,6 +1630,8 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
}
#endif
+ websocket = nxt_conf_get_path(conf, &websocket_path);
+
listeners = nxt_conf_get_path(conf, &listeners_path);
if (listeners != NULL) {
@@ -1697,6 +1671,10 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
skcf->body_read_timeout = 30 * 1000;
skcf->send_timeout = 30 * 1000;
+ skcf->websocket_conf.max_frame_size = 1024 * 1024;
+ skcf->websocket_conf.read_timeout = 60 * 1000;
+ skcf->websocket_conf.keepalive_interval = 30 * 1000;
+
if (http != NULL) {
ret = nxt_conf_map_object(mp, http, nxt_router_http_conf,
nxt_nitems(nxt_router_http_conf),
@@ -1707,6 +1685,17 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
}
}
+ if (websocket != NULL) {
+ ret = nxt_conf_map_object(mp, websocket,
+ nxt_router_websocket_conf,
+ nxt_nitems(nxt_router_websocket_conf),
+ &skcf->websocket_conf);
+ if (ret != NXT_OK) {
+ nxt_alert(task, "websocket map error");
+ goto fail;
+ }
+ }
+
#if (NXT_TLS)
value = nxt_conf_get_path(listener, &certificate_path);
@@ -3418,10 +3407,12 @@ nxt_router_response_ready_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg,
{
nxt_int_t ret;
nxt_buf_t *b;
+ nxt_port_t *app_port;
nxt_unit_field_t *f;
nxt_http_field_t *field;
nxt_http_request_t *r;
nxt_unit_response_t *resp;
+ nxt_request_app_link_t *req_app_link;
nxt_request_rpc_data_t *req_rpc_data;
b = msg->buf;
@@ -3542,7 +3533,48 @@ nxt_router_response_ready_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg,
nxt_http_request_header_send(task, r);
- r->state = &nxt_http_request_send_state;
+ if (r->websocket_handshake
+ && r->status == NXT_HTTP_SWITCHING_PROTOCOLS)
+ {
+ req_app_link = nxt_request_app_link_alloc(task,
+ req_rpc_data->req_app_link,
+ req_rpc_data);
+ if (nxt_slow_path(req_app_link == NULL)) {
+ goto fail;
+ }
+
+ app_port = req_app_link->app_port;
+
+ if (app_port == NULL && req_rpc_data->app_port != NULL) {
+ req_app_link->app_port = req_rpc_data->app_port;
+ app_port = req_app_link->app_port;
+ req_app_link->apr_action = req_rpc_data->apr_action;
+
+ req_rpc_data->app_port = NULL;
+ }
+
+ if (nxt_slow_path(app_port == NULL)) {
+ goto fail;
+ }
+
+ nxt_thread_mutex_lock(&req_rpc_data->app->mutex);
+
+ nxt_queue_insert_tail(&app_port->active_websockets,
+ &req_app_link->link_port_websockets);
+
+ nxt_thread_mutex_unlock(&req_rpc_data->app->mutex);
+
+ nxt_router_app_port_release(task, app_port, NXT_APR_UPGRADE);
+ req_app_link->apr_action = NXT_APR_CLOSE;
+
+ nxt_debug(task, "req_app_link stream #%uD upgrade",
+ req_app_link->stream);
+
+ r->state = &nxt_http_websocket;
+
+ } else {
+ r->state = &nxt_http_request_send_state;
+ }
if (r->out) {
nxt_work_queue_add(&task->thread->engine->fast_work_queue,
@@ -3924,6 +3956,10 @@ nxt_router_app_port_release(nxt_task_t *task, nxt_port_t *port,
got_response = 1;
inc_use = -1;
break;
+ case NXT_APR_UPGRADE:
+ dec_pending = 1;
+ got_response = 1;
+ break;
case NXT_APR_CLOSE:
inc_use = -1;
break;
@@ -4046,9 +4082,10 @@ re_ra_cancelled:
adjust_idle_timer = 0;
- if (port->pair[1] != -1 && !send_quit && port->app_pending_responses == 0) {
- nxt_assert(port->idle_link.next == NULL);
-
+ if (port->pair[1] != -1 && !send_quit && port->app_pending_responses == 0
+ && nxt_queue_is_empty(&port->active_websockets)
+ && port->idle_link.next == NULL)
+ {
if (app->idle_processes == app->spare_processes
&& app->adjust_idle_work.data == NULL)
{
@@ -4545,6 +4582,7 @@ nxt_router_process_http_request(nxt_task_t *task, nxt_http_request_t *r,
nxt_router_app_use(task, app, 1);
req_rpc_data->request = r;
+ r->req_rpc_data = req_rpc_data;
req_app_link = &ra_local;
nxt_request_app_link_init(task, req_app_link, req_rpc_data);
@@ -4635,7 +4673,7 @@ nxt_router_app_prepare_request(nxt_task_t *task,
goto release_port;
}
- res = nxt_port_socket_twrite(task, port, NXT_PORT_MSG_DATA,
+ res = nxt_port_socket_twrite(task, port, NXT_PORT_MSG_REQ_HEADERS,
-1, req_app_link->stream, reply_port->id, buf,
&req_app_link->msg_info.tracking);
@@ -4785,6 +4823,7 @@ nxt_router_prepare_msg(nxt_task_t *task, nxt_http_request_t *r,
*p++ = '\0';
req->tls = (r->tls != NULL);
+ req->websocket_handshake = r->websocket_handshake;
req->server_name_length = r->server_name.length;
nxt_unit_sptr_set(&req->server_name, p);
diff --git a/src/nxt_router.h b/src/nxt_router.h
index ff791e3d..b55a4de3 100644
--- a/src/nxt_router.h
+++ b/src/nxt_router.h
@@ -139,6 +139,13 @@ struct nxt_app_s {
typedef struct {
+ size_t max_frame_size;
+ nxt_msec_t read_timeout;
+ nxt_msec_t keepalive_interval;
+} nxt_websocket_conf_t;
+
+
+typedef struct {
uint32_t count;
nxt_queue_link_t link;
nxt_router_conf_t *router_conf;
@@ -164,6 +171,8 @@ typedef struct {
nxt_msec_t body_read_timeout;
nxt_msec_t send_timeout;
+ nxt_websocket_conf_t websocket_conf;
+
#if (NXT_TLS)
nxt_tls_conf_t *tls;
#endif
diff --git a/src/nxt_router_request.h b/src/nxt_router_request.h
new file mode 100644
index 00000000..c3d5767e
--- /dev/null
+++ b/src/nxt_router_request.h
@@ -0,0 +1,71 @@
+
+/*
+ * Copyright (C) NGINX, Inc.
+ */
+
+#ifndef _NXT_ROUTER_REQUEST_H_INCLUDED_
+#define _NXT_ROUTER_REQUEST_H_INCLUDED_
+
+
+typedef struct nxt_msg_info_s {
+ nxt_buf_t *buf;
+ nxt_port_mmap_tracking_t tracking;
+ nxt_work_handler_t completion_handler;
+} nxt_msg_info_t;
+
+
+typedef struct nxt_request_app_link_s nxt_request_app_link_t;
+
+
+typedef enum {
+ NXT_APR_NEW_PORT,
+ NXT_APR_REQUEST_FAILED,
+ NXT_APR_GOT_RESPONSE,
+ NXT_APR_UPGRADE,
+ NXT_APR_CLOSE,
+} nxt_apr_action_t;
+
+
+typedef struct {
+ uint32_t stream;
+ nxt_app_t *app;
+
+ nxt_port_t *app_port;
+ nxt_apr_action_t apr_action;
+
+ nxt_http_request_t *request;
+ nxt_msg_info_t msg_info;
+ nxt_request_app_link_t *req_app_link;
+} nxt_request_rpc_data_t;
+
+
+struct nxt_request_app_link_s {
+ uint32_t stream;
+ nxt_atomic_t use_count;
+
+ nxt_port_t *app_port;
+ nxt_apr_action_t apr_action;
+
+ nxt_port_t *reply_port;
+ nxt_http_request_t *request;
+ nxt_msg_info_t msg_info;
+ nxt_request_rpc_data_t *req_rpc_data;
+
+ nxt_nsec_t res_time;
+
+ nxt_queue_link_t link_app_requests; /* for nxt_app_t.requests */
+ /* for nxt_port_t.pending_requests */
+ nxt_queue_link_t link_port_pending;
+ nxt_queue_link_t link_app_pending; /* for nxt_app_t.pending */
+ /* for nxt_port_t.active_websockets */
+ nxt_queue_link_t link_port_websockets;
+
+ nxt_mp_t *mem_pool;
+ nxt_work_t work;
+
+ int err_code;
+ const char *err_str;
+};
+
+
+#endif /* _NXT_ROUTER_REQUEST_H_INCLUDED_ */
diff --git a/src/nxt_sha1.c b/src/nxt_sha1.c
new file mode 100644
index 00000000..407c5933
--- /dev/null
+++ b/src/nxt_sha1.c
@@ -0,0 +1,295 @@
+
+/*
+ * Copyright (C) Maxim Dounin
+ * Copyright (C) NGINX, Inc.
+ *
+ * An internal SHA1 implementation.
+ */
+
+
+#include <nxt_main.h>
+#include <nxt_sha1.h>
+
+
+static const u_char *nxt_sha1_body(nxt_sha1_t *ctx, const u_char *data,
+ size_t size);
+
+
+void
+nxt_sha1_init(nxt_sha1_t *ctx)
+{
+ ctx->a = 0x67452301;
+ ctx->b = 0xefcdab89;
+ ctx->c = 0x98badcfe;
+ ctx->d = 0x10325476;
+ ctx->e = 0xc3d2e1f0;
+
+ ctx->bytes = 0;
+}
+
+
+void
+nxt_sha1_update(nxt_sha1_t *ctx, const void *data, size_t size)
+{
+ size_t used, free;
+
+ used = (size_t) (ctx->bytes & 0x3f);
+ ctx->bytes += size;
+
+ if (used) {
+ free = 64 - used;
+
+ if (size < free) {
+ memcpy(&ctx->buffer[used], data, size);
+ return;
+ }
+
+ memcpy(&ctx->buffer[used], data, free);
+ data = (u_char *) data + free;
+ size -= free;
+ (void) nxt_sha1_body(ctx, ctx->buffer, 64);
+ }
+
+ if (size >= 64) {
+ data = nxt_sha1_body(ctx, data, size & ~(size_t) 0x3f);
+ size &= 0x3f;
+ }
+
+ memcpy(ctx->buffer, data, size);
+}
+
+
+void
+nxt_sha1_final(u_char result[20], nxt_sha1_t *ctx)
+{
+ size_t used, free;
+
+ used = (size_t) (ctx->bytes & 0x3f);
+
+ ctx->buffer[used++] = 0x80;
+
+ free = 64 - used;
+
+ if (free < 8) {
+ nxt_memzero(&ctx->buffer[used], free);
+ (void) nxt_sha1_body(ctx, ctx->buffer, 64);
+ used = 0;
+ free = 64;
+ }
+
+ nxt_memzero(&ctx->buffer[used], free - 8);
+
+ ctx->bytes <<= 3;
+ ctx->buffer[56] = (u_char) (ctx->bytes >> 56);
+ ctx->buffer[57] = (u_char) (ctx->bytes >> 48);
+ ctx->buffer[58] = (u_char) (ctx->bytes >> 40);
+ ctx->buffer[59] = (u_char) (ctx->bytes >> 32);
+ ctx->buffer[60] = (u_char) (ctx->bytes >> 24);
+ ctx->buffer[61] = (u_char) (ctx->bytes >> 16);
+ ctx->buffer[62] = (u_char) (ctx->bytes >> 8);
+ ctx->buffer[63] = (u_char) ctx->bytes;
+
+ (void) nxt_sha1_body(ctx, ctx->buffer, 64);
+
+ result[0] = (u_char) (ctx->a >> 24);
+ result[1] = (u_char) (ctx->a >> 16);
+ result[2] = (u_char) (ctx->a >> 8);
+ result[3] = (u_char) ctx->a;
+ result[4] = (u_char) (ctx->b >> 24);
+ result[5] = (u_char) (ctx->b >> 16);
+ result[6] = (u_char) (ctx->b >> 8);
+ result[7] = (u_char) ctx->b;
+ result[8] = (u_char) (ctx->c >> 24);
+ result[9] = (u_char) (ctx->c >> 16);
+ result[10] = (u_char) (ctx->c >> 8);
+ result[11] = (u_char) ctx->c;
+ result[12] = (u_char) (ctx->d >> 24);
+ result[13] = (u_char) (ctx->d >> 16);
+ result[14] = (u_char) (ctx->d >> 8);
+ result[15] = (u_char) ctx->d;
+ result[16] = (u_char) (ctx->e >> 24);
+ result[17] = (u_char) (ctx->e >> 16);
+ result[18] = (u_char) (ctx->e >> 8);
+ result[19] = (u_char) ctx->e;
+
+ nxt_memzero(ctx, sizeof(*ctx));
+}
+
+
+/*
+ * Helper functions.
+ */
+
+#define ROTATE(bits, word) (((word) << (bits)) | ((word) >> (32 - (bits))))
+
+#define F1(b, c, d) (((b) & (c)) | ((~(b)) & (d)))
+#define F2(b, c, d) ((b) ^ (c) ^ (d))
+#define F3(b, c, d) (((b) & (c)) | ((b) & (d)) | ((c) & (d)))
+
+#define STEP(f, a, b, c, d, e, w, t) \
+ temp = ROTATE(5, (a)) + f((b), (c), (d)) + (e) + (w) + (t); \
+ (e) = (d); \
+ (d) = (c); \
+ (c) = ROTATE(30, (b)); \
+ (b) = (a); \
+ (a) = temp;
+
+
+/*
+ * GET() reads 4 input bytes in big-endian byte order and returns
+ * them as uint32_t.
+ */
+
+#define GET(n) \
+ ( ((uint32_t) p[n * 4 + 3]) \
+ | ((uint32_t) p[n * 4 + 2] << 8) \
+ | ((uint32_t) p[n * 4 + 1] << 16) \
+ | ((uint32_t) p[n * 4] << 24))
+
+
+/*
+ * This processes one or more 64-byte data blocks, but does not update
+ * the bit counters. There are no alignment requirements.
+ */
+
+static const u_char *
+nxt_sha1_body(nxt_sha1_t *ctx, const u_char *data, size_t size)
+{
+ uint32_t a, b, c, d, e, temp;
+ uint32_t saved_a, saved_b, saved_c, saved_d, saved_e;
+ uint32_t words[80];
+ nxt_uint_t i;
+ const u_char *p;
+
+ p = data;
+
+ a = ctx->a;
+ b = ctx->b;
+ c = ctx->c;
+ d = ctx->d;
+ e = ctx->e;
+
+ do {
+ saved_a = a;
+ saved_b = b;
+ saved_c = c;
+ saved_d = d;
+ saved_e = e;
+
+ /* Load data block into the words array */
+
+ for (i = 0; i < 16; i++) {
+ words[i] = GET(i);
+ }
+
+ for (i = 16; i < 80; i++) {
+ words[i] = ROTATE(1, words[i - 3]
+ ^ words[i - 8]
+ ^ words[i - 14]
+ ^ words[i - 16]);
+ }
+
+ /* Transformations */
+
+ STEP(F1, a, b, c, d, e, words[0], 0x5a827999);
+ STEP(F1, a, b, c, d, e, words[1], 0x5a827999);
+ STEP(F1, a, b, c, d, e, words[2], 0x5a827999);
+ STEP(F1, a, b, c, d, e, words[3], 0x5a827999);
+ STEP(F1, a, b, c, d, e, words[4], 0x5a827999);
+ STEP(F1, a, b, c, d, e, words[5], 0x5a827999);
+ STEP(F1, a, b, c, d, e, words[6], 0x5a827999);
+ STEP(F1, a, b, c, d, e, words[7], 0x5a827999);
+ STEP(F1, a, b, c, d, e, words[8], 0x5a827999);
+ STEP(F1, a, b, c, d, e, words[9], 0x5a827999);
+ STEP(F1, a, b, c, d, e, words[10], 0x5a827999);
+ STEP(F1, a, b, c, d, e, words[11], 0x5a827999);
+ STEP(F1, a, b, c, d, e, words[12], 0x5a827999);
+ STEP(F1, a, b, c, d, e, words[13], 0x5a827999);
+ STEP(F1, a, b, c, d, e, words[14], 0x5a827999);
+ STEP(F1, a, b, c, d, e, words[15], 0x5a827999);
+ STEP(F1, a, b, c, d, e, words[16], 0x5a827999);
+ STEP(F1, a, b, c, d, e, words[17], 0x5a827999);
+ STEP(F1, a, b, c, d, e, words[18], 0x5a827999);
+ STEP(F1, a, b, c, d, e, words[19], 0x5a827999);
+
+ STEP(F2, a, b, c, d, e, words[20], 0x6ed9eba1);
+ STEP(F2, a, b, c, d, e, words[21], 0x6ed9eba1);
+ STEP(F2, a, b, c, d, e, words[22], 0x6ed9eba1);
+ STEP(F2, a, b, c, d, e, words[23], 0x6ed9eba1);
+ STEP(F2, a, b, c, d, e, words[24], 0x6ed9eba1);
+ STEP(F2, a, b, c, d, e, words[25], 0x6ed9eba1);
+ STEP(F2, a, b, c, d, e, words[26], 0x6ed9eba1);
+ STEP(F2, a, b, c, d, e, words[27], 0x6ed9eba1);
+ STEP(F2, a, b, c, d, e, words[28], 0x6ed9eba1);
+ STEP(F2, a, b, c, d, e, words[29], 0x6ed9eba1);
+ STEP(F2, a, b, c, d, e, words[30], 0x6ed9eba1);
+ STEP(F2, a, b, c, d, e, words[31], 0x6ed9eba1);
+ STEP(F2, a, b, c, d, e, words[32], 0x6ed9eba1);
+ STEP(F2, a, b, c, d, e, words[33], 0x6ed9eba1);
+ STEP(F2, a, b, c, d, e, words[34], 0x6ed9eba1);
+ STEP(F2, a, b, c, d, e, words[35], 0x6ed9eba1);
+ STEP(F2, a, b, c, d, e, words[36], 0x6ed9eba1);
+ STEP(F2, a, b, c, d, e, words[37], 0x6ed9eba1);
+ STEP(F2, a, b, c, d, e, words[38], 0x6ed9eba1);
+ STEP(F2, a, b, c, d, e, words[39], 0x6ed9eba1);
+
+ STEP(F3, a, b, c, d, e, words[40], 0x8f1bbcdc);
+ STEP(F3, a, b, c, d, e, words[41], 0x8f1bbcdc);
+ STEP(F3, a, b, c, d, e, words[42], 0x8f1bbcdc);
+ STEP(F3, a, b, c, d, e, words[43], 0x8f1bbcdc);
+ STEP(F3, a, b, c, d, e, words[44], 0x8f1bbcdc);
+ STEP(F3, a, b, c, d, e, words[45], 0x8f1bbcdc);
+ STEP(F3, a, b, c, d, e, words[46], 0x8f1bbcdc);
+ STEP(F3, a, b, c, d, e, words[47], 0x8f1bbcdc);
+ STEP(F3, a, b, c, d, e, words[48], 0x8f1bbcdc);
+ STEP(F3, a, b, c, d, e, words[49], 0x8f1bbcdc);
+ STEP(F3, a, b, c, d, e, words[50], 0x8f1bbcdc);
+ STEP(F3, a, b, c, d, e, words[51], 0x8f1bbcdc);
+ STEP(F3, a, b, c, d, e, words[52], 0x8f1bbcdc);
+ STEP(F3, a, b, c, d, e, words[53], 0x8f1bbcdc);
+ STEP(F3, a, b, c, d, e, words[54], 0x8f1bbcdc);
+ STEP(F3, a, b, c, d, e, words[55], 0x8f1bbcdc);
+ STEP(F3, a, b, c, d, e, words[56], 0x8f1bbcdc);
+ STEP(F3, a, b, c, d, e, words[57], 0x8f1bbcdc);
+ STEP(F3, a, b, c, d, e, words[58], 0x8f1bbcdc);
+ STEP(F3, a, b, c, d, e, words[59], 0x8f1bbcdc);
+
+ STEP(F2, a, b, c, d, e, words[60], 0xca62c1d6);
+ STEP(F2, a, b, c, d, e, words[61], 0xca62c1d6);
+ STEP(F2, a, b, c, d, e, words[62], 0xca62c1d6);
+ STEP(F2, a, b, c, d, e, words[63], 0xca62c1d6);
+ STEP(F2, a, b, c, d, e, words[64], 0xca62c1d6);
+ STEP(F2, a, b, c, d, e, words[65], 0xca62c1d6);
+ STEP(F2, a, b, c, d, e, words[66], 0xca62c1d6);
+ STEP(F2, a, b, c, d, e, words[67], 0xca62c1d6);
+ STEP(F2, a, b, c, d, e, words[68], 0xca62c1d6);
+ STEP(F2, a, b, c, d, e, words[69], 0xca62c1d6);
+ STEP(F2, a, b, c, d, e, words[70], 0xca62c1d6);
+ STEP(F2, a, b, c, d, e, words[71], 0xca62c1d6);
+ STEP(F2, a, b, c, d, e, words[72], 0xca62c1d6);
+ STEP(F2, a, b, c, d, e, words[73], 0xca62c1d6);
+ STEP(F2, a, b, c, d, e, words[74], 0xca62c1d6);
+ STEP(F2, a, b, c, d, e, words[75], 0xca62c1d6);
+ STEP(F2, a, b, c, d, e, words[76], 0xca62c1d6);
+ STEP(F2, a, b, c, d, e, words[77], 0xca62c1d6);
+ STEP(F2, a, b, c, d, e, words[78], 0xca62c1d6);
+ STEP(F2, a, b, c, d, e, words[79], 0xca62c1d6);
+
+ a += saved_a;
+ b += saved_b;
+ c += saved_c;
+ d += saved_d;
+ e += saved_e;
+
+ p += 64;
+
+ } while (size -= 64);
+
+ ctx->a = a;
+ ctx->b = b;
+ ctx->c = c;
+ ctx->d = d;
+ ctx->e = e;
+
+ return p;
+}
diff --git a/src/nxt_sha1.h b/src/nxt_sha1.h
new file mode 100644
index 00000000..2816982b
--- /dev/null
+++ b/src/nxt_sha1.h
@@ -0,0 +1,24 @@
+
+/*
+ * Copyright (C) Igor Sysoev
+ * Copyright (C) NGINX, Inc.
+ */
+
+
+#ifndef _NXT_SHA1_H_INCLUDED_
+#define _NXT_SHA1_H_INCLUDED_
+
+
+typedef struct {
+ uint64_t bytes;
+ uint32_t a, b, c, d, e;
+ u_char buffer[64];
+} nxt_sha1_t;
+
+
+NXT_EXPORT void nxt_sha1_init(nxt_sha1_t *ctx);
+NXT_EXPORT void nxt_sha1_update(nxt_sha1_t *ctx, const void *data, size_t size);
+NXT_EXPORT void nxt_sha1_final(u_char result[20], nxt_sha1_t *ctx);
+
+
+#endif /* _NXT_SHA1_H_INCLUDED_ */
diff --git a/src/nxt_unit.c b/src/nxt_unit.c
index 88c7fa6a..28a0de20 100644
--- a/src/nxt_unit.c
+++ b/src/nxt_unit.c
@@ -11,38 +11,60 @@
#include "nxt_unit.h"
#include "nxt_unit_request.h"
#include "nxt_unit_response.h"
+#include "nxt_unit_websocket.h"
+
+#include "nxt_websocket.h"
#if (NXT_HAVE_MEMFD_CREATE)
#include <linux/memfd.h>
#endif
-typedef struct nxt_unit_impl_s nxt_unit_impl_t;
-typedef struct nxt_unit_mmap_s nxt_unit_mmap_t;
-typedef struct nxt_unit_mmaps_s nxt_unit_mmaps_t;
-typedef struct nxt_unit_process_s nxt_unit_process_t;
-typedef struct nxt_unit_mmap_buf_s nxt_unit_mmap_buf_t;
-typedef struct nxt_unit_recv_msg_s nxt_unit_recv_msg_t;
-typedef struct nxt_unit_ctx_impl_s nxt_unit_ctx_impl_t;
-typedef struct nxt_unit_port_impl_s nxt_unit_port_impl_t;
-typedef struct nxt_unit_request_info_impl_s nxt_unit_request_info_impl_t;
+typedef struct nxt_unit_impl_s nxt_unit_impl_t;
+typedef struct nxt_unit_mmap_s nxt_unit_mmap_t;
+typedef struct nxt_unit_mmaps_s nxt_unit_mmaps_t;
+typedef struct nxt_unit_process_s nxt_unit_process_t;
+typedef struct nxt_unit_mmap_buf_s nxt_unit_mmap_buf_t;
+typedef struct nxt_unit_recv_msg_s nxt_unit_recv_msg_t;
+typedef struct nxt_unit_ctx_impl_s nxt_unit_ctx_impl_t;
+typedef struct nxt_unit_port_impl_s nxt_unit_port_impl_t;
+typedef struct nxt_unit_request_info_impl_s nxt_unit_request_info_impl_t;
+typedef struct nxt_unit_websocket_frame_impl_s nxt_unit_websocket_frame_impl_t;
static nxt_unit_impl_t *nxt_unit_create(nxt_unit_init_t *init);
static void nxt_unit_ctx_init(nxt_unit_impl_t *lib,
nxt_unit_ctx_impl_t *ctx_impl, void *data);
+nxt_inline void nxt_unit_mmap_buf_insert(nxt_unit_mmap_buf_t **head,
+ nxt_unit_mmap_buf_t *mmap_buf);
+nxt_inline void nxt_unit_mmap_buf_insert_tail(nxt_unit_mmap_buf_t **prev,
+ nxt_unit_mmap_buf_t *mmap_buf);
+nxt_inline void nxt_unit_mmap_buf_remove(nxt_unit_mmap_buf_t *mmap_buf);
static int nxt_unit_read_env(nxt_unit_port_t *ready_port,
nxt_unit_port_t *read_port, int *log_fd, uint32_t *stream);
static int nxt_unit_ready(nxt_unit_ctx_t *ctx, nxt_unit_port_id_t *port_id,
uint32_t stream);
+static int nxt_unit_process_new_port(nxt_unit_ctx_t *ctx,
+ nxt_unit_recv_msg_t *recv_msg);
+static int nxt_unit_process_req_headers(nxt_unit_ctx_t *ctx,
+ nxt_unit_recv_msg_t *recv_msg);
+static int nxt_unit_process_websocket(nxt_unit_ctx_t *ctx,
+ nxt_unit_recv_msg_t *recv_msg);
static nxt_unit_request_info_impl_t *nxt_unit_request_info_get(
nxt_unit_ctx_t *ctx);
static void nxt_unit_request_info_release(nxt_unit_request_info_t *req);
static void nxt_unit_request_info_free(nxt_unit_request_info_impl_t *req);
+static nxt_unit_websocket_frame_impl_t *nxt_unit_websocket_frame_get(
+ nxt_unit_ctx_t *ctx);
+static void nxt_unit_websocket_frame_release(nxt_unit_websocket_frame_t *ws);
+static void nxt_unit_websocket_frame_free(nxt_unit_websocket_frame_impl_t *ws);
static nxt_unit_process_t *nxt_unit_msg_get_process(nxt_unit_ctx_t *ctx,
nxt_unit_recv_msg_t *recv_msg);
static nxt_unit_mmap_buf_t *nxt_unit_mmap_buf_get(nxt_unit_ctx_t *ctx);
static void nxt_unit_mmap_buf_release(nxt_unit_mmap_buf_t *mmap_buf);
static int nxt_unit_mmap_buf_send(nxt_unit_ctx_t *ctx, uint32_t stream,
nxt_unit_mmap_buf_t *mmap_buf, int last);
+static void nxt_unit_mmap_buf_free(nxt_unit_mmap_buf_t *mmap_buf);
+static ssize_t nxt_unit_buf_read(nxt_unit_buf_t **b, uint64_t *len, void *dst,
+ size_t size);
static nxt_port_mmap_header_t *nxt_unit_mmap_get(nxt_unit_ctx_t *ctx,
nxt_unit_process_t *process, nxt_unit_port_id_t *port_id,
nxt_chunk_id_t *c, int n);
@@ -65,7 +87,7 @@ static nxt_port_mmap_header_t *nxt_unit_get_incoming_mmap(nxt_unit_ctx_t *ctx,
static int nxt_unit_tracking_read(nxt_unit_ctx_t *ctx,
nxt_unit_recv_msg_t *recv_msg);
static int nxt_unit_mmap_read(nxt_unit_ctx_t *ctx,
- nxt_unit_recv_msg_t *recv_msg, nxt_queue_t *incoming_buf);
+ nxt_unit_recv_msg_t *recv_msg);
static int nxt_unit_mmap_release(nxt_port_mmap_header_t *hdr, void *start,
uint32_t size);
@@ -98,14 +120,22 @@ static int nxt_unit_port_hash_add(nxt_lvlhsh_t *port_hash,
static nxt_unit_port_impl_t *nxt_unit_port_hash_find(nxt_lvlhsh_t *port_hash,
nxt_unit_port_id_t *port_id, int remove);
+static int nxt_unit_request_hash_add(nxt_lvlhsh_t *request_hash,
+ nxt_unit_request_info_impl_t *req_impl);
+static nxt_unit_request_info_impl_t *nxt_unit_request_hash_find(
+ nxt_lvlhsh_t *request_hash, uint32_t stream, int remove);
+
static char * nxt_unit_snprint_prefix(char *p, char *end, pid_t pid, int level);
struct nxt_unit_mmap_buf_s {
nxt_unit_buf_t buf;
+ nxt_unit_mmap_buf_t *next;
+ nxt_unit_mmap_buf_t **prev;
+
nxt_port_mmap_header_t *hdr;
- nxt_queue_link_t link;
+// nxt_queue_link_t link;
nxt_unit_port_id_t port_id;
nxt_unit_request_info_t *req;
nxt_unit_ctx_impl_t *ctx_impl;
@@ -113,12 +143,20 @@ struct nxt_unit_mmap_buf_s {
struct nxt_unit_recv_msg_s {
- nxt_port_msg_t port_msg;
+ uint32_t stream;
+ nxt_pid_t pid;
+ nxt_port_id_t reply_port;
+
+ uint8_t last; /* 1 bit */
+ uint8_t mmap; /* 1 bit */
void *start;
uint32_t size;
+ int fd;
nxt_unit_process_t *process;
+
+ nxt_unit_mmap_buf_t *incoming_buf;
};
@@ -127,18 +165,22 @@ typedef enum {
NXT_UNIT_RS_RESPONSE_INIT,
NXT_UNIT_RS_RESPONSE_HAS_CONTENT,
NXT_UNIT_RS_RESPONSE_SENT,
- NXT_UNIT_RS_DONE,
+ NXT_UNIT_RS_RELEASED,
} nxt_unit_req_state_t;
struct nxt_unit_request_info_impl_s {
nxt_unit_request_info_t req;
- nxt_unit_recv_msg_t recv_msg;
- nxt_queue_t outgoing_buf; /* of nxt_unit_mmap_buf_t */
- nxt_queue_t incoming_buf; /* of nxt_unit_mmap_buf_t */
+ uint32_t stream;
+
+ nxt_unit_process_t *process;
+
+ nxt_unit_mmap_buf_t *outgoing_buf;
+ nxt_unit_mmap_buf_t *incoming_buf;
nxt_unit_req_state_t state;
+ uint8_t websocket;
nxt_queue_link_t link;
@@ -146,6 +188,19 @@ struct nxt_unit_request_info_impl_s {
};
+struct nxt_unit_websocket_frame_impl_s {
+ nxt_unit_websocket_frame_t ws;
+
+ nxt_unit_mmap_buf_t *buf;
+
+ nxt_queue_link_t link;
+
+ nxt_unit_ctx_impl_t *ctx_impl;
+
+ void *retain_buf;
+};
+
+
struct nxt_unit_ctx_impl_s {
nxt_unit_ctx_t ctx;
@@ -154,14 +209,20 @@ struct nxt_unit_ctx_impl_s {
nxt_queue_link_t link;
- nxt_queue_t free_buf; /* of nxt_unit_mmap_buf_t */
+ nxt_unit_mmap_buf_t *free_buf;
/* of nxt_unit_request_info_impl_t */
nxt_queue_t free_req;
+ /* of nxt_unit_websocket_frame_impl_t */
+ nxt_queue_t free_ws;
+
/* of nxt_unit_request_info_impl_t */
nxt_queue_t active_req;
+ /* of nxt_unit_request_info_impl_t */
+ nxt_lvlhsh_t requests;
+
nxt_unit_mmap_buf_t ctx_buf[2];
nxt_unit_request_info_impl_t req;
@@ -394,18 +455,65 @@ nxt_unit_ctx_init(nxt_unit_impl_t *lib, nxt_unit_ctx_impl_t *ctx_impl,
nxt_queue_insert_tail(&lib->contexts, &ctx_impl->link);
- nxt_queue_init(&ctx_impl->free_buf);
nxt_queue_init(&ctx_impl->free_req);
+ nxt_queue_init(&ctx_impl->free_ws);
nxt_queue_init(&ctx_impl->active_req);
- nxt_queue_insert_tail(&ctx_impl->free_buf, &ctx_impl->ctx_buf[0].link);
- nxt_queue_insert_tail(&ctx_impl->free_buf, &ctx_impl->ctx_buf[1].link);
+ ctx_impl->free_buf = NULL;
+ nxt_unit_mmap_buf_insert(&ctx_impl->free_buf, &ctx_impl->ctx_buf[1]);
+ nxt_unit_mmap_buf_insert(&ctx_impl->free_buf, &ctx_impl->ctx_buf[0]);
+
nxt_queue_insert_tail(&ctx_impl->free_req, &ctx_impl->req.link);
ctx_impl->req.req.ctx = &ctx_impl->ctx;
ctx_impl->req.req.unit = &lib->unit;
ctx_impl->read_port_fd = -1;
+ ctx_impl->requests.slot = 0;
+}
+
+
+nxt_inline void
+nxt_unit_mmap_buf_insert(nxt_unit_mmap_buf_t **head,
+ nxt_unit_mmap_buf_t *mmap_buf)
+{
+ mmap_buf->next = *head;
+
+ if (mmap_buf->next != NULL) {
+ mmap_buf->next->prev = &mmap_buf->next;
+ }
+
+ *head = mmap_buf;
+ mmap_buf->prev = head;
+}
+
+
+nxt_inline void
+nxt_unit_mmap_buf_insert_tail(nxt_unit_mmap_buf_t **prev,
+ nxt_unit_mmap_buf_t *mmap_buf)
+{
+ while (*prev != NULL) {
+ prev = &(*prev)->next;
+ }
+
+ nxt_unit_mmap_buf_insert(prev, mmap_buf);
+}
+
+
+nxt_inline void
+nxt_unit_mmap_buf_remove(nxt_unit_mmap_buf_t *mmap_buf)
+{
+ nxt_unit_mmap_buf_t **prev;
+
+ prev = mmap_buf->prev;
+
+ if (mmap_buf->next != NULL) {
+ mmap_buf->next->prev = prev;
+ }
+
+ if (prev != NULL) {
+ *prev = mmap_buf->next;
+ }
}
@@ -509,26 +617,18 @@ int
nxt_unit_process_msg(nxt_unit_ctx_t *ctx, nxt_unit_port_id_t *port_id,
void *buf, size_t buf_size, void *oob, size_t oob_size)
{
- int fd, rc, nb;
- pid_t pid;
- nxt_queue_t incoming_buf;
- struct cmsghdr *cm;
- nxt_port_msg_t *port_msg;
- nxt_unit_impl_t *lib;
- nxt_unit_port_t new_port;
- nxt_queue_link_t *lnk;
- nxt_unit_request_t *r;
- nxt_unit_mmap_buf_t *b;
- nxt_unit_recv_msg_t recv_msg;
- nxt_unit_callbacks_t *cb;
- nxt_port_msg_new_port_t *new_port_msg;
- nxt_unit_request_info_t *req;
- nxt_unit_request_info_impl_t *req_impl;
+ int rc;
+ pid_t pid;
+ struct cmsghdr *cm;
+ nxt_port_msg_t *port_msg;
+ nxt_unit_impl_t *lib;
+ nxt_unit_recv_msg_t recv_msg;
+ nxt_unit_callbacks_t *cb;
lib = nxt_container_of(ctx->unit, nxt_unit_impl_t, unit);
rc = NXT_UNIT_ERROR;
- fd = -1;
+ recv_msg.fd = -1;
recv_msg.process = NULL;
port_msg = buf;
cm = oob;
@@ -538,17 +638,22 @@ nxt_unit_process_msg(nxt_unit_ctx_t *ctx, nxt_unit_port_id_t *port_id,
&& cm->cmsg_level == SOL_SOCKET
&& cm->cmsg_type == SCM_RIGHTS)
{
- memcpy(&fd, CMSG_DATA(cm), sizeof(int));
+ memcpy(&recv_msg.fd, CMSG_DATA(cm), sizeof(int));
}
- nxt_queue_init(&incoming_buf);
+ recv_msg.incoming_buf = NULL;
if (nxt_slow_path(buf_size < sizeof(nxt_port_msg_t))) {
nxt_unit_warn(ctx, "message too small (%d bytes)", (int) buf_size);
goto fail;
}
- recv_msg.port_msg = *port_msg;
+ recv_msg.stream = port_msg->stream;
+ recv_msg.pid = port_msg->pid;
+ recv_msg.reply_port = port_msg->reply_port;
+ recv_msg.last = port_msg->last;
+ recv_msg.mmap = port_msg->mmap;
+
recv_msg.start = port_msg + 1;
recv_msg.size = buf_size - sizeof(nxt_port_msg_t);
@@ -572,7 +677,7 @@ nxt_unit_process_msg(nxt_unit_ctx_t *ctx, nxt_unit_port_id_t *port_id,
}
if (port_msg->mmap) {
- if (nxt_unit_mmap_read(ctx, &recv_msg, &incoming_buf) != NXT_UNIT_OK) {
+ if (nxt_unit_mmap_read(ctx, &recv_msg) != NXT_UNIT_OK) {
goto fail;
}
}
@@ -589,187 +694,326 @@ nxt_unit_process_msg(nxt_unit_ctx_t *ctx, nxt_unit_port_id_t *port_id,
break;
case _NXT_PORT_MSG_NEW_PORT:
- if (nxt_slow_path(recv_msg.size != sizeof(nxt_port_msg_new_port_t))) {
- nxt_unit_warn(ctx, "#%"PRIu32": new_port: "
- "invalid message size (%d)",
- port_msg->stream, (int) recv_msg.size);
+ rc = nxt_unit_process_new_port(ctx, &recv_msg);
+ break;
- goto fail;
- }
+ case _NXT_PORT_MSG_CHANGE_FILE:
+ nxt_unit_debug(ctx, "#%"PRIu32": change_file: fd %d",
+ port_msg->stream, recv_msg.fd);
+ break;
- if (nxt_slow_path(fd < 0)) {
- nxt_unit_alert(ctx, "#%"PRIu32": invalid fd %d for new port",
- port_msg->stream, fd);
+ case _NXT_PORT_MSG_MMAP:
+ if (nxt_slow_path(recv_msg.fd < 0)) {
+ nxt_unit_alert(ctx, "#%"PRIu32": invalid fd %d for mmap",
+ port_msg->stream, recv_msg.fd);
goto fail;
}
- new_port_msg = recv_msg.start;
+ rc = nxt_unit_incoming_mmap(ctx, port_msg->pid, recv_msg.fd);
+ break;
- nxt_unit_debug(ctx, "#%"PRIu32": new_port: %d,%d fd %d",
- port_msg->stream, (int) new_port_msg->pid,
- (int) new_port_msg->id, fd);
+ case _NXT_PORT_MSG_REQ_HEADERS:
+ rc = nxt_unit_process_req_headers(ctx, &recv_msg);
+ break;
- nb = 0;
+ case _NXT_PORT_MSG_WEBSOCKET:
+ rc = nxt_unit_process_websocket(ctx, &recv_msg);
+ break;
- if (nxt_slow_path(ioctl(fd, FIONBIO, &nb) == -1)) {
- nxt_unit_alert(ctx, "#%"PRIu32": new_port: ioctl(%d, FIONBIO, 0) "
- "failed: %s (%d)", fd, strerror(errno), errno);
+ case _NXT_PORT_MSG_REMOVE_PID:
+ if (nxt_slow_path(recv_msg.size != sizeof(pid))) {
+ nxt_unit_warn(ctx, "#%"PRIu32": remove_pid: invalid message size "
+ "(%d != %d)", port_msg->stream, (int) recv_msg.size,
+ (int) sizeof(pid));
goto fail;
}
- nxt_unit_port_id_init(&new_port.id, new_port_msg->pid,
- new_port_msg->id);
+ memcpy(&pid, recv_msg.start, sizeof(pid));
- new_port.in_fd = -1;
- new_port.out_fd = fd;
- new_port.data = NULL;
+ nxt_unit_debug(ctx, "#%"PRIu32": remove_pid: %d",
+ port_msg->stream, (int) pid);
- fd = -1;
+ cb->remove_pid(ctx, pid);
- rc = cb->add_port(ctx, &new_port);
+ rc = NXT_UNIT_OK;
break;
- case _NXT_PORT_MSG_CHANGE_FILE:
- nxt_unit_debug(ctx, "#%"PRIu32": change_file: fd %d",
- port_msg->stream, fd);
- break;
+ default:
+ nxt_unit_debug(ctx, "#%"PRIu32": ignore message type: %d",
+ port_msg->stream, (int) port_msg->type);
- case _NXT_PORT_MSG_MMAP:
- if (nxt_slow_path(fd < 0)) {
- nxt_unit_alert(ctx, "#%"PRIu32": invalid fd %d for mmap",
- port_msg->stream, fd);
+ goto fail;
+ }
- goto fail;
- }
+fail:
- rc = nxt_unit_incoming_mmap(ctx, port_msg->pid, fd);
- break;
+ if (recv_msg.fd != -1) {
+ close(recv_msg.fd);
+ }
- case _NXT_PORT_MSG_DATA:
- if (nxt_slow_path(port_msg->mmap == 0)) {
- nxt_unit_warn(ctx, "#%"PRIu32": data is not in shared memory",
- port_msg->stream);
+ while (recv_msg.incoming_buf != NULL) {
+ nxt_unit_mmap_buf_free(recv_msg.incoming_buf);
+ }
- goto fail;
- }
+ if (recv_msg.process != NULL) {
+ nxt_unit_process_use(ctx, recv_msg.process, -1);
+ }
- if (nxt_slow_path(recv_msg.size < sizeof(nxt_unit_request_t))) {
- nxt_unit_warn(ctx, "#%"PRIu32": data too short: %d while at least "
- "%d expected", port_msg->stream, (int) recv_msg.size,
- (int) sizeof(nxt_unit_request_t));
+ return rc;
+}
- goto fail;
- }
- req_impl = nxt_unit_request_info_get(ctx);
- if (nxt_slow_path(req_impl == NULL)) {
- nxt_unit_warn(ctx, "#%"PRIu32": request info allocation failed",
- port_msg->stream);
+static int
+nxt_unit_process_new_port(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg)
+{
+ int nb;
+ nxt_unit_impl_t *lib;
+ nxt_unit_port_t new_port;
+ nxt_port_msg_new_port_t *new_port_msg;
- goto fail;
- }
+ if (nxt_slow_path(recv_msg->size != sizeof(nxt_port_msg_new_port_t))) {
+ nxt_unit_warn(ctx, "#%"PRIu32": new_port: "
+ "invalid message size (%d)",
+ recv_msg->stream, (int) recv_msg->size);
- req = &req_impl->req;
+ return NXT_UNIT_ERROR;
+ }
- req->request_port = *port_id;
+ if (nxt_slow_path(recv_msg->fd < 0)) {
+ nxt_unit_alert(ctx, "#%"PRIu32": invalid fd %d for new port",
+ recv_msg->stream, recv_msg->fd);
- nxt_unit_port_id_init(&req->response_port, port_msg->pid,
- port_msg->reply_port);
+ return NXT_UNIT_ERROR;
+ }
- req->request = recv_msg.start;
+ new_port_msg = recv_msg->start;
- lnk = nxt_queue_first(&incoming_buf);
- b = nxt_container_of(lnk, nxt_unit_mmap_buf_t, link);
+ nxt_unit_debug(ctx, "#%"PRIu32": new_port: %d,%d fd %d",
+ recv_msg->stream, (int) new_port_msg->pid,
+ (int) new_port_msg->id, recv_msg->fd);
- req->request_buf = &b->buf;
- req->response = NULL;
- req->response_buf = NULL;
+ nb = 0;
- r = req->request;
+ if (nxt_slow_path(ioctl(recv_msg->fd, FIONBIO, &nb) == -1)) {
+ nxt_unit_alert(ctx, "#%"PRIu32": new_port: ioctl(%d, FIONBIO, 0) "
+ "failed: %s (%d)", recv_msg->fd, strerror(errno), errno);
- req->content_length = r->content_length;
+ return NXT_UNIT_ERROR;
+ }
- req->content_buf = req->request_buf;
- req->content_buf->free = nxt_unit_sptr_get(&r->preread_content);
+ nxt_unit_port_id_init(&new_port.id, new_port_msg->pid,
+ new_port_msg->id);
- /* Move process to req_impl. */
- req_impl->recv_msg = recv_msg;
+ new_port.in_fd = -1;
+ new_port.out_fd = recv_msg->fd;
+ new_port.data = NULL;
- recv_msg.process = NULL;
+ recv_msg->fd = -1;
- nxt_queue_init(&req_impl->outgoing_buf);
- nxt_queue_init(&req_impl->incoming_buf);
+ lib = nxt_container_of(ctx->unit, nxt_unit_impl_t, unit);
- nxt_queue_each(b, &incoming_buf, nxt_unit_mmap_buf_t, link)
- {
- b->req = req;
- } nxt_queue_loop;
+ return lib->callbacks.add_port(ctx, &new_port);
+}
- nxt_queue_add(&req_impl->incoming_buf, &incoming_buf);
- nxt_queue_init(&incoming_buf);
- req->response_max_fields = 0;
- req_impl->state = NXT_UNIT_RS_START;
+static int
+nxt_unit_process_req_headers(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg)
+{
+ nxt_unit_impl_t *lib;
+ nxt_unit_request_t *r;
+ nxt_unit_mmap_buf_t *b;
+ nxt_unit_request_info_t *req;
+ nxt_unit_request_info_impl_t *req_impl;
- nxt_unit_debug(ctx, "#%"PRIu32": %.*s %.*s (%d)", port_msg->stream,
- (int) r->method_length, nxt_unit_sptr_get(&r->method),
- (int) r->target_length, nxt_unit_sptr_get(&r->target),
- (int) r->content_length);
+ if (nxt_slow_path(recv_msg->mmap == 0)) {
+ nxt_unit_warn(ctx, "#%"PRIu32": data is not in shared memory",
+ recv_msg->stream);
- cb->request_handler(req);
+ return NXT_UNIT_ERROR;
+ }
- rc = NXT_UNIT_OK;
- break;
+ if (nxt_slow_path(recv_msg->size < sizeof(nxt_unit_request_t))) {
+ nxt_unit_warn(ctx, "#%"PRIu32": data too short: %d while at least "
+ "%d expected", recv_msg->stream, (int) recv_msg->size,
+ (int) sizeof(nxt_unit_request_t));
- case _NXT_PORT_MSG_REMOVE_PID:
- if (nxt_slow_path(recv_msg.size != sizeof(pid))) {
- nxt_unit_warn(ctx, "#%"PRIu32": remove_pid: invalid message size "
- "(%d != %d)", port_msg->stream, (int) recv_msg.size,
- (int) sizeof(pid));
+ return NXT_UNIT_ERROR;
+ }
- goto fail;
- }
+ req_impl = nxt_unit_request_info_get(ctx);
+ if (nxt_slow_path(req_impl == NULL)) {
+ nxt_unit_warn(ctx, "#%"PRIu32": request info allocation failed",
+ recv_msg->stream);
- memcpy(&pid, recv_msg.start, sizeof(pid));
+ return NXT_UNIT_ERROR;
+ }
- nxt_unit_debug(ctx, "#%"PRIu32": remove_pid: %d",
- port_msg->stream, (int) pid);
+ req = &req_impl->req;
- cb->remove_pid(ctx, pid);
+ nxt_unit_port_id_init(&req->response_port, recv_msg->pid,
+ recv_msg->reply_port);
- rc = NXT_UNIT_OK;
- break;
+ req->request = recv_msg->start;
- default:
- nxt_unit_debug(ctx, "#%"PRIu32": ignore message type: %d",
- port_msg->stream, (int) port_msg->type);
+ b = recv_msg->incoming_buf;
- goto fail;
+ req->request_buf = &b->buf;
+ req->response = NULL;
+ req->response_buf = NULL;
+
+ r = req->request;
+
+ req->content_length = r->content_length;
+
+ req->content_buf = req->request_buf;
+ req->content_buf->free = nxt_unit_sptr_get(&r->preread_content);
+
+ /* "Move" process reference to req_impl. */
+ req_impl->process = nxt_unit_msg_get_process(ctx, recv_msg);
+ if (nxt_slow_path(req_impl->process == NULL)) {
+ return NXT_UNIT_ERROR;
}
-fail:
+ recv_msg->process = NULL;
- if (fd != -1) {
- close(fd);
+ req_impl->stream = recv_msg->stream;
+
+ req_impl->outgoing_buf = NULL;
+
+ for (b = recv_msg->incoming_buf; b != NULL; b = b->next) {
+ b->req = req;
}
- if (port_msg->mmap) {
- nxt_queue_each(b, &incoming_buf, nxt_unit_mmap_buf_t, link)
- {
- nxt_unit_mmap_release(b->hdr, b->buf.start,
- b->buf.end - b->buf.start);
+ /* "Move" incoming buffer list to req_impl. */
+ req_impl->incoming_buf = recv_msg->incoming_buf;
+ req_impl->incoming_buf->prev = &req_impl->incoming_buf;
+ recv_msg->incoming_buf = NULL;
+
+ req->response_max_fields = 0;
+ req_impl->state = NXT_UNIT_RS_START;
+ req_impl->websocket = 0;
+
+ nxt_unit_debug(ctx, "#%"PRIu32": %.*s %.*s (%d)", recv_msg->stream,
+ (int) r->method_length, nxt_unit_sptr_get(&r->method),
+ (int) r->target_length, nxt_unit_sptr_get(&r->target),
+ (int) r->content_length);
+
+ lib = nxt_container_of(ctx->unit, nxt_unit_impl_t, unit);
+
+ lib->callbacks.request_handler(req);
- nxt_unit_mmap_buf_release(b);
- } nxt_queue_loop;
+ return NXT_UNIT_OK;
+}
+
+
+static int
+nxt_unit_process_websocket(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg)
+{
+ size_t hsize;
+ nxt_unit_impl_t *lib;
+ nxt_unit_mmap_buf_t *b;
+ nxt_unit_ctx_impl_t *ctx_impl;
+ nxt_unit_callbacks_t *cb;
+ nxt_unit_request_info_t *req;
+ nxt_unit_request_info_impl_t *req_impl;
+ nxt_unit_websocket_frame_impl_t *ws_impl;
+
+ ctx_impl = nxt_container_of(ctx, nxt_unit_ctx_impl_t, ctx);
+
+ req_impl = nxt_unit_request_hash_find(&ctx_impl->requests, recv_msg->stream,
+ recv_msg->last);
+ if (req_impl == NULL) {
+ return NXT_UNIT_OK;
}
- if (recv_msg.process != NULL) {
- nxt_unit_process_use(ctx, recv_msg.process, -1);
+ req = &req_impl->req;
+
+ lib = nxt_container_of(ctx->unit, nxt_unit_impl_t, unit);
+ cb = &lib->callbacks;
+
+ if (cb->websocket_handler && recv_msg->size >= 2) {
+ ws_impl = nxt_unit_websocket_frame_get(ctx);
+ if (nxt_slow_path(ws_impl == NULL)) {
+ nxt_unit_warn(ctx, "#%"PRIu32": websocket frame allocation failed",
+ req_impl->stream);
+
+ return NXT_UNIT_ERROR;
+ }
+
+ ws_impl->ws.req = req;
+
+ ws_impl->buf = NULL;
+ ws_impl->retain_buf = NULL;
+
+ if (recv_msg->mmap) {
+ for (b = recv_msg->incoming_buf; b != NULL; b = b->next) {
+ b->req = req;
+ }
+
+ /* "Move" incoming buffer list to ws_impl. */
+ ws_impl->buf = recv_msg->incoming_buf;
+ ws_impl->buf->prev = &ws_impl->buf;
+ recv_msg->incoming_buf = NULL;
+
+ b = ws_impl->buf;
+
+ } else {
+ b = nxt_unit_mmap_buf_get(ctx);
+ if (nxt_slow_path(b == NULL)) {
+ return NXT_UNIT_ERROR;
+ }
+
+ b->hdr = NULL;
+ b->req = req;
+ b->buf.start = recv_msg->start;
+ b->buf.free = b->buf.start;
+ b->buf.end = b->buf.start + recv_msg->size;
+
+ nxt_unit_mmap_buf_insert(&ws_impl->buf, b);
+ }
+
+ ws_impl->ws.header = (void *) b->buf.start;
+ ws_impl->ws.payload_len = nxt_websocket_frame_payload_len(
+ ws_impl->ws.header);
+
+ hsize = nxt_websocket_frame_header_size(ws_impl->ws.header);
+
+ if (ws_impl->ws.header->mask) {
+ ws_impl->ws.mask = (uint8_t *) b->buf.start + hsize - 4;
+
+ } else {
+ ws_impl->ws.mask = NULL;
+ }
+
+ b->buf.free += hsize;
+
+ ws_impl->ws.content_buf = &b->buf;
+ ws_impl->ws.content_length = ws_impl->ws.payload_len;
+
+ nxt_unit_req_debug(req, "websocket_handler: opcode=%d, "
+ "payload_len=%"PRIu64,
+ ws_impl->ws.header->opcode,
+ ws_impl->ws.payload_len);
+
+ cb->websocket_handler(&ws_impl->ws);
}
- return rc;
+ if (recv_msg->last) {
+ req_impl->websocket = 0;
+
+ if (cb->close_handler) {
+ nxt_unit_req_debug(req, "close_handler");
+
+ cb->close_handler(req);
+
+ } else {
+ nxt_unit_request_done(req, NXT_UNIT_ERROR);
+ }
+ }
+
+ return NXT_UNIT_OK;
}
@@ -815,9 +1059,7 @@ nxt_unit_request_info_get(nxt_unit_ctx_t *ctx)
static void
nxt_unit_request_info_release(nxt_unit_request_info_t *req)
{
- nxt_unit_mmap_buf_t *b;
nxt_unit_ctx_impl_t *ctx_impl;
- nxt_unit_recv_msg_t *recv_msg;
nxt_unit_request_info_impl_t *req_impl;
ctx_impl = nxt_container_of(req->ctx, nxt_unit_ctx_impl_t, ctx);
@@ -826,30 +1068,31 @@ nxt_unit_request_info_release(nxt_unit_request_info_t *req)
req->response = NULL;
req->response_buf = NULL;
- recv_msg = &req_impl->recv_msg;
+ if (req_impl->process != NULL) {
+ nxt_unit_process_use(req->ctx, req_impl->process, -1);
- if (recv_msg->process != NULL) {
- nxt_unit_process_use(req->ctx, recv_msg->process, -1);
-
- recv_msg->process = NULL;
+ req_impl->process = NULL;
}
- nxt_queue_each(b, &req_impl->outgoing_buf, nxt_unit_mmap_buf_t, link) {
-
- nxt_unit_buf_free(&b->buf);
+ if (req_impl->websocket) {
+ nxt_unit_request_hash_find(&ctx_impl->requests, req_impl->stream, 1);
- } nxt_queue_loop;
-
- nxt_queue_each(b, &req_impl->incoming_buf, nxt_unit_mmap_buf_t, link) {
+ req_impl->websocket = 0;
+ }
- nxt_unit_mmap_release(b->hdr, b->buf.start, b->buf.end - b->buf.start);
- nxt_unit_mmap_buf_release(b);
+ while (req_impl->outgoing_buf != NULL) {
+ nxt_unit_mmap_buf_free(req_impl->outgoing_buf);
+ }
- } nxt_queue_loop;
+ while (req_impl->incoming_buf != NULL) {
+ nxt_unit_mmap_buf_free(req_impl->incoming_buf);
+ }
nxt_queue_remove(&req_impl->link);
nxt_queue_insert_tail(&ctx_impl->free_req, &req_impl->link);
+
+ req_impl->state = NXT_UNIT_RS_RELEASED;
}
@@ -868,6 +1111,68 @@ nxt_unit_request_info_free(nxt_unit_request_info_impl_t *req_impl)
}
+static nxt_unit_websocket_frame_impl_t *
+nxt_unit_websocket_frame_get(nxt_unit_ctx_t *ctx)
+{
+ nxt_queue_link_t *lnk;
+ nxt_unit_ctx_impl_t *ctx_impl;
+ nxt_unit_websocket_frame_impl_t *ws_impl;
+
+ ctx_impl = nxt_container_of(ctx, nxt_unit_ctx_impl_t, ctx);
+
+ if (nxt_queue_is_empty(&ctx_impl->free_ws)) {
+ ws_impl = malloc(sizeof(nxt_unit_websocket_frame_impl_t));
+ if (nxt_slow_path(ws_impl == NULL)) {
+ nxt_unit_warn(ctx, "websocket frame allocation failed");
+
+ return NULL;
+ }
+
+ } else {
+ lnk = nxt_queue_first(&ctx_impl->free_ws);
+ nxt_queue_remove(lnk);
+
+ ws_impl = nxt_container_of(lnk, nxt_unit_websocket_frame_impl_t, link);
+ }
+
+ ws_impl->ctx_impl = ctx_impl;
+
+ return ws_impl;
+}
+
+
+static void
+nxt_unit_websocket_frame_release(nxt_unit_websocket_frame_t *ws)
+{
+ nxt_unit_websocket_frame_impl_t *ws_impl;
+
+ ws_impl = nxt_container_of(ws, nxt_unit_websocket_frame_impl_t, ws);
+
+ while (ws_impl->buf != NULL) {
+ nxt_unit_mmap_buf_free(ws_impl->buf);
+ }
+
+ ws->req = NULL;
+
+ if (ws_impl->retain_buf != NULL) {
+ free(ws_impl->retain_buf);
+
+ ws_impl->retain_buf = NULL;
+ }
+
+ nxt_queue_insert_tail(&ws_impl->ctx_impl->free_ws, &ws_impl->link);
+}
+
+
+static void
+nxt_unit_websocket_frame_free(nxt_unit_websocket_frame_impl_t *ws_impl)
+{
+ nxt_queue_remove(&ws_impl->link);
+
+ free(ws_impl);
+}
+
+
uint16_t
nxt_unit_field_hash(const char *name, size_t name_length)
{
@@ -1275,6 +1580,10 @@ nxt_unit_response_send(nxt_unit_request_info_t *req)
return NXT_UNIT_ERROR;
}
+ if (req->request->websocket_handshake && req->response->status == 101) {
+ nxt_unit_response_upgrade(req);
+ }
+
nxt_unit_req_debug(req, "send: %"PRIu32" fields, %d bytes",
req->response->fields_count,
(int) (req->response_buf->free
@@ -1282,9 +1591,7 @@ nxt_unit_response_send(nxt_unit_request_info_t *req)
mmap_buf = nxt_container_of(req->response_buf, nxt_unit_mmap_buf_t, buf);
- rc = nxt_unit_mmap_buf_send(req->ctx,
- req_impl->recv_msg.port_msg.stream,
- mmap_buf, 0);
+ rc = nxt_unit_mmap_buf_send(req->ctx, req_impl->stream, mmap_buf, 0);
if (nxt_fast_path(rc == NXT_UNIT_OK)) {
req->response = NULL;
req->response_buf = NULL;
@@ -1312,7 +1619,6 @@ nxt_unit_buf_t *
nxt_unit_response_buf_alloc(nxt_unit_request_info_t *req, uint32_t size)
{
int rc;
- nxt_unit_process_t *process;
nxt_unit_mmap_buf_t *mmap_buf;
nxt_unit_request_info_impl_t *req_impl;
@@ -1327,11 +1633,6 @@ nxt_unit_response_buf_alloc(nxt_unit_request_info_t *req, uint32_t size)
req_impl = nxt_container_of(req, nxt_unit_request_info_impl_t, req);
- process = nxt_unit_msg_get_process(req->ctx, &req_impl->recv_msg);
- if (nxt_slow_path(process == NULL)) {
- return NULL;
- }
-
mmap_buf = nxt_unit_mmap_buf_get(req->ctx);
if (nxt_slow_path(mmap_buf == NULL)) {
return NULL;
@@ -1339,10 +1640,10 @@ nxt_unit_response_buf_alloc(nxt_unit_request_info_t *req, uint32_t size)
mmap_buf->req = req;
- nxt_queue_insert_tail(&req_impl->outgoing_buf, &mmap_buf->link);
+ nxt_unit_mmap_buf_insert_tail(&req_impl->outgoing_buf, mmap_buf);
- rc = nxt_unit_get_outgoing_buf(req->ctx, process, &req->response_port,
- size, mmap_buf);
+ rc = nxt_unit_get_outgoing_buf(req->ctx, req_impl->process,
+ &req->response_port, size, mmap_buf);
if (nxt_slow_path(rc != NXT_UNIT_OK)) {
nxt_unit_mmap_buf_release(mmap_buf);
@@ -1366,13 +1667,13 @@ nxt_unit_msg_get_process(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg)
pthread_mutex_lock(&lib->mutex);
- recv_msg->process = nxt_unit_process_find(ctx, recv_msg->port_msg.pid, 0);
+ recv_msg->process = nxt_unit_process_find(ctx, recv_msg->pid, 0);
pthread_mutex_unlock(&lib->mutex);
if (recv_msg->process == NULL) {
nxt_unit_warn(ctx, "#%"PRIu32": process %d not found",
- recv_msg->port_msg.stream, (int) recv_msg->port_msg.pid);
+ recv_msg->stream, (int) recv_msg->pid);
}
return recv_msg->process;
@@ -1382,23 +1683,21 @@ nxt_unit_msg_get_process(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg)
static nxt_unit_mmap_buf_t *
nxt_unit_mmap_buf_get(nxt_unit_ctx_t *ctx)
{
- nxt_queue_link_t *lnk;
nxt_unit_mmap_buf_t *mmap_buf;
nxt_unit_ctx_impl_t *ctx_impl;
ctx_impl = nxt_container_of(ctx, nxt_unit_ctx_impl_t, ctx);
- if (nxt_queue_is_empty(&ctx_impl->free_buf)) {
+ if (ctx_impl->free_buf == NULL) {
mmap_buf = malloc(sizeof(nxt_unit_mmap_buf_t));
if (nxt_slow_path(mmap_buf == NULL)) {
nxt_unit_warn(ctx, "failed to allocate buf");
}
} else {
- lnk = nxt_queue_first(&ctx_impl->free_buf);
- nxt_queue_remove(lnk);
+ mmap_buf = ctx_impl->free_buf;
- mmap_buf = nxt_container_of(lnk, nxt_unit_mmap_buf_t, link);
+ nxt_unit_mmap_buf_remove(mmap_buf);
}
mmap_buf->ctx_impl = ctx_impl;
@@ -1410,9 +1709,91 @@ nxt_unit_mmap_buf_get(nxt_unit_ctx_t *ctx)
static void
nxt_unit_mmap_buf_release(nxt_unit_mmap_buf_t *mmap_buf)
{
- nxt_queue_remove(&mmap_buf->link);
+ nxt_unit_mmap_buf_remove(mmap_buf);
+
+ nxt_unit_mmap_buf_insert(&mmap_buf->ctx_impl->free_buf, mmap_buf);
+}
+
+
+typedef struct {
+ size_t len;
+ const char *str;
+} nxt_unit_str_t;
+
+
+#define nxt_unit_str(str) { nxt_length(str), str }
+
+
+int
+nxt_unit_request_is_websocket_handshake(nxt_unit_request_info_t *req)
+{
+ return req->request->websocket_handshake;
+}
+
+
+int
+nxt_unit_response_upgrade(nxt_unit_request_info_t *req)
+{
+ int rc;
+ nxt_unit_ctx_impl_t *ctx_impl;
+ nxt_unit_request_info_impl_t *req_impl;
+
+ req_impl = nxt_container_of(req, nxt_unit_request_info_impl_t, req);
+
+ if (nxt_slow_path(req_impl->websocket != 0)) {
+ nxt_unit_req_debug(req, "upgrade: already upgraded");
+
+ return NXT_UNIT_OK;
+ }
+
+ if (nxt_slow_path(req_impl->state < NXT_UNIT_RS_RESPONSE_INIT)) {
+ nxt_unit_req_warn(req, "upgrade: response is not initialized yet");
+
+ return NXT_UNIT_ERROR;
+ }
+
+ if (nxt_slow_path(req_impl->state >= NXT_UNIT_RS_RESPONSE_SENT)) {
+ nxt_unit_req_warn(req, "upgrade: response already sent");
+
+ return NXT_UNIT_ERROR;
+ }
+
+ ctx_impl = nxt_container_of(req->ctx, nxt_unit_ctx_impl_t, ctx);
+
+ rc = nxt_unit_request_hash_add(&ctx_impl->requests, req_impl);
+ if (nxt_slow_path(rc != NXT_UNIT_OK)) {
+ nxt_unit_req_warn(req, "upgrade: failed to add request to hash");
+
+ return NXT_UNIT_ERROR;
+ }
+
+ req_impl->websocket = 1;
+
+ req->response->status = 101;
+
+ return NXT_UNIT_OK;
+}
+
+
+int
+nxt_unit_response_is_websocket(nxt_unit_request_info_t *req)
+{
+ nxt_unit_request_info_impl_t *req_impl;
- nxt_queue_insert_tail(&mmap_buf->ctx_impl->free_buf, &mmap_buf->link);
+ req_impl = nxt_container_of(req, nxt_unit_request_info_impl_t, req);
+
+ return req_impl->websocket;
+}
+
+
+nxt_unit_request_info_t *
+nxt_unit_get_request_info_from_data(void *data)
+{
+ nxt_unit_request_info_impl_t *req_impl;
+
+ req_impl = nxt_container_of(data, nxt_unit_request_info_impl_t, extra_data);
+
+ return &req_impl->req;
}
@@ -1445,9 +1826,7 @@ nxt_unit_buf_send(nxt_unit_buf_t *buf)
}
if (nxt_fast_path(buf->free > buf->start)) {
- rc = nxt_unit_mmap_buf_send(req->ctx,
- req_impl->recv_msg.port_msg.stream,
- mmap_buf, 0);
+ rc = nxt_unit_mmap_buf_send(req->ctx, req_impl->stream, mmap_buf, 0);
if (nxt_slow_path(rc != NXT_UNIT_OK)) {
return rc;
}
@@ -1472,10 +1851,7 @@ nxt_unit_buf_send_done(nxt_unit_buf_t *buf)
req = mmap_buf->req;
req_impl = nxt_container_of(req, nxt_unit_request_info_impl_t, req);
- rc = nxt_unit_mmap_buf_send(req->ctx,
- req_impl->recv_msg.port_msg.stream,
- mmap_buf, 1);
-
+ rc = nxt_unit_mmap_buf_send(req->ctx, req_impl->stream, mmap_buf, 1);
if (nxt_slow_path(rc == NXT_UNIT_OK)) {
nxt_unit_mmap_buf_release(mmap_buf);
@@ -1506,6 +1882,7 @@ nxt_unit_mmap_buf_send(nxt_unit_ctx_t *ctx, uint32_t stream,
lib = nxt_container_of(ctx->unit, nxt_unit_impl_t, unit);
buf = &mmap_buf->buf;
+ hdr = mmap_buf->hdr;
m.mmap_msg.size = buf->free - buf->start;
@@ -1514,15 +1891,15 @@ nxt_unit_mmap_buf_send(nxt_unit_ctx_t *ctx, uint32_t stream,
m.msg.reply_port = 0;
m.msg.type = _NXT_PORT_MSG_DATA;
m.msg.last = last != 0;
- m.msg.mmap = m.mmap_msg.size > 0;
+ m.msg.mmap = hdr != NULL && m.mmap_msg.size > 0;
m.msg.nf = 0;
m.msg.mf = 0;
m.msg.tracking = 0;
- hdr = mmap_buf->hdr;
-
- m.mmap_msg.mmap_id = hdr->id;
- m.mmap_msg.chunk_id = nxt_port_mmap_chunk_id(hdr, (u_char *) buf->start);
+ if (hdr != NULL) {
+ m.mmap_msg.mmap_id = hdr->id;
+ m.mmap_msg.chunk_id = nxt_port_mmap_chunk_id(hdr, (u_char *) buf->start);
+ }
nxt_unit_debug(ctx, "#%"PRIu32": send mmap: (%d,%d,%d)",
stream,
@@ -1531,14 +1908,13 @@ nxt_unit_mmap_buf_send(nxt_unit_ctx_t *ctx, uint32_t stream,
(int) m.mmap_msg.size);
res = lib->callbacks.port_send(ctx, &mmap_buf->port_id, &m,
- m.mmap_msg.size > 0 ? sizeof(m)
- : sizeof(m.msg),
+ m.msg.mmap ? sizeof(m) : sizeof(m.msg),
NULL, 0);
if (nxt_slow_path(res != sizeof(m))) {
return NXT_UNIT_ERROR;
}
- if (buf->end - buf->free >= PORT_MMAP_CHUNK_SIZE) {
+ if (buf->end - buf->free >= PORT_MMAP_CHUNK_SIZE && hdr != NULL) {
last_used = (u_char *) buf->free - 1;
first_free_chunk = nxt_port_mmap_chunk_id(hdr, last_used) + 1;
@@ -1557,11 +1933,17 @@ nxt_unit_mmap_buf_send(nxt_unit_ctx_t *ctx, uint32_t stream,
void
nxt_unit_buf_free(nxt_unit_buf_t *buf)
{
- nxt_unit_mmap_buf_t *mmap_buf;
+ nxt_unit_mmap_buf_free(nxt_container_of(buf, nxt_unit_mmap_buf_t, buf));
+}
- mmap_buf = nxt_container_of(buf, nxt_unit_mmap_buf_t, buf);
- nxt_unit_mmap_release(mmap_buf->hdr, buf->start, buf->end - buf->start);
+static void
+nxt_unit_mmap_buf_free(nxt_unit_mmap_buf_t *mmap_buf)
+{
+ if (nxt_fast_path(mmap_buf->hdr != NULL)) {
+ nxt_unit_mmap_release(mmap_buf->hdr, mmap_buf->buf.start,
+ mmap_buf->buf.end - mmap_buf->buf.start);
+ }
nxt_unit_mmap_buf_release(mmap_buf);
}
@@ -1570,26 +1952,15 @@ nxt_unit_buf_free(nxt_unit_buf_t *buf)
nxt_unit_buf_t *
nxt_unit_buf_next(nxt_unit_buf_t *buf)
{
- nxt_queue_link_t *lnk;
- nxt_unit_mmap_buf_t *mmap_buf;
- nxt_unit_request_info_impl_t *req_impl;
+ nxt_unit_mmap_buf_t *mmap_buf;
mmap_buf = nxt_container_of(buf, nxt_unit_mmap_buf_t, buf);
- req_impl = nxt_container_of(mmap_buf->req, nxt_unit_request_info_impl_t,
- req);
- lnk = &mmap_buf->link;
-
- if (lnk == nxt_queue_last(&req_impl->incoming_buf)
- || lnk == nxt_queue_last(&req_impl->outgoing_buf))
- {
+ if (mmap_buf->next == NULL) {
return NULL;
}
- lnk = nxt_queue_next(lnk);
- mmap_buf = nxt_container_of(lnk, nxt_unit_mmap_buf_t, link);
-
- return &mmap_buf->buf;
+ return &mmap_buf->next->buf;
}
@@ -1614,7 +1985,6 @@ nxt_unit_response_write(nxt_unit_request_info_t *req, const void *start,
int rc;
uint32_t part_size;
const char *part_start;
- nxt_unit_process_t *process;
nxt_unit_mmap_buf_t mmap_buf;
nxt_unit_request_info_impl_t *req_impl;
@@ -1641,16 +2011,12 @@ nxt_unit_response_write(nxt_unit_request_info_t *req, const void *start,
part_start += part_size;
}
- process = nxt_unit_msg_get_process(req->ctx, &req_impl->recv_msg);
- if (nxt_slow_path(process == NULL)) {
- return NXT_UNIT_ERROR;
- }
-
while (size > 0) {
part_size = nxt_min(size, PORT_MMAP_DATA_SIZE);
- rc = nxt_unit_get_outgoing_buf(req->ctx, process, &req->response_port,
- part_size, &mmap_buf);
+ rc = nxt_unit_get_outgoing_buf(req->ctx, req_impl->process,
+ &req->response_port, part_size,
+ &mmap_buf);
if (nxt_slow_path(rc != NXT_UNIT_OK)) {
return rc;
}
@@ -1658,9 +2024,7 @@ nxt_unit_response_write(nxt_unit_request_info_t *req, const void *start,
mmap_buf.buf.free = nxt_cpymem(mmap_buf.buf.free,
part_start, part_size);
- rc = nxt_unit_mmap_buf_send(req->ctx,
- req_impl->recv_msg.port_msg.stream,
- &mmap_buf, 0);
+ rc = nxt_unit_mmap_buf_send(req->ctx, req_impl->stream, &mmap_buf, 0);
if (nxt_slow_path(rc != NXT_UNIT_OK)) {
nxt_unit_mmap_release(mmap_buf.hdr, mmap_buf.buf.start,
mmap_buf.buf.end - mmap_buf.buf.start);
@@ -1766,6 +2130,14 @@ nxt_unit_response_write_cb(nxt_unit_request_info_t *req,
ssize_t
nxt_unit_request_read(nxt_unit_request_info_t *req, void *dst, size_t size)
{
+ return nxt_unit_buf_read(&req->content_buf, &req->content_length,
+ dst, size);
+}
+
+
+static ssize_t
+nxt_unit_buf_read(nxt_unit_buf_t **b, uint64_t *len, void *dst, size_t size)
+{
u_char *p;
size_t rest, copy, read;
nxt_unit_buf_t *buf;
@@ -1773,7 +2145,7 @@ nxt_unit_request_read(nxt_unit_request_info_t *req, void *dst, size_t size)
p = dst;
rest = size;
- buf = req->content_buf;
+ buf = *b;
while (buf != NULL) {
copy = buf->end - buf->free;
@@ -1795,11 +2167,11 @@ nxt_unit_request_read(nxt_unit_request_info_t *req, void *dst, size_t size)
buf = nxt_unit_buf_next(buf);
}
- req->content_buf = buf;
+ *b = buf;
read = size - rest;
- req->content_length -= read;
+ *len -= read;
return read;
}
@@ -1852,7 +2224,7 @@ skip_response_send:
lib = nxt_container_of(req->unit, nxt_unit_impl_t, unit);
- msg.stream = req_impl->recv_msg.port_msg.stream;
+ msg.stream = req_impl->stream;
msg.pid = lib->pid;
msg.reply_port = 0;
msg.type = (rc == NXT_UNIT_OK) ? _NXT_PORT_MSG_DATA
@@ -1874,6 +2246,162 @@ skip_response_send:
}
+int
+nxt_unit_websocket_send(nxt_unit_request_info_t *req, uint8_t opcode,
+ uint8_t last, const void *start, size_t size)
+{
+ const struct iovec iov = { (void *) start, size };
+
+ return nxt_unit_websocket_sendv(req, opcode, last, &iov, 1);
+}
+
+
+int
+nxt_unit_websocket_sendv(nxt_unit_request_info_t *req, uint8_t opcode,
+ uint8_t last, const struct iovec *iov, int iovcnt)
+{
+ int i, rc;
+ size_t l, copy;
+ uint32_t payload_len, buf_size;
+ const uint8_t *b;
+ nxt_unit_buf_t *buf;
+ nxt_websocket_header_t *wh;
+
+ payload_len = 0;
+
+ for (i = 0; i < iovcnt; i++) {
+ payload_len += iov[i].iov_len;
+ }
+
+ buf_size = 10 + payload_len;
+
+ buf = nxt_unit_response_buf_alloc(req, nxt_min(buf_size,
+ PORT_MMAP_DATA_SIZE));
+ if (nxt_slow_path(buf == NULL)) {
+ nxt_unit_req_error(req, "Failed to allocate buf for content");
+
+ return NXT_UNIT_ERROR;
+ }
+
+ buf->start[0] = 0;
+ buf->start[1] = 0;
+
+ wh = (void *) buf->free;
+
+ buf->free = nxt_websocket_frame_init(wh, payload_len);
+ wh->fin = last;
+ wh->opcode = opcode;
+
+ for (i = 0; i < iovcnt; i++) {
+ b = iov[i].iov_base;
+ l = iov[i].iov_len;
+
+ while (l > 0) {
+ copy = buf->end - buf->free;
+ copy = nxt_min(l, copy);
+
+ buf->free = nxt_cpymem(buf->free, b, copy);
+ b += copy;
+ l -= copy;
+
+ if (l > 0) {
+ buf_size -= buf->end - buf->start;
+
+ rc = nxt_unit_buf_send(buf);
+ if (nxt_slow_path(rc != NXT_UNIT_OK)) {
+ nxt_unit_req_error(req, "Failed to send content");
+
+ return NXT_UNIT_ERROR;
+ }
+
+ buf = nxt_unit_response_buf_alloc(req, nxt_min(buf_size,
+ PORT_MMAP_DATA_SIZE));
+ if (nxt_slow_path(buf == NULL)) {
+ nxt_unit_req_error(req,
+ "Failed to allocate buf for content");
+
+ return NXT_UNIT_ERROR;
+ }
+ }
+ }
+ }
+
+ if (buf->free > buf->start) {
+ rc = nxt_unit_buf_send(buf);
+ if (nxt_slow_path(rc != NXT_UNIT_OK)) {
+ nxt_unit_req_error(req, "Failed to send content");
+ }
+ }
+
+ return rc;
+}
+
+
+ssize_t
+nxt_unit_websocket_read(nxt_unit_websocket_frame_t *ws, void *dst,
+ size_t size)
+{
+ ssize_t res;
+ uint8_t *b;
+ uint64_t i, d;
+
+ res = nxt_unit_buf_read(&ws->content_buf, &ws->content_length,
+ dst, size);
+
+ if (ws->mask == NULL) {
+ return res;
+ }
+
+ b = dst;
+ d = (ws->payload_len - ws->content_length - res) % 4;
+
+ for (i = 0; i < (uint64_t) res; i++) {
+ b[i] ^= ws->mask[ (i + d) % 4 ];
+ }
+
+ return res;
+}
+
+
+int
+nxt_unit_websocket_retain(nxt_unit_websocket_frame_t *ws)
+{
+ char *b;
+ size_t size;
+ nxt_unit_websocket_frame_impl_t *ws_impl;
+
+ ws_impl = nxt_container_of(ws, nxt_unit_websocket_frame_impl_t, ws);
+
+ if (ws_impl->retain_buf != NULL || ws_impl->buf->hdr != NULL) {
+ return NXT_UNIT_OK;
+ }
+
+ size = ws_impl->buf->buf.end - ws_impl->buf->buf.start;
+
+ b = malloc(size);
+ if (nxt_slow_path(b == NULL)) {
+ return NXT_UNIT_ERROR;
+ }
+
+ memcpy(b, ws_impl->buf->buf.start, size);
+
+ ws_impl->buf->buf.start = b;
+ ws_impl->buf->buf.free = b;
+ ws_impl->buf->buf.end = b + size;
+
+ ws_impl->retain_buf = b;
+
+ return NXT_UNIT_OK;
+}
+
+
+void
+nxt_unit_websocket_done(nxt_unit_websocket_frame_t *ws)
+{
+ nxt_unit_websocket_frame_release(ws);
+}
+
+
static nxt_port_mmap_header_t *
nxt_unit_mmap_get(nxt_unit_ctx_t *ctx, nxt_unit_process_t *process,
nxt_unit_port_id_t *port_id, nxt_chunk_id_t *c, int n)
@@ -2355,7 +2883,7 @@ nxt_unit_tracking_read(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg)
if (recv_msg->size < (int) sizeof(nxt_port_mmap_tracking_msg_t)) {
nxt_unit_warn(ctx, "#%"PRIu32": tracking_read: too small message (%d)",
- recv_msg->port_msg.stream, (int) recv_msg->size);
+ recv_msg->stream, (int) recv_msg->size);
return 0;
}
@@ -2378,18 +2906,18 @@ nxt_unit_tracking_read(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg)
nxt_unit_warn(ctx, "#%"PRIu32": tracking_read: "
"invalid mmap id %d,%"PRIu32,
- recv_msg->port_msg.stream,
- (int) process->pid, tracking_msg->mmap_id);
+ recv_msg->stream, (int) process->pid,
+ tracking_msg->mmap_id);
return 0;
}
c = tracking_msg->tracking_id;
- rc = nxt_atomic_cmp_set(hdr->tracking + c, recv_msg->port_msg.stream, 0);
+ rc = nxt_atomic_cmp_set(hdr->tracking + c, recv_msg->stream, 0);
if (rc == 0) {
nxt_unit_debug(ctx, "#%"PRIu32": tracking cancelled",
- recv_msg->port_msg.stream);
+ recv_msg->stream);
nxt_port_mmap_set_chunk_free(hdr->free_tracking_map, c);
}
@@ -2401,19 +2929,18 @@ nxt_unit_tracking_read(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg)
static int
-nxt_unit_mmap_read(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg,
- nxt_queue_t *incoming_buf)
+nxt_unit_mmap_read(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg)
{
void *start;
uint32_t size;
nxt_unit_process_t *process;
- nxt_unit_mmap_buf_t *b;
+ nxt_unit_mmap_buf_t *b, **incoming_tail;
nxt_port_mmap_msg_t *mmap_msg, *end;
nxt_port_mmap_header_t *hdr;
if (nxt_slow_path(recv_msg->size < sizeof(nxt_port_mmap_msg_t))) {
nxt_unit_warn(ctx, "#%"PRIu32": mmap_read: too small message (%d)",
- recv_msg->port_msg.stream, (int) recv_msg->size);
+ recv_msg->stream, (int) recv_msg->size);
return NXT_UNIT_ERROR;
}
@@ -2426,6 +2953,8 @@ nxt_unit_mmap_read(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg,
mmap_msg = recv_msg->start;
end = nxt_pointer_to(recv_msg->start, recv_msg->size);
+ incoming_tail = &recv_msg->incoming_buf;
+
pthread_mutex_lock(&process->incoming.mutex);
for (; mmap_msg < end; mmap_msg++) {
@@ -2435,8 +2964,8 @@ nxt_unit_mmap_read(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg,
nxt_unit_warn(ctx, "#%"PRIu32": mmap_read: "
"invalid mmap id %d,%"PRIu32,
- recv_msg->port_msg.stream,
- (int) process->pid, mmap_msg->mmap_id);
+ recv_msg->stream, (int) process->pid,
+ mmap_msg->mmap_id);
return NXT_UNIT_ERROR;
}
@@ -2453,16 +2982,16 @@ nxt_unit_mmap_read(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg,
if (nxt_slow_path(b == NULL)) {
pthread_mutex_unlock(&process->incoming.mutex);
- nxt_unit_warn(ctx, "#%"PRIu32": mmap_read: "
- "failed to allocate buf",
- recv_msg->port_msg.stream);
+ nxt_unit_warn(ctx, "#%"PRIu32": mmap_read: failed to allocate buf",
+ recv_msg->stream);
nxt_unit_mmap_release(hdr, start, size);
return NXT_UNIT_ERROR;
}
- nxt_queue_insert_tail(incoming_buf, &b->link);
+ nxt_unit_mmap_buf_insert(incoming_tail, b);
+ incoming_tail = &b->next;
b->buf.start = start;
b->buf.free = start;
@@ -2470,7 +2999,7 @@ nxt_unit_mmap_read(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg,
b->hdr = hdr;
nxt_unit_debug(ctx, "#%"PRIu32": mmap_read: [%p,%d] %d->%d,(%d,%d,%d)",
- recv_msg->port_msg.stream,
+ recv_msg->stream,
start, (int) size,
(int) hdr->src_pid, (int) hdr->dst_pid,
(int) hdr->id, (int) mmap_msg->chunk_id,
@@ -2685,6 +3214,11 @@ nxt_unit_run_once(nxt_unit_ctx_t *ctx)
if (nxt_fast_path(rsize > 0)) {
rc = nxt_unit_process_msg(ctx, &ctx_impl->read_port_id, buf, rsize,
oob, sizeof(oob));
+
+#if (NXT_DEBUG)
+ memset(buf, 0xAC, rsize);
+#endif
+
} else {
rc = NXT_UNIT_ERROR;
}
@@ -2775,10 +3309,11 @@ nxt_unit_ctx_alloc(nxt_unit_ctx_t *ctx, void *data)
void
nxt_unit_ctx_free(nxt_unit_ctx_t *ctx)
{
- nxt_unit_impl_t *lib;
- nxt_unit_ctx_impl_t *ctx_impl;
- nxt_unit_mmap_buf_t *mmap_buf;
- nxt_unit_request_info_impl_t *req_impl;
+ nxt_unit_impl_t *lib;
+ nxt_unit_ctx_impl_t *ctx_impl;
+ nxt_unit_mmap_buf_t *mmap_buf;
+ nxt_unit_request_info_impl_t *req_impl;
+ nxt_unit_websocket_frame_impl_t *ws_impl;
ctx_impl = nxt_container_of(ctx, nxt_unit_ctx_impl_t, ctx);
lib = nxt_container_of(ctx->unit, nxt_unit_impl_t, unit);
@@ -2792,15 +3327,14 @@ nxt_unit_ctx_free(nxt_unit_ctx_t *ctx)
} nxt_queue_loop;
- nxt_queue_remove(&ctx_impl->ctx_buf[0].link);
- nxt_queue_remove(&ctx_impl->ctx_buf[1].link);
-
- nxt_queue_each(mmap_buf, &ctx_impl->free_buf, nxt_unit_mmap_buf_t, link) {
+ nxt_unit_mmap_buf_remove(&ctx_impl->ctx_buf[0]);
+ nxt_unit_mmap_buf_remove(&ctx_impl->ctx_buf[1]);
- nxt_queue_remove(&mmap_buf->link);
+ while (ctx_impl->free_buf != NULL) {
+ mmap_buf = ctx_impl->free_buf;
+ nxt_unit_mmap_buf_remove(mmap_buf);
free(mmap_buf);
-
- } nxt_queue_loop;
+ }
nxt_queue_each(req_impl, &ctx_impl->free_req,
nxt_unit_request_info_impl_t, link)
@@ -2809,6 +3343,13 @@ nxt_unit_ctx_free(nxt_unit_ctx_t *ctx)
} nxt_queue_loop;
+ nxt_queue_each(ws_impl, &ctx_impl->free_ws,
+ nxt_unit_websocket_frame_impl_t, link)
+ {
+ nxt_unit_websocket_frame_free(ws_impl);
+
+ } nxt_queue_loop;
+
nxt_queue_remove(&ctx_impl->link);
if (ctx_impl != &lib->main_ctx) {
@@ -3454,6 +3995,83 @@ nxt_unit_port_hash_find(nxt_lvlhsh_t *port_hash, nxt_unit_port_id_t *port_id,
}
+static nxt_int_t
+nxt_unit_request_hash_test(nxt_lvlhsh_query_t *lhq, void *data)
+{
+ return NXT_OK;
+}
+
+
+static const nxt_lvlhsh_proto_t lvlhsh_requests_proto nxt_aligned(64) = {
+ NXT_LVLHSH_DEFAULT,
+ nxt_unit_request_hash_test,
+ nxt_lvlhsh_alloc,
+ nxt_lvlhsh_free,
+};
+
+
+static int
+nxt_unit_request_hash_add(nxt_lvlhsh_t *request_hash,
+ nxt_unit_request_info_impl_t *req_impl)
+{
+ uint32_t *stream;
+ nxt_int_t res;
+ nxt_lvlhsh_query_t lhq;
+
+ stream = &req_impl->stream;
+
+ lhq.key_hash = nxt_murmur_hash2(stream, sizeof(*stream));
+ lhq.key.length = sizeof(*stream);
+ lhq.key.start = (u_char *) stream;
+ lhq.proto = &lvlhsh_requests_proto;
+ lhq.pool = NULL;
+ lhq.replace = 0;
+ lhq.value = req_impl;
+
+ res = nxt_lvlhsh_insert(request_hash, &lhq);
+
+ switch (res) {
+
+ case NXT_OK:
+ return NXT_UNIT_OK;
+
+ default:
+ return NXT_UNIT_ERROR;
+ }
+}
+
+
+static nxt_unit_request_info_impl_t *
+nxt_unit_request_hash_find(nxt_lvlhsh_t *request_hash, uint32_t stream,
+ int remove)
+{
+ nxt_int_t res;
+ nxt_lvlhsh_query_t lhq;
+
+ lhq.key_hash = nxt_murmur_hash2(&stream, sizeof(stream));
+ lhq.key.length = sizeof(stream);
+ lhq.key.start = (u_char *) &stream;
+ lhq.proto = &lvlhsh_requests_proto;
+ lhq.pool = NULL;
+
+ if (remove) {
+ res = nxt_lvlhsh_delete(request_hash, &lhq);
+
+ } else {
+ res = nxt_lvlhsh_find(request_hash, &lhq);
+ }
+
+ switch (res) {
+
+ case NXT_OK:
+ return lhq.value;
+
+ default:
+ return NULL;
+ }
+}
+
+
void
nxt_unit_log(nxt_unit_ctx_t *ctx, int level, const char *fmt, ...)
{
@@ -3526,8 +4144,7 @@ nxt_unit_req_log(nxt_unit_request_info_t *req, int level, const char *fmt, ...)
if (nxt_fast_path(req != NULL)) {
req_impl = nxt_container_of(req, nxt_unit_request_info_impl_t, req);
- p += snprintf(p, end - p,
- "#%"PRIu32": ", req_impl->recv_msg.port_msg.stream);
+ p += snprintf(p, end - p, "#%"PRIu32": ", req_impl->stream);
}
va_start(ap, fmt);
diff --git a/src/nxt_unit.h b/src/nxt_unit.h
index 532de20d..3471a758 100644
--- a/src/nxt_unit.h
+++ b/src/nxt_unit.h
@@ -9,6 +9,7 @@
#include <inttypes.h>
#include <sys/types.h>
+#include <sys/uio.h>
#include <string.h>
#include "nxt_version.h"
@@ -106,17 +107,24 @@ struct nxt_unit_request_info_s {
void *data;
};
+
/*
* Set of application-specific callbacks. Application may leave all optional
* callbacks as NULL.
*/
struct nxt_unit_callbacks_s {
/*
- * Process request data. Unlike all other callback, this callback
+ * Process request. Unlike all other callback, this callback
* need to be defined by application.
*/
void (*request_handler)(nxt_unit_request_info_t *req);
+ /* Process websocket frame. */
+ void (*websocket_handler)(nxt_unit_websocket_frame_t *ws);
+
+ /* Connection closed. */
+ void (*close_handler)(nxt_unit_request_info_t *req);
+
/* Add new Unit port to communicate with process pid. Optional. */
int (*add_port)(nxt_unit_ctx_t *, nxt_unit_port_t *port);
@@ -293,6 +301,14 @@ int nxt_unit_response_is_sent(nxt_unit_request_info_t *req);
nxt_unit_buf_t *nxt_unit_response_buf_alloc(nxt_unit_request_info_t *req,
uint32_t size);
+int nxt_unit_request_is_websocket_handshake(nxt_unit_request_info_t *req);
+
+int nxt_unit_response_upgrade(nxt_unit_request_info_t *req);
+
+int nxt_unit_response_is_websocket(nxt_unit_request_info_t *req);
+
+nxt_unit_request_info_t *nxt_unit_get_request_info_from_data(void *data);
+
int nxt_unit_buf_send(nxt_unit_buf_t *buf);
void nxt_unit_buf_free(nxt_unit_buf_t *buf);
@@ -315,6 +331,20 @@ ssize_t nxt_unit_request_read(nxt_unit_request_info_t *req, void *dst,
void nxt_unit_request_done(nxt_unit_request_info_t *req, int rc);
+int nxt_unit_websocket_send(nxt_unit_request_info_t *req, uint8_t opcode,
+ uint8_t last, const void *start, size_t size);
+
+int nxt_unit_websocket_sendv(nxt_unit_request_info_t *req, uint8_t opcode,
+ uint8_t last, const struct iovec *iov, int iovcnt);
+
+ssize_t nxt_unit_websocket_read(nxt_unit_websocket_frame_t *ws, void *dst,
+ size_t size);
+
+int nxt_unit_websocket_retain(nxt_unit_websocket_frame_t *ws);
+
+void nxt_unit_websocket_done(nxt_unit_websocket_frame_t *ws);
+
+
void nxt_unit_log(nxt_unit_ctx_t *ctx, int level, const char* fmt, ...);
void nxt_unit_req_log(nxt_unit_request_info_t *req, int level,
diff --git a/src/nxt_unit_request.h b/src/nxt_unit_request.h
index 2207cefa..52017a42 100644
--- a/src/nxt_unit_request.h
+++ b/src/nxt_unit_request.h
@@ -20,6 +20,7 @@ struct nxt_unit_request_s {
uint8_t remote_length;
uint8_t local_length;
uint8_t tls;
+ uint8_t websocket_handshake;
uint32_t server_name_length;
uint32_t target_length;
uint32_t path_length;
diff --git a/src/nxt_unit_typedefs.h b/src/nxt_unit_typedefs.h
index 871ce25b..26e54f91 100644
--- a/src/nxt_unit_typedefs.h
+++ b/src/nxt_unit_typedefs.h
@@ -7,19 +7,20 @@
#define _NXT_UNIT_TYPEDEFS_H_INCLUDED_
-typedef struct nxt_unit_s nxt_unit_t;
-typedef struct nxt_unit_ctx_s nxt_unit_ctx_t;
-typedef struct nxt_unit_port_id_s nxt_unit_port_id_t;
-typedef struct nxt_unit_port_s nxt_unit_port_t;
-typedef struct nxt_unit_buf_s nxt_unit_buf_t;
-typedef struct nxt_unit_request_info_s nxt_unit_request_info_t;
-typedef struct nxt_unit_callbacks_s nxt_unit_callbacks_t;
-typedef struct nxt_unit_init_s nxt_unit_init_t;
-typedef union nxt_unit_sptr_u nxt_unit_sptr_t;
-typedef struct nxt_unit_field_s nxt_unit_field_t;
-typedef struct nxt_unit_request_s nxt_unit_request_t;
-typedef struct nxt_unit_response_s nxt_unit_response_t;
-typedef struct nxt_unit_read_info_s nxt_unit_read_info_t;
+typedef struct nxt_unit_s nxt_unit_t;
+typedef struct nxt_unit_ctx_s nxt_unit_ctx_t;
+typedef struct nxt_unit_port_id_s nxt_unit_port_id_t;
+typedef struct nxt_unit_port_s nxt_unit_port_t;
+typedef struct nxt_unit_buf_s nxt_unit_buf_t;
+typedef struct nxt_unit_request_info_s nxt_unit_request_info_t;
+typedef struct nxt_unit_callbacks_s nxt_unit_callbacks_t;
+typedef struct nxt_unit_init_s nxt_unit_init_t;
+typedef union nxt_unit_sptr_u nxt_unit_sptr_t;
+typedef struct nxt_unit_field_s nxt_unit_field_t;
+typedef struct nxt_unit_request_s nxt_unit_request_t;
+typedef struct nxt_unit_response_s nxt_unit_response_t;
+typedef struct nxt_unit_read_info_s nxt_unit_read_info_t;
+typedef struct nxt_unit_websocket_frame_s nxt_unit_websocket_frame_t;
#endif /* _NXT_UNIT_TYPEDEFS_H_INCLUDED_ */
diff --git a/src/nxt_unit_websocket.h b/src/nxt_unit_websocket.h
new file mode 100644
index 00000000..beb2536e
--- /dev/null
+++ b/src/nxt_unit_websocket.h
@@ -0,0 +1,27 @@
+
+/*
+ * Copyright (C) NGINX, Inc.
+ */
+
+#ifndef _NXT_UNIT_WEBSOCKET_H_INCLUDED_
+#define _NXT_UNIT_WEBSOCKET_H_INCLUDED_
+
+#include <inttypes.h>
+
+#include "nxt_unit_typedefs.h"
+#include "nxt_websocket_header.h"
+
+
+struct nxt_unit_websocket_frame_s {
+ nxt_unit_request_info_t *req;
+
+ uint64_t payload_len;
+ nxt_websocket_header_t *header;
+ uint8_t *mask;
+
+ nxt_unit_buf_t *content_buf;
+ uint64_t content_length;
+};
+
+
+#endif /* _NXT_UNIT_WEBSOCKET_H_INCLUDED_ */
diff --git a/src/nxt_websocket.c b/src/nxt_websocket.c
new file mode 100644
index 00000000..9a099760
--- /dev/null
+++ b/src/nxt_websocket.c
@@ -0,0 +1,122 @@
+
+/*
+ * Copyright (C) NGINX, Inc.
+ */
+
+#include <nxt_main.h>
+#include <nxt_websocket.h>
+#include <nxt_websocket_header.h>
+
+
+nxt_inline uint16_t
+nxt_ntoh16(const uint8_t *b)
+{
+ return ((uint16_t) b[0]) << 8 | ((uint16_t) b[1]);
+}
+
+
+nxt_inline void
+nxt_hton16(uint8_t *b, uint16_t v)
+{
+ b[0] = (v >> 8);
+ b[1] = (v & 0xFFu);
+}
+
+
+nxt_inline uint64_t
+nxt_ntoh64(const uint8_t *b)
+{
+ return ((uint64_t) b[0]) << 56
+ | ((uint64_t) b[1]) << 48
+ | ((uint64_t) b[2]) << 40
+ | ((uint64_t) b[3]) << 32
+ | ((uint64_t) b[4]) << 24
+ | ((uint64_t) b[5]) << 16
+ | ((uint64_t) b[6]) << 8
+ | ((uint64_t) b[7]);
+}
+
+
+nxt_inline void
+nxt_hton64(uint8_t *b, uint64_t v)
+{
+ b[0] = (v >> 56);
+ b[1] = (v >> 48) & 0xFFu;
+ b[2] = (v >> 40) & 0xFFu;
+ b[3] = (v >> 32) & 0xFFu;
+ b[4] = (v >> 24) & 0xFFu;
+ b[5] = (v >> 16) & 0xFFu;
+ b[6] = (v >> 8) & 0xFFu;
+ b[7] = v & 0xFFu;
+}
+
+
+size_t
+nxt_websocket_frame_header_size(const void *data)
+{
+ size_t res;
+ uint64_t p;
+ const nxt_websocket_header_t *h;
+
+ h = data;
+ p = h->payload_len;
+
+ res = 2;
+
+ if (p == 126) {
+ res += 2;
+ } else if (p == 127) {
+ res += 8;
+ }
+
+ if (h->mask) {
+ res += 4;
+ }
+
+ return res;
+}
+
+
+uint64_t
+nxt_websocket_frame_payload_len(const void *data)
+{
+ uint64_t p;
+ const nxt_websocket_header_t *h;
+
+ h = data;
+ p = h->payload_len;
+
+ if (p == 126) {
+ p = nxt_ntoh16(h->payload_len_);
+ } else if (p == 127) {
+ p = nxt_ntoh64(h->payload_len_);
+ }
+
+ return p;
+}
+
+
+void *
+nxt_websocket_frame_init(void *data, uint64_t payload_len)
+{
+ uint8_t *p;
+ nxt_websocket_header_t *h;
+
+ h = data;
+ p = data;
+
+ if (payload_len < 126) {
+ h->payload_len = payload_len;
+ return p + 2;
+ }
+
+ if (payload_len < 65536) {
+ h->payload_len = 126;
+ nxt_hton16(h->payload_len_, payload_len);
+ return p + 4;
+ }
+
+ h->payload_len = 127;
+ nxt_hton64(h->payload_len_, payload_len);
+ return p + 10;
+}
diff --git a/src/nxt_websocket.h b/src/nxt_websocket.h
new file mode 100644
index 00000000..499a3268
--- /dev/null
+++ b/src/nxt_websocket.h
@@ -0,0 +1,21 @@
+
+/*
+ * Copyright (C) NGINX, Inc.
+ */
+
+#ifndef _NXT_WEBSOCKET_H_INCLUDED_
+#define _NXT_WEBSOCKET_H_INCLUDED_
+
+
+enum {
+ NXT_WEBSOCKET_ACCEPT_SIZE = 28,
+};
+
+
+NXT_EXPORT size_t nxt_websocket_frame_header_size(const void *data);
+NXT_EXPORT uint64_t nxt_websocket_frame_payload_len(const void *data);
+NXT_EXPORT void *nxt_websocket_frame_init(void *data, uint64_t payload_len);
+NXT_EXPORT void nxt_websocket_accept(u_char *accept, const void *key);
+
+
+#endif /* _NXT_WEBSOCKET_H_INCLUDED_ */
diff --git a/src/nxt_websocket_accept.c b/src/nxt_websocket_accept.c
new file mode 100644
index 00000000..05cbcb56
--- /dev/null
+++ b/src/nxt_websocket_accept.c
@@ -0,0 +1,68 @@
+
+/*
+ * Copyright (C) NGINX, Inc.
+ */
+
+#include <nxt_main.h>
+#include <nxt_websocket.h>
+#include <nxt_sha1.h>
+
+
+static void
+nxt_websocket_base64_encode(u_char *d, const uint8_t *s, size_t len)
+{
+ u_char c0, c1, c2;
+ static u_char basis[] =
+ "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
+
+ while (len > 2) {
+ c0 = s[0];
+ c1 = s[1];
+ c2 = s[2];
+
+ *d++ = basis[c0 >> 2];
+ *d++ = basis[((c0 & 0x03) << 4) | (c1 >> 4)];
+ *d++ = basis[((c1 & 0x0f) << 2) | (c2 >> 6)];
+ *d++ = basis[c2 & 0x3f];
+
+ s += 3;
+ len -= 3;
+ }
+
+ if (len > 0) {
+ c0 = s[0];
+ *d++ = basis[c0 >> 2];
+
+ if (len == 1) {
+ *d++ = basis[(c0 & 0x03) << 4];
+ *d++ = '=';
+ *d++ = '=';
+
+ } else {
+ c1 = s[1];
+
+ *d++ = basis[((c0 & 0x03) << 4) | (c1 >> 4)];
+ *d++ = basis[(c1 & 0x0f) << 2];
+
+ *d++ = '=';
+ }
+ }
+}
+
+
+void
+nxt_websocket_accept(u_char *accept, const void *key)
+{
+ u_char bin_accept[20];
+ nxt_sha1_t ctx;
+ static const char accept_guid[] = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
+
+ nxt_sha1_init(&ctx);
+ nxt_sha1_update(&ctx, key, 24);
+ nxt_sha1_update(&ctx, accept_guid, nxt_length(accept_guid));
+ nxt_sha1_final(bin_accept, &ctx);
+
+ nxt_websocket_base64_encode(accept, bin_accept, sizeof(bin_accept));
+}
+
+
diff --git a/src/nxt_websocket_header.h b/src/nxt_websocket_header.h
new file mode 100644
index 00000000..f75dfacd
--- /dev/null
+++ b/src/nxt_websocket_header.h
@@ -0,0 +1,68 @@
+
+/*
+ * Copyright (C) NGINX, Inc.
+ */
+
+#ifndef _NXT_WEBSOCKET_HEADER_H_INCLUDED_
+#define _NXT_WEBSOCKET_HEADER_H_INCLUDED_
+
+#include <netinet/in.h>
+
+
+typedef struct {
+#if (BYTE_ORDER == BIG_ENDIAN)
+ uint8_t fin:1;
+ uint8_t rsv1:1;
+ uint8_t rsv2:1;
+ uint8_t rsv3:1;
+ uint8_t opcode:4;
+
+ uint8_t mask:1;
+ uint8_t payload_len:7;
+#endif
+
+#if (BYTE_ORDER == LITTLE_ENDIAN)
+ uint8_t opcode:4;
+ uint8_t rsv3:1;
+ uint8_t rsv2:1;
+ uint8_t rsv1:1;
+ uint8_t fin:1;
+
+ uint8_t payload_len:7;
+ uint8_t mask:1;
+#endif
+
+ uint8_t payload_len_[8];
+} nxt_websocket_header_t;
+
+
+enum {
+ NXT_WEBSOCKET_OP_CONT = 0x00,
+ NXT_WEBSOCKET_OP_TEXT = 0x01,
+ NXT_WEBSOCKET_OP_BINARY = 0x02,
+ NXT_WEBSOCKET_OP_CLOSE = 0x08,
+ NXT_WEBSOCKET_OP_PING = 0x09,
+ NXT_WEBSOCKET_OP_PONG = 0x0A,
+
+ NXT_WEBSOCKET_OP_CTRL = 0x08,
+};
+
+
+enum {
+ NXT_WEBSOCKET_CR_NORMAL = 1000,
+ NXT_WEBSOCKET_CR_GOING_AWAY = 1001,
+ NXT_WEBSOCKET_CR_PROTOCOL_ERROR = 1002,
+ NXT_WEBSOCKET_CR_UNPROCESSABLE_INPUT = 1003,
+ NXT_WEBSOCKET_CR_RESERVED = 1004,
+ NXT_WEBSOCKET_CR_NOT_PROVIDED = 1005,
+ NXT_WEBSOCKET_CR_ABNORMAL = 1006,
+ NXT_WEBSOCKET_CR_INVALID_DATA = 1007,
+ NXT_WEBSOCKET_CR_POLICY_VIOLATION = 1008,
+ NXT_WEBSOCKET_CR_MESSAGE_TOO_BIG = 1009,
+ NXT_WEBSOCKET_CR_EXTENSION_REQUIRED = 1010,
+ NXT_WEBSOCKET_CR_INTERNAL_SERVER_ERROR = 1011,
+ NXT_WEBSOCKET_CR_TLS_HANDSHAKE_FAILED = 1015,
+};
+
+
+#endif /* _NXT_WEBSOCKET_HEADER_H_INCLUDED_ */
diff --git a/src/test/nxt_unit_websocket_chat.c b/src/test/nxt_unit_websocket_chat.c
new file mode 100644
index 00000000..ecc9a243
--- /dev/null
+++ b/src/test/nxt_unit_websocket_chat.c
@@ -0,0 +1,348 @@
+
+/*
+ * Copyright (C) NGINX, Inc.
+ */
+
+#include <fcntl.h>
+#include <stdio.h>
+#include <string.h>
+#include <errno.h>
+
+#include <sys/mman.h>
+#include <sys/stat.h>
+
+#include <nxt_unit.h>
+#include <nxt_unit_request.h>
+#include <nxt_clang.h>
+#include <nxt_websocket.h>
+#include <nxt_unit_websocket.h>
+#include <nxt_main.h>
+
+
+#define CONTENT_TYPE "Content-Type"
+#define CONTENT_LENGTH "Content-Length"
+#define TEXT_HTML "text/html"
+
+typedef struct {
+ nxt_queue_link_t link;
+ int id;
+} ws_chat_request_data_t;
+
+
+static int ws_chat_root(nxt_unit_request_info_t *req);
+static void ws_chat_broadcast(const void *buf, size_t size);
+
+
+static const char ws_chat_index_html[];
+static const int ws_chat_index_html_size;
+
+static char ws_chat_index_content_length[34];
+static int ws_chat_index_content_length_size;
+
+static nxt_queue_t ws_chat_sessions;
+static int ws_chat_next_id = 0;
+
+
+static void
+ws_chat_request_handler(nxt_unit_request_info_t *req)
+{
+ static char buf[1024];
+ int buf_size;
+ int rc = NXT_UNIT_OK;
+ nxt_unit_request_t *r;
+ ws_chat_request_data_t *data;
+
+ r = req->request;
+
+ const char* target = nxt_unit_sptr_get(&r->target);
+
+ if (strcmp(target, "/") == 0) {
+ rc = ws_chat_root(req);
+ goto fail;
+ }
+
+ if (strcmp(target, "/chat") == 0) {
+ if (!nxt_unit_request_is_websocket_handshake(req)) {
+ goto notfound;
+ }
+
+ rc = nxt_unit_response_init(req, 101, 0, 0);
+ if (nxt_slow_path(rc != NXT_UNIT_OK)) {
+ goto fail;
+ }
+
+ data = req->data;
+ nxt_queue_insert_tail(&ws_chat_sessions, &data->link);
+
+ data->id = ws_chat_next_id++;
+
+ nxt_unit_response_upgrade(req);
+ nxt_unit_response_send(req);
+
+
+ buf_size = snprintf(buf, sizeof(buf), "Guest #%d has joined.", data->id);
+
+ ws_chat_broadcast(buf, buf_size);
+
+ return;
+ }
+
+notfound:
+
+ rc = nxt_unit_response_init(req, 404, 0, 0);
+
+fail:
+
+ nxt_unit_request_done(req, rc);
+}
+
+
+static int
+ws_chat_root(nxt_unit_request_info_t *req)
+{
+ int rc;
+
+ rc = nxt_unit_response_init(req, 200 /* Status code. */,
+ 2 /* Number of response headers. */,
+ nxt_length(CONTENT_TYPE) + 1
+ + nxt_length(TEXT_HTML) + 1
+ + nxt_length(CONTENT_LENGTH) + 1
+ + ws_chat_index_content_length_size + 1
+ + ws_chat_index_html_size);
+ if (nxt_slow_path(rc != NXT_UNIT_OK)) {
+ return rc;
+ }
+
+ rc = nxt_unit_response_add_field(req,
+ CONTENT_TYPE, nxt_length(CONTENT_TYPE),
+ TEXT_HTML, nxt_length(TEXT_HTML));
+ if (nxt_slow_path(rc != NXT_UNIT_OK)) {
+ return rc;
+ }
+
+ rc = nxt_unit_response_add_field(req,
+ CONTENT_LENGTH, nxt_length(CONTENT_LENGTH),
+ ws_chat_index_content_length,
+ ws_chat_index_content_length_size);
+ if (nxt_slow_path(rc != NXT_UNIT_OK)) {
+ return rc;
+ }
+
+ rc = nxt_unit_response_add_content(req, ws_chat_index_html,
+ ws_chat_index_html_size);
+ if (nxt_slow_path(rc != NXT_UNIT_OK)) {
+ return rc;
+ }
+
+ return nxt_unit_response_send(req);
+}
+
+
+static void
+ws_chat_broadcast(const void *buf, size_t size)
+{
+ ws_chat_request_data_t *data;
+ nxt_unit_request_info_t *req;
+
+ nxt_unit_debug(NULL, "broadcast: %s", buf);
+
+ nxt_queue_each(data, &ws_chat_sessions, ws_chat_request_data_t, link) {
+
+ req = nxt_unit_get_request_info_from_data(data);
+
+ nxt_unit_req_debug(req, "broadcast: %s", buf);
+
+ nxt_unit_websocket_send(req, NXT_WEBSOCKET_OP_TEXT, 1, buf, size);
+ } nxt_queue_loop;
+}
+
+
+static void
+ws_chat_websocket_handler(nxt_unit_websocket_frame_t *ws)
+{
+ int buf_size;
+ static char buf[1024];
+ ws_chat_request_data_t *data;
+
+ if (ws->header->opcode != NXT_WEBSOCKET_OP_TEXT) {
+ return;
+ }
+
+ data = ws->req->data;
+
+ buf_size = snprintf(buf, sizeof(buf), "Guest #%d: ", data->id);
+
+ buf_size += nxt_unit_websocket_read(ws, buf + buf_size,
+ nxt_min(sizeof(buf),
+ ws->content_length));
+
+ ws_chat_broadcast(buf, buf_size);
+
+ nxt_unit_websocket_done(ws);
+}
+
+
+static void
+ws_chat_close_handler(nxt_unit_request_info_t *req)
+{
+ int buf_size;
+ static char buf[1024];
+ ws_chat_request_data_t *data;
+
+ data = req->data;
+ buf_size = snprintf(buf, sizeof(buf), "Guest #%d has disconnected.",
+ data->id);
+
+ nxt_queue_remove(&data->link);
+ nxt_unit_request_done(req, NXT_UNIT_OK);
+
+ ws_chat_broadcast(buf, buf_size);
+}
+
+
+int
+main()
+{
+ nxt_unit_ctx_t *ctx;
+ nxt_unit_init_t init;
+
+ ws_chat_index_content_length_size =
+ snprintf(ws_chat_index_content_length,
+ sizeof(ws_chat_index_content_length), "%d",
+ ws_chat_index_html_size);
+
+ nxt_queue_init(&ws_chat_sessions);
+
+ memset(&init, 0, sizeof(nxt_unit_init_t));
+
+ init.callbacks.request_handler = ws_chat_request_handler;
+ init.callbacks.websocket_handler = ws_chat_websocket_handler;
+ init.callbacks.close_handler = ws_chat_close_handler;
+
+ init.request_data_size = sizeof(ws_chat_request_data_t);
+
+ ctx = nxt_unit_init(&init);
+ if (ctx == NULL) {
+ return 1;
+ }
+
+ nxt_unit_run(ctx);
+
+ nxt_unit_done(ctx);
+
+ return 0;
+}
+
+
+static const char ws_chat_index_html[] =
+"<html>\n"
+"<head>\n"
+" <title>WebSocket Chat Examples</title>\n"
+" <style type=\"text/css\">\n"
+" input#chat {\n"
+" width: 410px\n"
+" }\n"
+"\n"
+" #container {\n"
+" width: 400px;\n"
+" }\n"
+"\n"
+" #console {\n"
+" border: 1px solid #CCCCCC;\n"
+" border-right-color: #999999;\n"
+" border-bottom-color: #999999;\n"
+" height: 170px;\n"
+" overflow-y: scroll;\n"
+" padding: 5px;\n"
+" width: 100%;\n"
+" }\n"
+"\n"
+" #console p {\n"
+" padding: 0;\n"
+" margin: 0;\n"
+" }\n"
+" </style>\n"
+" <script>\n"
+" \"use strict\";\n"
+"\n"
+" var Chat = {};\n"
+"\n"
+" Chat.socket = null;\n"
+"\n"
+" Chat.connect = (function(host) {\n"
+" if ('WebSocket' in window) {\n"
+" Chat.socket = new WebSocket(host);\n"
+" } else if ('MozWebSocket' in window) {\n"
+" Chat.socket = new MozWebSocket(host);\n"
+" } else {\n"
+" Console.log('Error: WebSocket is not supported by this browser.');\n"
+" return;\n"
+" }\n"
+"\n"
+" Chat.socket.onopen = function () {\n"
+" Console.log('Info: WebSocket connection opened.');\n"
+" document.getElementById('chat').onkeydown = function(event) {\n"
+" if (event.keyCode == 13) {\n"
+" Chat.sendMessage();\n"
+" }\n"
+" };\n"
+" };\n"
+"\n"
+" Chat.socket.onclose = function () {\n"
+" document.getElementById('chat').onkeydown = null;\n"
+" Console.log('Info: WebSocket closed.');\n"
+" };\n"
+"\n"
+" Chat.socket.onmessage = function (message) {\n"
+" Console.log(message.data);\n"
+" };\n"
+" });\n"
+"\n"
+" Chat.initialize = function() {\n"
+" var proto = 'ws://';\n"
+" if (window.location.protocol == 'https:') {\n"
+" proto = 'wss://'\n"
+" }\n"
+" Chat.connect(proto + window.location.host + '/chat');\n"
+" };\n"
+"\n"
+" Chat.sendMessage = (function() {\n"
+" var message = document.getElementById('chat').value;\n"
+" if (message != '') {\n"
+" Chat.socket.send(message);\n"
+" document.getElementById('chat').value = '';\n"
+" }\n"
+" });\n"
+"\n"
+" var Console = {};\n"
+"\n"
+" Console.log = (function(message) {\n"
+" var console = document.getElementById('console');\n"
+" var p = document.createElement('p');\n"
+" p.style.wordWrap = 'break-word';\n"
+" p.innerHTML = message;\n"
+" console.appendChild(p);\n"
+" while (console.childNodes.length > 25) {\n"
+" console.removeChild(console.firstChild);\n"
+" }\n"
+" console.scrollTop = console.scrollHeight;\n"
+" });\n"
+"\n"
+" Chat.initialize();\n"
+"\n"
+" </script>\n"
+"</head>\n"
+"<body>\n"
+"<noscript><h2 style=\"color: #ff0000\">Seems your browser doesn't support Javascript! Websockets rely on Javascript being enabled. Please enable\n"
+" Javascript and reload this page!</h2></noscript>\n"
+"<div>\n"
+" <p><input type=\"text\" placeholder=\"type and press enter to chat\" id=\"chat\" /></p>\n"
+" <div id=\"container\">\n"
+" <div id=\"console\"/>\n"
+" </div>\n"
+"</div>\n"
+"</body>\n"
+"</html>\n"
+;
+
+static const int ws_chat_index_html_size = nxt_length(ws_chat_index_html);
diff --git a/src/test/nxt_unit_websocket_echo.c b/src/test/nxt_unit_websocket_echo.c
new file mode 100644
index 00000000..2a89cdc0
--- /dev/null
+++ b/src/test/nxt_unit_websocket_echo.c
@@ -0,0 +1,105 @@
+
+/*
+ * Copyright (C) NGINX, Inc.
+ */
+
+#include <string.h>
+#include <stdlib.h>
+
+#include <nxt_unit.h>
+#include <nxt_unit_request.h>
+#include <nxt_clang.h>
+#include <nxt_websocket.h>
+#include <nxt_unit_websocket.h>
+
+
+static void
+ws_echo_request_handler(nxt_unit_request_info_t *req)
+{
+ int rc;
+ const char *target;
+
+ rc = NXT_UNIT_OK;
+ target = nxt_unit_sptr_get(&req->request->target);
+
+ if (strcmp(target, "/") == 0) {
+ if (!nxt_unit_request_is_websocket_handshake(req)) {
+ goto notfound;
+ }
+
+ rc = nxt_unit_response_init(req, 101, 0, 0);
+ if (nxt_slow_path(rc != NXT_UNIT_OK)) {
+ goto fail;
+ }
+
+ nxt_unit_response_upgrade(req);
+ nxt_unit_response_send(req);
+
+ return;
+ }
+
+notfound:
+
+ rc = nxt_unit_response_init(req, 404, 0, 0);
+
+fail:
+
+ nxt_unit_request_done(req, rc);
+}
+
+
+static void
+ws_echo_websocket_handler(nxt_unit_websocket_frame_t *ws)
+{
+ uint8_t opcode;
+ ssize_t size;
+ nxt_unit_request_info_t *req;
+
+ static size_t buf_size = 0;
+ static uint8_t *buf = NULL;
+
+ if (buf_size < ws->content_length) {
+ buf = realloc(buf, ws->content_length);
+ buf_size = ws->content_length;
+ }
+
+ req = ws->req;
+ opcode = ws->header->opcode;
+
+ if (opcode == NXT_WEBSOCKET_OP_PONG) {
+ nxt_unit_websocket_done(ws);
+ return;
+ }
+
+ size = nxt_unit_websocket_read(ws, buf, ws->content_length);
+
+ nxt_unit_websocket_send(req, opcode, ws->header->fin, buf, size);
+ nxt_unit_websocket_done(ws);
+
+ if (opcode == NXT_WEBSOCKET_OP_CLOSE) {
+ nxt_unit_request_done(req, NXT_UNIT_OK);
+ }
+}
+
+
+int
+main()
+{
+ nxt_unit_ctx_t *ctx;
+ nxt_unit_init_t init;
+
+ memset(&init, 0, sizeof(nxt_unit_init_t));
+
+ init.callbacks.request_handler = ws_echo_request_handler;
+ init.callbacks.websocket_handler = ws_echo_websocket_handler;
+
+ ctx = nxt_unit_init(&init);
+ if (ctx == NULL) {
+ return 1;
+ }
+
+ nxt_unit_run(ctx);
+ nxt_unit_done(ctx);
+
+ return 0;
+}
generated by cgit (git 2.34.1) at 2024-11-19 03:26:35 +0000