//
|
// Copyright 2021 Staysail Systems, Inc. <info@staysail.tech>
|
// Copyright 2018 Capitar IT Group BV <info@capitar.com>
|
// Copyright 2019 Devolutions <info@devolutions.net>
|
//
|
// This software is supplied under the terms of the MIT License, a
|
// copy of which should be located in the distribution where this
|
// file was obtained (LICENSE.txt). A copy of the license may also be
|
// found online at https://opensource.org/licenses/MIT.
|
//
|
|
#include <stdbool.h>
|
#include <stdlib.h>
|
#include <string.h>
|
|
#include "core/nng_impl.h"
|
#include "supplemental/base64/base64.h"
|
#include "supplemental/http/http_api.h"
|
#include "supplemental/sha1/sha1.h"
|
|
#include <nng/transport/ws/websocket.h>
|
|
#include "websocket.h"
|
|
// This should be removed or handled differently in the future.
|
typedef int (*nni_ws_listen_hook)(void *, nng_http_req *, nng_http_res *);
|
|
// We have chosen to be a bit more stringent in the size of the frames that
|
// we send, while we more generously allow larger incoming frames. These
|
// may be tuned by options.
|
#define WS_DEF_RECVMAX (1U << 20) // 1MB Message limit (message mode only)
|
#define WS_DEF_MAXRXFRAME (1U << 20) // 1MB Frame size (recv)
|
#define WS_DEF_MAXTXFRAME (1U << 16) // 64KB Frame size (send)
|
|
// Alias for checking the prefix of a string.
|
#define startswith(s, t) (strncmp(s, t, strlen(t)) == 0)
|
|
// Pre-defined types for some prototypes. These are from other subsystems.
|
typedef struct ws_frame ws_frame;
|
|
typedef struct ws_header {
|
nni_list_node node;
|
char * name;
|
char * value;
|
} ws_header;
|
|
struct nni_ws {
|
nng_stream ops;
|
nni_list_node node;
|
nni_reap_node reap;
|
bool server;
|
bool closed;
|
bool ready;
|
bool wclose;
|
bool isstream;
|
bool inmsg;
|
bool send_text;
|
bool recv_text;
|
nni_mtx mtx;
|
nni_list sendq;
|
nni_list recvq;
|
nni_list txq;
|
nni_list rxq;
|
ws_frame * txframe;
|
ws_frame * rxframe;
|
nni_aio * txaio; // physical aios
|
nni_aio * rxaio;
|
nni_aio * closeaio;
|
nni_aio * httpaio;
|
nni_aio * connaio; // connect aio
|
nni_aio * useraio; // user aio, during HTTP negotiation
|
nni_http_conn * http;
|
nni_http_req * req;
|
nni_http_res * res;
|
char * reqhdrs;
|
char * reshdrs;
|
size_t maxframe;
|
size_t fragsize;
|
size_t recvmax; // largest message size
|
nni_ws_listener *listener;
|
nni_ws_dialer * dialer;
|
};
|
|
struct nni_ws_listener {
|
nng_stream_listener ops;
|
nni_http_server * server;
|
char * proto;
|
nni_mtx mtx;
|
nni_cv cv;
|
nni_list pend;
|
nni_list reply;
|
nni_list aios;
|
nng_url * url;
|
bool started;
|
bool closed;
|
bool isstream;
|
bool send_text;
|
bool recv_text;
|
nni_http_handler * handler;
|
nni_ws_listen_hook hookfn;
|
void * hookarg;
|
nni_list headers; // response headers
|
size_t maxframe;
|
size_t fragsize;
|
size_t recvmax; // largest message size
|
};
|
|
// The dialer tracks user aios in two lists. The first list is for aios
|
// waiting for the http connection to be established, while the second
|
// are waiting for the HTTP negotiation to complete. We keep two lists
|
// so we know whether to initiate another outgoing connection after the
|
// completion of an earlier connection. (We don't want to establish
|
// requests when we already have connects negotiating.)
|
struct nni_ws_dialer {
|
nng_stream_dialer ops;
|
nni_http_req * req;
|
nni_http_res * res;
|
nni_http_client * client;
|
nni_mtx mtx;
|
nni_cv cv;
|
char * proto;
|
nng_url * url;
|
nni_list wspend; // ws structures still negotiating
|
bool closed;
|
bool isstream;
|
bool send_text;
|
bool recv_text;
|
nni_list headers; // request headers
|
size_t maxframe;
|
size_t fragsize;
|
size_t recvmax;
|
};
|
|
typedef enum ws_type {
|
WS_CONT = 0x0,
|
WS_TEXT = 0x1,
|
WS_BINARY = 0x2,
|
WS_CLOSE = 0x8,
|
WS_PING = 0x9,
|
WS_PONG = 0xA,
|
} ws_type;
|
|
typedef enum ws_reason {
|
WS_CLOSE_NORMAL_CLOSE = 1000,
|
WS_CLOSE_GOING_AWAY = 1001,
|
WS_CLOSE_PROTOCOL_ERR = 1002,
|
WS_CLOSE_UNSUPP_FORMAT = 1003,
|
WS_CLOSE_INVALID_DATA = 1007,
|
WS_CLOSE_POLICY = 1008,
|
WS_CLOSE_TOO_BIG = 1009,
|
WS_CLOSE_NO_EXTENSION = 1010,
|
WS_CLOSE_INTERNAL = 1011,
|
} ws_reason;
|
|
struct ws_frame {
|
nni_list_node node;
|
uint8_t head[14]; // maximum header size
|
uint8_t mask[4]; // read by server, sent by client
|
uint8_t sdata[125]; // short data (for short frames only)
|
size_t hlen; // header length
|
size_t len; // payload length
|
enum ws_type op;
|
bool final;
|
bool masked;
|
size_t asize; // allocated size
|
uint8_t * adata;
|
uint8_t * buf;
|
nng_aio * aio;
|
};
|
|
static void ws_send_close(nni_ws *ws, uint16_t code);
|
static void ws_conn_cb(void *);
|
static void ws_close_cb(void *);
|
static void ws_read_cb(void *);
|
static void ws_write_cb(void *);
|
static void ws_close_error(nni_ws *ws, uint16_t code);
|
|
static void ws_str_free(void *);
|
static void ws_str_close(void *);
|
static void ws_str_send(void *, nng_aio *);
|
static void ws_str_recv(void *, nng_aio *);
|
static int ws_str_get(void *, const char *, void *, size_t *, nni_type);
|
static int ws_str_set(void *, const char *, const void *, size_t, nni_type);
|
|
static void ws_listener_close(void *);
|
static void ws_listener_free(void *);
|
|
static int
|
ws_check_string(const void *v, size_t sz, nni_opt_type t)
|
{
|
if ((t != NNI_TYPE_OPAQUE) && (t != NNI_TYPE_STRING)) {
|
return (NNG_EBADTYPE);
|
}
|
if (nni_strnlen(v, sz) >= sz) {
|
return (NNG_EINVAL);
|
}
|
return (0);
|
}
|
|
static int
|
ws_set_header_ext(nni_list *l, const char *n, const char *v, bool strip_dups)
|
{
|
ws_header *hdr;
|
char * nv;
|
|
if ((nv = nni_strdup(v)) == NULL) {
|
return (NNG_ENOMEM);
|
}
|
|
if (strip_dups) {
|
NNI_LIST_FOREACH (l, hdr) {
|
if (nni_strcasecmp(hdr->name, n) == 0) {
|
nni_strfree(hdr->value);
|
hdr->value = nv;
|
return (0);
|
}
|
}
|
}
|
|
if ((hdr = NNI_ALLOC_STRUCT(hdr)) == NULL) {
|
nni_strfree(nv);
|
return (NNG_ENOMEM);
|
}
|
if ((hdr->name = nni_strdup(n)) == NULL) {
|
nni_strfree(nv);
|
NNI_FREE_STRUCT(hdr);
|
return (NNG_ENOMEM);
|
}
|
hdr->value = nv;
|
nni_list_append(l, hdr);
|
return (0);
|
}
|
|
static int
|
ws_set_header(nni_list *l, const char *n, const char *v)
|
{
|
return (ws_set_header_ext(l, n, v, true));
|
}
|
|
static int
|
ws_set_headers(nni_list *l, const char *str)
|
{
|
char * dupstr;
|
size_t duplen;
|
char * n;
|
char * v;
|
char * nl;
|
int rv;
|
|
if ((dupstr = nni_strdup(str)) == NULL) {
|
return (NNG_ENOMEM);
|
}
|
duplen = strlen(dupstr) + 1; // so we can free it later
|
|
n = dupstr;
|
for (;;) {
|
if ((v = strchr(n, ':')) == NULL) {
|
// Note that this also means that if
|
// a bare word is present, we ignore it.
|
break;
|
}
|
*v = '\0';
|
v++;
|
while (*v == ' ') {
|
// Skip leading whitespace. Not strictly
|
// necessary, but still a good idea.
|
v++;
|
}
|
nl = v;
|
// Find the end of the line -- should be CRLF, but can
|
// also be unterminated or just LF if user
|
while ((*nl != '\0') && (*nl != '\r') && (*nl != '\n')) {
|
nl++;
|
}
|
while ((*nl == '\r') || (*nl == '\n')) {
|
*nl = '\0';
|
nl++;
|
}
|
|
// Note that this can lead to a partial failure. As this
|
// is most likely ENOMEM, don't worry too much about it.
|
// This method does *not* eliminate duplicates.
|
if ((rv = ws_set_header_ext(l, n, v, false)) != 0) {
|
goto done;
|
}
|
|
// Advance to the next name.
|
n = nl;
|
}
|
|
rv = 0;
|
|
done:
|
nni_free(dupstr, duplen);
|
return (rv);
|
}
|
|
// This looks, case independently for a word in a list, which is either
|
// space or comma separated.
|
static bool
|
ws_contains_word(const char *phrase, const char *word)
|
{
|
size_t len = strlen(word);
|
|
while ((phrase != NULL) && (*phrase != '\0')) {
|
if ((nni_strncasecmp(phrase, word, len) == 0) &&
|
((phrase[len] == 0) || (phrase[len] == ' ') ||
|
(phrase[len] == ','))) {
|
return (true);
|
}
|
// Skip to next word.
|
if ((phrase = strchr(phrase, ' ')) != NULL) {
|
while ((*phrase == ' ') || (*phrase == ',')) {
|
phrase++;
|
}
|
}
|
}
|
return (false);
|
}
|
|
// input is base64 challenge, output is the accepted. input should be
|
// 24 character base 64, output is 28 character base64 reply. (output
|
// must be large enough to hold 29 bytes to allow for termination.)
|
// Returns 0 on success, NNG_EINVAL if the input is malformed somehow.
|
static int
|
ws_make_accept(const char *key, char *accept)
|
{
|
uint8_t digest[20];
|
nni_sha1_ctx ctx;
|
|
#define WS_KEY_GUID "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
|
#define WS_KEY_GUIDLEN 36
|
|
if (strlen(key) != 24) {
|
return (NNG_EINVAL);
|
}
|
|
nni_sha1_init(&ctx);
|
nni_sha1_update(&ctx, key, 24);
|
nni_sha1_update(&ctx, WS_KEY_GUID, WS_KEY_GUIDLEN);
|
nni_sha1_final(&ctx, digest);
|
|
nni_base64_encode(digest, 20, accept, 28);
|
accept[28] = '\0';
|
return (0);
|
}
|
|
static void
|
ws_frame_fini(ws_frame *frame)
|
{
|
if (frame->asize != 0) {
|
nni_free(frame->adata, frame->asize);
|
}
|
NNI_FREE_STRUCT(frame);
|
}
|
|
static void
|
ws_mask_frame(ws_frame *frame)
|
{
|
uint32_t r;
|
// frames sent by client need mask.
|
if (frame->masked) {
|
return;
|
}
|
r = nni_random();
|
NNI_PUT32(frame->mask, r);
|
for (size_t i = 0; i < frame->len; i++) {
|
frame->buf[i] ^= frame->mask[i % 4];
|
}
|
memcpy(frame->head + frame->hlen, frame->mask, 4);
|
frame->hlen += 4;
|
frame->head[1] |= 0x80; // set masked bit
|
frame->masked = true;
|
}
|
|
static void
|
ws_unmask_frame(ws_frame *frame)
|
{
|
// frames sent by client need mask.
|
if (!frame->masked) {
|
return;
|
}
|
for (size_t i = 0; i < frame->len; i++) {
|
frame->buf[i] ^= frame->mask[i % 4];
|
}
|
frame->hlen -= 4;
|
frame->head[1] &= 0x7f; // clear masked bit
|
frame->masked = false;
|
}
|
|
static int
|
ws_msg_init_control(
|
ws_frame **framep, nni_ws *ws, uint8_t op, const uint8_t *buf, size_t len)
|
{
|
ws_frame *frame;
|
|
if (len > 125) {
|
return (NNG_EINVAL);
|
}
|
|
if ((frame = NNI_ALLOC_STRUCT(frame)) == NULL) {
|
return (NNG_ENOMEM);
|
}
|
|
memcpy(frame->sdata, buf, len);
|
frame->len = len;
|
frame->final = true;
|
frame->op = op;
|
frame->head[0] = op | 0x80; // final frame (control)
|
frame->head[1] = len & 0x7F;
|
frame->hlen = 2;
|
frame->buf = frame->sdata;
|
frame->asize = 0;
|
|
if (ws->server) {
|
frame->masked = false;
|
} else {
|
ws_mask_frame(frame);
|
}
|
|
*framep = frame;
|
return (0);
|
}
|
|
static int
|
ws_frame_prep_tx(nni_ws *ws, ws_frame *frame)
|
{
|
nng_aio *aio = frame->aio;
|
nni_iov *iov;
|
unsigned niov;
|
size_t len;
|
uint8_t *buf;
|
|
// Figure out how much we need for the entire aio.
|
frame->len = 0;
|
nni_aio_get_iov(aio, &niov, &iov);
|
for (unsigned i = 0; i < niov; i++) {
|
frame->len += iov[i].iov_len;
|
}
|
|
if ((frame->len > ws->fragsize) && (ws->fragsize > 0)) {
|
// Limit it to a single frame per policy (fragsize), as needed.
|
frame->len = ws->fragsize;
|
// For stream mode, we constrain ourselves to one frame
|
// per message. Submitter may see a partial transmit, and
|
// should resubmit as needed. For message mode, we will
|
// continue to resubmit.
|
frame->final = ws->isstream ? true : false;
|
} else {
|
// It all fits in this frame (which might not be the first),
|
// so we're done.
|
frame->final = true;
|
}
|
// Potentially allocate space for the data if we need to.
|
// Note that an empty message is legal.
|
if ((frame->asize < frame->len) && (frame->len > 0)) {
|
nni_free(frame->adata, frame->asize);
|
frame->adata = nni_alloc(frame->len);
|
if (frame->adata == NULL) {
|
frame->asize = 0;
|
return (NNG_ENOMEM);
|
}
|
frame->asize = frame->len;
|
frame->buf = frame->adata;
|
}
|
buf = frame->buf;
|
|
// Now copy the data into the frame.
|
len = frame->len;
|
while (len != 0) {
|
size_t n = len;
|
if (n > iov->iov_len) {
|
n = iov->iov_len;
|
}
|
memcpy(buf, iov->iov_buf, n);
|
iov++;
|
len -= n;
|
buf += n;
|
}
|
|
if (nni_aio_count(aio) == 0) {
|
// This is the first frame.
|
if (ws->send_text) {
|
frame->op = WS_TEXT;
|
} else {
|
frame->op = WS_BINARY;
|
}
|
} else {
|
frame->op = WS_CONT;
|
}
|
|
// Populate the frame header.
|
frame->head[0] = frame->op;
|
frame->hlen = 2;
|
if (frame->final) {
|
frame->head[0] |= 0x80; // final frame bit
|
}
|
if (frame->len < 126) {
|
frame->head[1] = frame->len & 0x7f;
|
} else if (frame->len < 65536) {
|
frame->head[1] = 126;
|
NNI_PUT16(frame->head + 2, (frame->len & 0xffff));
|
frame->hlen += 2;
|
} else {
|
frame->head[1] = 127;
|
NNI_PUT64(frame->head + 2, (uint64_t) frame->len);
|
frame->hlen += 8;
|
}
|
|
// If we are on the client, then we need to mask the frame.
|
frame->masked = false;
|
if (!ws->server) {
|
ws_mask_frame(frame);
|
}
|
return (0);
|
}
|
|
static void
|
ws_close_cb(void *arg)
|
{
|
nni_ws * ws = arg;
|
ws_frame *frame;
|
|
nni_aio_close(ws->txaio);
|
nni_aio_close(ws->rxaio);
|
nni_aio_close(ws->httpaio);
|
|
// Either we sent a close frame, or we didn't. Either way,
|
// we are done, and its time to abort everything else.
|
nni_mtx_lock(&ws->mtx);
|
|
nni_http_conn_close(ws->http);
|
|
while ((frame = nni_list_first(&ws->txq)) != NULL) {
|
nni_list_remove(&ws->txq, frame);
|
if (frame->aio != NULL) {
|
nni_aio_list_remove(frame->aio);
|
nni_aio_finish_error(frame->aio, NNG_ECLOSED);
|
}
|
ws_frame_fini(frame);
|
}
|
|
// Any txframe should have been killed with its wmsg.
|
nni_mtx_unlock(&ws->mtx);
|
}
|
|
static void
|
ws_close(nni_ws *ws, uint16_t code)
|
{
|
nng_aio *aio;
|
|
// Receive stuff gets aborted always. No further receives
|
// once we get a close.
|
while ((aio = nni_list_first(&ws->recvq)) != NULL) {
|
nni_aio_list_remove(aio);
|
nni_aio_finish_error(aio, NNG_ECLOSED);
|
}
|
|
// If were closing "gracefully", then don't abort in-flight
|
// stuff yet. Note that reads should have stopped already.
|
// However, we *do* abort any inflight HTTP negotiation, or
|
// pending connect request.
|
if (!ws->closed) {
|
// ABORT connection negotiation.
|
nni_aio_close(ws->connaio);
|
nni_aio_close(ws->httpaio);
|
ws_send_close(ws, code);
|
}
|
}
|
|
static void
|
ws_start_write(nni_ws *ws)
|
{
|
ws_frame *frame;
|
nni_iov iov[2];
|
int niov;
|
|
if ((ws->txframe != NULL) || (!ws->ready)) {
|
return; // busy
|
}
|
|
if ((frame = nni_list_first(&ws->txq)) == NULL) {
|
return; // nothing to send
|
}
|
NNI_ASSERT(frame != NULL);
|
nni_list_remove(&ws->txq, frame);
|
|
// Push it out.
|
ws->txframe = frame;
|
niov = 1;
|
iov[0].iov_len = frame->hlen;
|
iov[0].iov_buf = frame->head;
|
if (frame->len > 0) {
|
niov++;
|
iov[1].iov_len = frame->len;
|
iov[1].iov_buf = frame->buf;
|
}
|
nni_aio_set_iov(ws->txaio, niov, iov);
|
nni_http_write_full(ws->http, ws->txaio);
|
}
|
|
static void
|
ws_cancel_close(nni_aio *aio, void *arg, int rv)
|
{
|
nni_ws *ws = arg;
|
nni_mtx_lock(&ws->mtx);
|
if (ws->wclose) {
|
ws->wclose = false;
|
nni_aio_finish_error(aio, rv);
|
}
|
nni_mtx_unlock(&ws->mtx);
|
}
|
|
static void
|
ws_write_cb(void *arg)
|
{
|
nni_ws * ws = arg;
|
ws_frame *frame;
|
nni_aio * aio;
|
int rv;
|
|
nni_mtx_lock(&ws->mtx);
|
|
if ((frame = ws->txframe) == NULL) {
|
nni_mtx_unlock(&ws->mtx);
|
return;
|
}
|
ws->txframe = NULL;
|
|
if (frame->op == WS_CLOSE) {
|
// If this was a close frame, we are done.
|
// No other messages may succeed..
|
ws->txframe = NULL;
|
ws_frame_fini(frame);
|
while ((frame = nni_list_first(&ws->txq)) != NULL) {
|
nni_list_remove(&ws->txq, frame);
|
if ((aio = frame->aio) != NULL) {
|
frame->aio = NULL;
|
nni_aio_list_remove(aio);
|
nni_aio_finish_error(aio, NNG_ECLOSED);
|
ws_frame_fini(frame);
|
}
|
}
|
if (ws->wclose) {
|
ws->wclose = false;
|
nni_aio_finish(ws->closeaio, 0, 0);
|
}
|
nni_mtx_unlock(&ws->mtx);
|
return;
|
}
|
|
aio = frame->aio;
|
if ((rv = nni_aio_result(ws->txaio)) != 0) {
|
frame->aio = NULL;
|
if (aio != NULL) {
|
nni_aio_list_remove(aio);
|
nni_aio_finish_error(aio, rv);
|
}
|
ws_frame_fini(frame);
|
ws->closed = true;
|
nni_http_conn_close(ws->http);
|
nni_mtx_unlock(&ws->mtx);
|
return;
|
}
|
|
if (aio != NULL) {
|
nni_aio_iov_advance(aio, frame->len);
|
nni_aio_bump_count(aio, frame->len);
|
if (frame->final) {
|
frame->aio = NULL;
|
nni_aio_list_remove(aio);
|
} else {
|
// Clear the aio so that we won't attempt to finish
|
// it outside the lock
|
aio = NULL;
|
}
|
}
|
|
if (frame->final) {
|
ws_frame_fini(frame);
|
} else {
|
// This one cannot fail here, since we only do allocation
|
// at initial scheduling.
|
ws_frame_prep_tx(ws, frame);
|
// Schedule at end. This permits other frames to interleave.
|
nni_list_append(&ws->txq, frame);
|
}
|
|
ws_start_write(ws);
|
nni_mtx_unlock(&ws->mtx);
|
|
// We attempt to finish the operation synchronously, outside the lock.
|
if (aio != NULL) {
|
nni_msg *msg;
|
// Successful send, don't leak the message!
|
if ((msg = nni_aio_get_msg(aio)) != NULL) {
|
nni_aio_set_msg(aio, NULL);
|
nni_msg_free(msg);
|
}
|
nni_aio_finish_sync(aio, 0, nni_aio_count(aio));
|
}
|
}
|
|
static void
|
ws_write_cancel(nni_aio *aio, void *arg, int rv)
|
{
|
nni_ws * ws = arg;
|
ws_frame *frame;
|
|
// Is this aio active? We can tell by looking at the active tx frame.
|
|
nni_mtx_lock(&ws->mtx);
|
if (!nni_aio_list_active(aio)) {
|
nni_mtx_unlock(&ws->mtx);
|
return;
|
}
|
frame = nni_aio_get_prov_extra(aio, 0);
|
if (frame == ws->txframe) {
|
nni_aio_abort(ws->txaio, rv);
|
// We will wait for callback on the txaio to finish aio.
|
} else {
|
// If scheduled, just need to remove node and complete it.
|
nni_list_remove(&ws->txq, frame);
|
frame->aio = NULL;
|
nni_aio_list_remove(aio);
|
nni_aio_finish_error(aio, rv);
|
ws_frame_fini(frame);
|
}
|
nni_mtx_unlock(&ws->mtx);
|
}
|
|
static void
|
ws_send_close(nni_ws *ws, uint16_t code)
|
{
|
ws_frame *frame;
|
uint8_t buf[sizeof(uint16_t)];
|
int rv;
|
nni_aio * aio;
|
|
NNI_PUT16(buf, code);
|
|
if (ws->closed || !ws->ready) {
|
return;
|
}
|
ws->closed = true;
|
aio = ws->closeaio;
|
|
if (nni_aio_begin(aio) != 0) {
|
return;
|
}
|
ws->wclose = true;
|
rv = ws_msg_init_control(&frame, ws, WS_CLOSE, buf, sizeof(buf));
|
if (rv != 0) {
|
ws->wclose = false;
|
nni_aio_finish_error(aio, rv);
|
return;
|
}
|
if ((rv = nni_aio_schedule(aio, ws_cancel_close, ws)) != 0) {
|
ws->wclose = false;
|
nni_aio_finish_error(aio, rv);
|
ws_frame_fini(frame);
|
return;
|
}
|
// This gets inserted at the head.
|
nni_list_prepend(&ws->txq, frame);
|
ws_start_write(ws);
|
}
|
|
static void
|
ws_send_control(nni_ws *ws, uint8_t op, uint8_t *buf, size_t len)
|
{
|
ws_frame *frame;
|
|
// Note that we do not care if this works or not. So no AIO needed.
|
|
if ((ws->closed) ||
|
(ws_msg_init_control(&frame, ws, op, buf, len) != 0)) {
|
return;
|
}
|
|
// Control frames at head of list. (Note that this may preempt
|
// the close frame or other ping/pong requests. Oh well.)
|
nni_list_prepend(&ws->txq, frame);
|
ws_start_write(ws);
|
}
|
|
static void
|
ws_start_read(nni_ws *ws)
|
{
|
ws_frame *frame;
|
nni_aio * aio;
|
nni_iov iov;
|
|
if ((ws->rxframe != NULL) || ws->closed) {
|
return; // already reading or closed
|
}
|
|
// If nobody is waiting for recv, and we already have a data
|
// frame, stop reading. This keeps us from buffering infinitely.
|
if (nni_list_empty(&ws->recvq) && !nni_list_empty(&ws->rxq)) {
|
return;
|
}
|
|
if ((frame = NNI_ALLOC_STRUCT(frame)) == NULL) {
|
if ((aio = nni_list_first(&ws->recvq)) != NULL) {
|
nni_aio_list_remove(aio);
|
nni_aio_finish_error(aio, NNG_ENOMEM);
|
}
|
ws_close(ws, WS_CLOSE_INTERNAL);
|
return;
|
}
|
|
// Note that the frame is *not* associated with the message
|
// as yet, because we don't know if that's right until we receive it.
|
frame->hlen = 0;
|
frame->len = 0;
|
ws->rxframe = frame;
|
|
aio = ws->rxaio;
|
iov.iov_len = 2; // We want the first two bytes.
|
iov.iov_buf = frame->head;
|
nni_aio_set_iov(aio, 1, &iov);
|
nni_http_read_full(ws->http, aio);
|
}
|
|
static void
|
ws_read_finish_str(nni_ws *ws)
|
{
|
for (;;) {
|
nni_aio * aio;
|
nni_iov * iov;
|
unsigned niov;
|
ws_frame *frame;
|
|
if ((aio = nni_list_first(&ws->recvq)) == NULL) {
|
return;
|
}
|
|
if ((frame = nni_list_first(&ws->rxq)) == NULL) {
|
return;
|
}
|
|
// Discard 0 length frames -- in stream mode they are not used.
|
if (frame->len == 0) {
|
nni_list_remove(&ws->rxq, frame);
|
ws_frame_fini(frame);
|
continue;
|
}
|
|
// We are completing this aio one way or the other.
|
nni_aio_list_remove(aio);
|
nni_aio_get_iov(aio, &niov, &iov);
|
|
while ((frame != NULL) && (niov != 0)) {
|
size_t n;
|
|
if ((n = frame->len) > iov->iov_len) {
|
// This eats the entire iov.
|
n = iov->iov_len;
|
}
|
if (n != 0) {
|
memcpy(iov->iov_buf, frame->buf, n);
|
iov->iov_buf = ((uint8_t *) iov->iov_buf) + n;
|
iov->iov_len -= n;
|
if (iov->iov_len == 0) {
|
iov++;
|
niov--;
|
}
|
}
|
|
if (frame->len == n) {
|
nni_list_remove(&ws->rxq, frame);
|
ws_frame_fini(frame);
|
frame = nni_list_first(&ws->rxq);
|
} else {
|
frame->len -= n;
|
frame->buf += n;
|
}
|
|
nni_aio_bump_count(aio, n);
|
}
|
|
nni_aio_finish(aio, 0, nni_aio_count(aio));
|
}
|
}
|
|
static void
|
ws_read_finish_msg(nni_ws *ws)
|
{
|
nni_aio * aio;
|
size_t len;
|
ws_frame *frame;
|
nni_msg * msg;
|
int rv;
|
uint8_t * body;
|
|
// If we have no data, no waiter, or have not received the complete
|
// message yet, then there is nothing to do.
|
if (ws->inmsg || nni_list_empty(&ws->rxq) ||
|
((aio = nni_list_first(&ws->recvq)) == NULL)) {
|
return;
|
}
|
|
// At this point, we have both a complete message in the queue (and
|
// there should not be any frames other than the for the message),
|
// and a waiting reader.
|
len = 0;
|
NNI_LIST_FOREACH (&ws->rxq, frame) {
|
len += frame->len;
|
}
|
|
nni_aio_list_remove(aio);
|
|
if ((rv = nni_msg_alloc(&msg, len)) != 0) {
|
nni_aio_finish_error(aio, rv);
|
ws_close_error(ws, WS_CLOSE_INTERNAL);
|
return;
|
}
|
body = nni_msg_body(msg);
|
while ((frame = nni_list_first(&ws->rxq)) != NULL) {
|
nni_list_remove(&ws->rxq, frame);
|
memcpy(body, frame->buf, frame->len);
|
body += frame->len;
|
ws_frame_fini(frame);
|
}
|
|
nni_aio_set_msg(aio, msg);
|
nni_aio_bump_count(aio, nni_msg_len(msg));
|
nni_aio_finish(aio, 0, nni_msg_len(msg));
|
}
|
|
static void
|
ws_read_finish(nni_ws *ws)
|
{
|
if (ws->isstream) {
|
ws_read_finish_str(ws);
|
} else {
|
ws_read_finish_msg(ws);
|
}
|
}
|
|
static void
|
ws_read_frame_cb(nni_ws *ws, ws_frame *frame)
|
{
|
switch (frame->op) {
|
case WS_CONT:
|
if (!ws->inmsg) {
|
ws_close(ws, WS_CLOSE_PROTOCOL_ERR);
|
return;
|
}
|
if (frame->final) {
|
ws->inmsg = false;
|
}
|
ws->rxframe = NULL;
|
nni_list_append(&ws->rxq, frame);
|
break;
|
case WS_TEXT:
|
if (!ws->recv_text) {
|
// No support for text mode at present.
|
ws_close(ws, WS_CLOSE_UNSUPP_FORMAT);
|
}
|
// FALLTHROUGH
|
case WS_BINARY:
|
if (ws->inmsg) {
|
ws_close(ws, WS_CLOSE_PROTOCOL_ERR);
|
return;
|
}
|
if (!frame->final) {
|
ws->inmsg = true;
|
}
|
ws->rxframe = NULL;
|
nni_list_append(&ws->rxq, frame);
|
break;
|
|
case WS_PING:
|
if (frame->len > 125) {
|
ws_close(ws, WS_CLOSE_PROTOCOL_ERR);
|
return;
|
}
|
ws_send_control(ws, WS_PONG, frame->buf, frame->len);
|
ws->rxframe = NULL;
|
ws_frame_fini(frame);
|
break;
|
case WS_PONG:
|
if (frame->len > 125) {
|
ws_close(ws, WS_CLOSE_PROTOCOL_ERR);
|
return;
|
}
|
ws->rxframe = NULL;
|
ws_frame_fini(frame);
|
break;
|
case WS_CLOSE:
|
ws->closed = true; // no need to send close reply
|
ws_close(ws, 0);
|
return;
|
default:
|
ws_close(ws, WS_CLOSE_PROTOCOL_ERR);
|
return;
|
}
|
|
ws_read_finish(ws);
|
}
|
|
static void
|
ws_read_cb(void *arg)
|
{
|
nni_ws * ws = arg;
|
nni_aio * aio = ws->rxaio;
|
ws_frame *frame;
|
int rv;
|
|
nni_mtx_lock(&ws->mtx);
|
if ((frame = ws->rxframe) == NULL) {
|
nni_mtx_unlock(&ws->mtx); // canceled during close
|
return;
|
}
|
|
if ((rv = nni_aio_result(aio)) != 0) {
|
ws->closed = true; // do not send a close frame
|
ws_close(ws, 0);
|
nni_mtx_unlock(&ws->mtx);
|
return;
|
}
|
|
if (frame->hlen == 0) {
|
frame->hlen = 2;
|
frame->op = frame->head[0] & 0x7fu;
|
frame->final = (frame->head[0] & 0x80u) ? 1 : 0;
|
frame->masked = (frame->head[1] & 0x80u) ? 1 : 0;
|
if (frame->masked) {
|
frame->hlen += 4;
|
}
|
if ((frame->head[1] & 0x7Fu) == 127) {
|
frame->hlen += 8;
|
} else if ((frame->head[1] & 0x7Fu) == 126) {
|
frame->hlen += 2;
|
}
|
|
// If we didn't read the full header yet, then read
|
// the rest of it.
|
if (frame->hlen != 2) {
|
nni_iov iov;
|
iov.iov_buf = frame->head + 2;
|
iov.iov_len = frame->hlen - 2;
|
nni_aio_set_iov(aio, 1, &iov);
|
nni_http_read_full(ws->http, aio);
|
nni_mtx_unlock(&ws->mtx);
|
return;
|
}
|
}
|
|
// If we are returning from a read of additional data, then
|
// the buf will be set. Otherwise we need to determine
|
// how much data to read. As our headers are complete, we take
|
// this time to do some protocol checks -- no point in waiting
|
// to read data. (Frame size check needs to be done first
|
// anyway to prevent DoS.)
|
|
if (frame->buf == NULL) {
|
|
// Determine expected frame size.
|
switch ((frame->len = (frame->head[1] & 0x7Fu))) {
|
case 127:
|
NNI_GET64(frame->head + 2, frame->len);
|
if (frame->len < 65536) {
|
ws_close(ws, WS_CLOSE_PROTOCOL_ERR);
|
nni_mtx_unlock(&ws->mtx);
|
return;
|
}
|
break;
|
case 126:
|
NNI_GET16(frame->head + 2, frame->len);
|
if (frame->len < 126) {
|
ws_close(ws, WS_CLOSE_PROTOCOL_ERR);
|
nni_mtx_unlock(&ws->mtx);
|
return;
|
}
|
|
break;
|
}
|
|
if ((frame->len > ws->maxframe) && (ws->maxframe > 0)) {
|
ws_close(ws, WS_CLOSE_TOO_BIG);
|
nni_mtx_unlock(&ws->mtx);
|
return;
|
}
|
// For message mode, also check to make sure that the overall
|
// length of the message has not exceeded our recvmax.
|
// (Protect against an infinite stream of small messages!)
|
if ((!ws->isstream) && (ws->recvmax > 0)) {
|
size_t totlen = frame->len;
|
ws_frame *fr2;
|
NNI_LIST_FOREACH (&ws->rxq, fr2) {
|
totlen += fr2->len;
|
}
|
if (totlen > ws->recvmax) {
|
ws_close(ws, WS_CLOSE_TOO_BIG);
|
nni_mtx_unlock(&ws->mtx);
|
return;
|
}
|
}
|
|
// Check for masking. (We don't actually do the unmask
|
// here, because we don't have data yet.)
|
if (frame->masked) {
|
memcpy(frame->mask, frame->head + frame->hlen - 4, 4);
|
if (!ws->server) {
|
ws_close(ws, WS_CLOSE_PROTOCOL_ERR);
|
nni_mtx_unlock(&ws->mtx);
|
return;
|
}
|
} else if (ws->server) {
|
ws_close(ws, WS_CLOSE_PROTOCOL_ERR);
|
nni_mtx_unlock(&ws->mtx);
|
return;
|
}
|
|
// If we expected data, then ask for it.
|
if (frame->len != 0) {
|
|
nni_iov iov;
|
|
// Short frames can avoid an alloc
|
if (frame->len < 126) {
|
frame->buf = frame->sdata;
|
frame->asize = 0;
|
} else {
|
frame->adata = nni_alloc(frame->len);
|
if (frame->adata == NULL) {
|
ws_close(ws, WS_CLOSE_INTERNAL);
|
nni_mtx_unlock(&ws->mtx);
|
return;
|
}
|
frame->asize = frame->len;
|
frame->buf = frame->adata;
|
}
|
|
iov.iov_buf = frame->buf;
|
iov.iov_len = frame->len;
|
nni_aio_set_iov(aio, 1, &iov);
|
nni_http_read_full(ws->http, aio);
|
nni_mtx_unlock(&ws->mtx);
|
return;
|
}
|
}
|
|
// At this point, we have a complete frame.
|
ws_unmask_frame(frame); // idempotent
|
|
ws_read_frame_cb(ws, frame);
|
ws_start_read(ws);
|
nni_mtx_unlock(&ws->mtx);
|
}
|
|
static void
|
ws_read_cancel(nni_aio *aio, void *arg, int rv)
|
{
|
nni_ws *ws = arg;
|
|
nni_mtx_lock(&ws->mtx);
|
if (nni_aio_list_active(aio)) {
|
nni_aio_list_remove(aio);
|
nni_aio_finish_error(aio, rv);
|
}
|
nni_mtx_unlock(&ws->mtx);
|
}
|
|
static void
|
ws_close_error(nni_ws *ws, uint16_t code)
|
{
|
nni_mtx_lock(&ws->mtx);
|
ws_close(ws, code);
|
nni_mtx_unlock(&ws->mtx);
|
}
|
|
static void
|
ws_fini(void *arg)
|
{
|
nni_ws * ws = arg;
|
ws_frame *frame;
|
nng_aio * aio;
|
|
ws_close_error(ws, WS_CLOSE_NORMAL_CLOSE);
|
|
// Give a chance for the close frame to drain.
|
if (ws->closeaio) {
|
nni_aio_wait(ws->closeaio);
|
}
|
|
nni_aio_stop(ws->rxaio);
|
nni_aio_stop(ws->txaio);
|
nni_aio_stop(ws->closeaio);
|
nni_aio_stop(ws->httpaio);
|
nni_aio_stop(ws->connaio);
|
|
if (nni_list_node_active(&ws->node)) {
|
nni_ws_dialer *d;
|
|
if ((d = ws->dialer) != NULL) {
|
nni_mtx_lock(&d->mtx);
|
nni_list_node_remove(&ws->node);
|
ws->dialer = NULL;
|
nni_mtx_unlock(&d->mtx);
|
}
|
}
|
|
nni_mtx_lock(&ws->mtx);
|
while ((frame = nni_list_first(&ws->rxq)) != NULL) {
|
nni_list_remove(&ws->rxq, frame);
|
ws_frame_fini(frame);
|
}
|
|
while ((frame = nni_list_first(&ws->txq)) != NULL) {
|
nni_list_remove(&ws->txq, frame);
|
ws_frame_fini(frame);
|
}
|
|
if (ws->rxframe != NULL) {
|
ws_frame_fini(ws->rxframe);
|
}
|
if (ws->txframe != NULL) {
|
ws_frame_fini(ws->txframe);
|
}
|
|
while (((aio = nni_list_first(&ws->recvq)) != NULL) ||
|
((aio = nni_list_first(&ws->sendq)) != NULL)) {
|
nni_aio_list_remove(aio);
|
nni_aio_finish_error(aio, NNG_ECLOSED);
|
}
|
|
nni_mtx_unlock(&ws->mtx);
|
|
if (ws->http) {
|
nni_http_conn_fini(ws->http);
|
}
|
if (ws->req) {
|
nni_http_req_free(ws->req);
|
}
|
if (ws->res) {
|
nni_http_res_free(ws->res);
|
}
|
|
nni_strfree(ws->reqhdrs);
|
nni_strfree(ws->reshdrs);
|
nni_aio_free(ws->rxaio);
|
nni_aio_free(ws->txaio);
|
nni_aio_free(ws->closeaio);
|
nni_aio_free(ws->httpaio);
|
nni_aio_free(ws->connaio);
|
nni_mtx_fini(&ws->mtx);
|
NNI_FREE_STRUCT(ws);
|
}
|
|
static nni_reap_list ws_reap_list = {
|
.rl_offset = offsetof(nni_ws, reap),
|
.rl_func = ws_fini,
|
};
|
|
static void
|
ws_reap(nni_ws *ws)
|
{
|
nni_reap(&ws_reap_list, ws);
|
}
|
|
static void
|
ws_http_cb_listener(nni_ws *ws, nni_aio *aio)
|
{
|
nni_ws_listener *l = ws->listener;
|
|
nni_mtx_lock(&l->mtx);
|
nni_list_remove(&l->reply, ws);
|
if (nni_aio_result(aio) != 0) {
|
nni_mtx_unlock(&l->mtx);
|
ws_reap(ws);
|
return;
|
}
|
ws->ready = true;
|
if ((aio = nni_list_first(&l->aios)) != NULL) {
|
nni_list_remove(&l->aios, aio);
|
nni_aio_set_output(aio, 0, ws);
|
nni_aio_finish(aio, 0, 0);
|
} else {
|
nni_list_append(&l->pend, ws);
|
}
|
if (nni_list_empty(&l->reply)) {
|
nni_cv_wake(&l->cv);
|
}
|
nni_mtx_unlock(&l->mtx);
|
}
|
|
static void
|
ws_http_cb_dialer(nni_ws *ws, nni_aio *aio)
|
{
|
nni_ws_dialer *d;
|
nni_aio * uaio;
|
int rv;
|
uint16_t status;
|
char wskey[29];
|
const char * ptr;
|
|
d = ws->dialer;
|
nni_mtx_lock(&d->mtx);
|
uaio = ws->useraio;
|
|
// We have two steps. In step 1, we just sent the request,
|
// and need to retrieve the reply. In step two we have
|
// received the reply, and need to validate it.
|
// Note that its possible that the user canceled the request,
|
// in which case we no longer care, and just go to the error
|
// case to discard the ws.
|
if (((rv = nni_aio_result(aio)) != 0) || (uaio == NULL)) {
|
goto err;
|
}
|
|
// If we have no response structure, then this was completion
|
// of the send of the request. Prepare an empty response, and
|
// read it.
|
if (ws->res == NULL) {
|
if ((rv = nni_http_res_alloc(&ws->res)) != 0) {
|
goto err;
|
}
|
nni_http_read_res(ws->http, ws->res, ws->httpaio);
|
nni_mtx_unlock(&d->mtx);
|
return;
|
}
|
|
status = nni_http_res_get_status(ws->res);
|
switch (status) {
|
case NNG_HTTP_STATUS_SWITCHING:
|
break;
|
case NNG_HTTP_STATUS_FORBIDDEN:
|
case NNG_HTTP_STATUS_UNAUTHORIZED:
|
rv = NNG_EPERM;
|
goto err;
|
case NNG_HTTP_STATUS_NOT_FOUND:
|
case NNG_HTTP_STATUS_METHOD_NOT_ALLOWED:
|
rv = NNG_ECONNREFUSED; // Treat these as refusals.
|
goto err;
|
case NNG_HTTP_STATUS_BAD_REQUEST:
|
default:
|
// Perhaps we should use NNG_ETRANERR...
|
rv = NNG_EPROTO;
|
goto err;
|
}
|
|
// Check that the server gave us back the right key.
|
rv = ws_make_accept(
|
nni_http_req_get_header(ws->req, "Sec-WebSocket-Key"), wskey);
|
if (rv != 0) {
|
goto err;
|
}
|
|
#define GETH(h) nni_http_res_get_header(ws->res, h)
|
|
if (((ptr = GETH("Sec-WebSocket-Accept")) == NULL) ||
|
(strcmp(ptr, wskey) != 0) ||
|
((ptr = GETH("Connection")) == NULL) ||
|
(!ws_contains_word(ptr, "upgrade")) ||
|
((ptr = GETH("Upgrade")) == NULL) ||
|
(strcmp(ptr, "websocket") != 0)) {
|
ws_close_error(ws, WS_CLOSE_PROTOCOL_ERR);
|
rv = NNG_EPROTO;
|
goto err;
|
}
|
if (d->proto != NULL) {
|
if (((ptr = GETH("Sec-WebSocket-Protocol")) == NULL) ||
|
(!ws_contains_word(d->proto, ptr))) {
|
ws_close_error(ws, WS_CLOSE_PROTOCOL_ERR);
|
rv = NNG_EPROTO;
|
goto err;
|
}
|
}
|
#undef GETH
|
|
// At this point, we are in business!
|
nni_list_remove(&d->wspend, ws);
|
ws->ready = true;
|
ws->useraio = NULL;
|
ws->dialer = NULL;
|
nni_aio_set_output(uaio, 0, ws);
|
nni_aio_finish(uaio, 0, 0);
|
if (nni_list_empty(&d->wspend)) {
|
nni_cv_wake(&d->cv);
|
}
|
nni_mtx_unlock(&d->mtx);
|
return;
|
err:
|
nni_list_remove(&d->wspend, ws);
|
ws->useraio = NULL;
|
ws->dialer = NULL;
|
if (nni_list_empty(&d->wspend)) {
|
nni_cv_wake(&d->cv);
|
}
|
if (uaio != NULL) {
|
nni_aio_finish_error(uaio, rv);
|
}
|
nni_mtx_unlock(&d->mtx);
|
|
ws_reap(ws);
|
}
|
|
static void
|
ws_http_cb(void *arg)
|
{
|
nni_ws * ws = arg;
|
nni_aio *aio = ws->httpaio;
|
|
if (ws->server) {
|
ws_http_cb_listener(ws, aio);
|
} else {
|
ws_http_cb_dialer(ws, aio);
|
}
|
}
|
|
static int
|
ws_init(nni_ws **wsp)
|
{
|
nni_ws *ws;
|
int rv;
|
|
if ((ws = NNI_ALLOC_STRUCT(ws)) == NULL) {
|
return (NNG_ENOMEM);
|
}
|
nni_mtx_init(&ws->mtx);
|
NNI_LIST_INIT(&ws->rxq, ws_frame, node);
|
NNI_LIST_INIT(&ws->txq, ws_frame, node);
|
nni_aio_list_init(&ws->sendq);
|
nni_aio_list_init(&ws->recvq);
|
|
if (((rv = nni_aio_alloc(&ws->closeaio, ws_close_cb, ws)) != 0) ||
|
((rv = nni_aio_alloc(&ws->txaio, ws_write_cb, ws)) != 0) ||
|
((rv = nni_aio_alloc(&ws->rxaio, ws_read_cb, ws)) != 0) ||
|
((rv = nni_aio_alloc(&ws->httpaio, ws_http_cb, ws)) != 0) ||
|
((rv = nni_aio_alloc(&ws->connaio, ws_conn_cb, ws)) != 0)) {
|
ws_fini(ws);
|
return (rv);
|
}
|
|
nni_aio_set_timeout(ws->closeaio, 100);
|
nni_aio_set_timeout(ws->httpaio, 2000);
|
|
ws->ops.s_close = ws_str_close;
|
ws->ops.s_free = ws_str_free;
|
ws->ops.s_send = ws_str_send;
|
ws->ops.s_recv = ws_str_recv;
|
ws->ops.s_get = ws_str_get;
|
ws->ops.s_set = ws_str_set;
|
|
ws->fragsize = 1 << 20; // we won't send a frame larger than this
|
*wsp = ws;
|
return (0);
|
}
|
|
static void
|
ws_listener_free(void *arg)
|
{
|
nni_ws_listener *l = arg;
|
ws_header * hdr;
|
|
ws_listener_close(l);
|
|
nni_mtx_lock(&l->mtx);
|
while (!nni_list_empty(&l->reply)) {
|
nni_cv_wait(&l->cv);
|
}
|
nni_mtx_unlock(&l->mtx);
|
|
if (l->handler != NULL) {
|
nni_http_handler_fini(l->handler);
|
l->handler = NULL;
|
}
|
if (l->server != NULL) {
|
nni_http_server_fini(l->server);
|
l->server = NULL;
|
}
|
|
nni_cv_fini(&l->cv);
|
nni_mtx_fini(&l->mtx);
|
nni_strfree(l->proto);
|
while ((hdr = nni_list_first(&l->headers)) != NULL) {
|
nni_list_remove(&l->headers, hdr);
|
nni_strfree(hdr->name);
|
nni_strfree(hdr->value);
|
NNI_FREE_STRUCT(hdr);
|
}
|
if (l->url) {
|
nng_url_free(l->url);
|
}
|
NNI_FREE_STRUCT(l);
|
}
|
|
static void
|
ws_handler(nni_aio *aio)
|
{
|
nni_ws_listener * l;
|
nni_ws * ws;
|
nni_http_conn * conn;
|
nni_http_req * req;
|
nni_http_res * res;
|
nni_http_handler *h;
|
const char * ptr;
|
const char * proto;
|
uint16_t status;
|
int rv;
|
char key[29];
|
ws_header * hdr;
|
|
req = nni_aio_get_input(aio, 0);
|
h = nni_aio_get_input(aio, 1);
|
conn = nni_aio_get_input(aio, 2);
|
l = nni_http_handler_get_data(h);
|
|
nni_mtx_lock(&l->mtx);
|
if (l->closed) {
|
status = NNG_HTTP_STATUS_SERVICE_UNAVAILABLE;
|
goto err;
|
}
|
|
// Now check the headers, etc.
|
if (strcmp(nni_http_req_get_version(req), "HTTP/1.1") != 0) {
|
status = NNG_HTTP_STATUS_HTTP_VERSION_NOT_SUPP;
|
goto err;
|
}
|
|
if (strcmp(nni_http_req_get_method(req), "GET") != 0) {
|
// HEAD request. We can't really deal with it.
|
status = NNG_HTTP_STATUS_BAD_REQUEST;
|
goto err;
|
}
|
|
#define GETH(h) nni_http_req_get_header(req, h)
|
#define SETH(h, v) nni_http_res_set_header(res, h, v)
|
|
if ((((ptr = GETH("Content-Length")) != NULL) && (atoi(ptr) > 0)) ||
|
(((ptr = GETH("Transfer-Encoding")) != NULL) &&
|
(nni_strcasestr(ptr, "chunked") != NULL))) {
|
status = NNG_HTTP_STATUS_PAYLOAD_TOO_LARGE;
|
goto err;
|
}
|
|
// These headers have to be present.
|
if (((ptr = GETH("Upgrade")) == NULL) ||
|
(!ws_contains_word(ptr, "websocket")) ||
|
((ptr = GETH("Connection")) == NULL) ||
|
(!ws_contains_word(ptr, "upgrade")) ||
|
((ptr = GETH("Sec-WebSocket-Version")) == NULL) ||
|
(strcmp(ptr, "13") != 0)) {
|
status = NNG_HTTP_STATUS_BAD_REQUEST;
|
goto err;
|
}
|
|
if (((ptr = GETH("Sec-WebSocket-Key")) == NULL) ||
|
(ws_make_accept(ptr, key) != 0)) {
|
status = NNG_HTTP_STATUS_BAD_REQUEST;
|
goto err;
|
}
|
|
// If the client has requested a specific subprotocol, then
|
// we need to try to match it to what the handler says we
|
// support. (If no suitable option is found in the handler, we
|
// fail the request.)
|
proto = GETH("Sec-WebSocket-Protocol");
|
if (proto == NULL) {
|
if (l->proto != NULL) {
|
status = NNG_HTTP_STATUS_BAD_REQUEST;
|
goto err;
|
}
|
} else if ((l->proto == NULL) ||
|
(!ws_contains_word(l->proto, proto))) {
|
status = NNG_HTTP_STATUS_BAD_REQUEST;
|
goto err;
|
}
|
|
if ((rv = nni_http_res_alloc(&res)) != 0) {
|
// Give a chance to reply to client.
|
status = NNG_HTTP_STATUS_INTERNAL_SERVER_ERROR;
|
goto err;
|
}
|
|
if (nni_http_res_set_status(res, NNG_HTTP_STATUS_SWITCHING) != 0) {
|
nni_http_res_free(res);
|
status = NNG_HTTP_STATUS_INTERNAL_SERVER_ERROR;
|
goto err;
|
}
|
|
if ((SETH("Connection", "Upgrade") != 0) ||
|
(SETH("Upgrade", "websocket") != 0) ||
|
(SETH("Sec-WebSocket-Accept", key) != 0)) {
|
status = NNG_HTTP_STATUS_INTERNAL_SERVER_ERROR;
|
nni_http_res_free(res);
|
goto err;
|
}
|
if ((proto != NULL) && (SETH("Sec-WebSocket-Protocol", proto) != 0)) {
|
status = NNG_HTTP_STATUS_INTERNAL_SERVER_ERROR;
|
nni_http_res_free(res);
|
goto err;
|
}
|
|
// Set any user supplied headers. This is better than using a hook
|
// for most things, because it is loads easier.
|
NNI_LIST_FOREACH (&l->headers, hdr) {
|
if (SETH(hdr->name, hdr->value) != 0) {
|
status = NNG_HTTP_STATUS_INTERNAL_SERVER_ERROR;
|
nni_http_res_free(res);
|
goto err;
|
}
|
}
|
|
// The hook function gives us the ability to intercept the HTTP
|
// response altogether. Its best not to do this unless you really
|
// need to, because it's much more complex. But if you want to set
|
// up an HTTP Authorization handler this might be the only choice.
|
if (l->hookfn != NULL) {
|
rv = l->hookfn(l->hookarg, req, res);
|
if (rv != 0) {
|
nni_http_res_free(res);
|
nni_aio_finish_error(aio, rv);
|
nni_mtx_unlock(&l->mtx);
|
return;
|
}
|
|
if (nni_http_res_get_status(res) !=
|
NNG_HTTP_STATUS_SWITCHING) {
|
// The hook has decided to give back a
|
// different reply and we are not upgrading
|
// anymore. For example the Origin might not
|
// be permitted, or another level of
|
// authentication may be required. (Note that
|
// the hook can also give back various other
|
// headers, but it would be bad for it to alter
|
// the websocket mandated headers.)
|
nni_http_req_free(req);
|
nni_aio_set_output(aio, 0, res);
|
nni_aio_finish(aio, 0, 0);
|
nni_mtx_unlock(&l->mtx);
|
return;
|
}
|
}
|
|
#undef GETH
|
#undef SETH
|
|
// We are good to go, provided we can get the websocket struct,
|
// and send the reply.
|
if ((rv = ws_init(&ws)) != 0) {
|
nni_http_req_free(req);
|
nni_http_res_free(res);
|
status = NNG_HTTP_STATUS_INTERNAL_SERVER_ERROR;
|
goto err;
|
}
|
ws->http = conn;
|
ws->req = req;
|
ws->res = res;
|
ws->server = true;
|
ws->maxframe = l->maxframe;
|
ws->fragsize = l->fragsize;
|
ws->recvmax = l->recvmax;
|
ws->isstream = l->isstream;
|
ws->recv_text = l->recv_text;
|
ws->send_text = l->send_text;
|
ws->listener = l;
|
|
nni_list_append(&l->reply, ws);
|
nni_http_write_res(conn, res, ws->httpaio);
|
(void) nni_http_hijack(conn);
|
nni_aio_set_output(aio, 0, NULL);
|
nni_aio_finish(aio, 0, 0);
|
nni_mtx_unlock(&l->mtx);
|
return;
|
|
err:
|
if ((rv = nni_http_res_alloc_error(&res, status)) != 0) {
|
nni_aio_finish_error(aio, rv);
|
} else {
|
nni_aio_set_output(aio, 0, res);
|
nni_aio_finish(aio, 0, 0);
|
}
|
nni_mtx_unlock(&l->mtx);
|
}
|
|
static void
|
ws_accept_cancel(nni_aio *aio, void *arg, int rv)
|
{
|
nni_ws_listener *l = arg;
|
|
nni_mtx_lock(&l->mtx);
|
if (nni_aio_list_active(aio)) {
|
nni_aio_list_remove(aio);
|
nni_aio_finish_error(aio, rv);
|
}
|
nni_mtx_unlock(&l->mtx);
|
}
|
|
static void
|
ws_listener_accept(void *arg, nni_aio *aio)
|
{
|
nni_ws_listener *l = arg;
|
nni_ws * ws;
|
int rv;
|
|
if (nni_aio_begin(aio) != 0) {
|
return;
|
}
|
nni_mtx_lock(&l->mtx);
|
if (l->closed) {
|
nni_aio_finish_error(aio, NNG_ECLOSED);
|
nni_mtx_unlock(&l->mtx);
|
return;
|
}
|
if (!l->started) {
|
nni_aio_finish_error(aio, NNG_ESTATE);
|
nni_mtx_unlock(&l->mtx);
|
return;
|
}
|
if ((ws = nni_list_first(&l->pend)) != NULL) {
|
nni_list_remove(&l->pend, ws);
|
nni_mtx_unlock(&l->mtx);
|
nni_aio_set_output(aio, 0, ws);
|
nni_aio_finish(aio, 0, 0);
|
return;
|
}
|
if ((rv = nni_aio_schedule(aio, ws_accept_cancel, l)) != 0) {
|
nni_aio_finish_error(aio, rv);
|
nni_mtx_unlock(&l->mtx);
|
return;
|
}
|
nni_list_append(&l->aios, aio);
|
nni_mtx_unlock(&l->mtx);
|
}
|
|
static void
|
ws_listener_close(void *arg)
|
{
|
nni_ws_listener *l = arg;
|
nni_ws * ws;
|
nni_mtx_lock(&l->mtx);
|
if (l->closed) {
|
nni_mtx_unlock(&l->mtx);
|
return;
|
}
|
l->closed = true;
|
if (l->started) {
|
nni_http_server_del_handler(l->server, l->handler);
|
nni_http_server_stop(l->server);
|
l->started = false;
|
}
|
NNI_LIST_FOREACH (&l->pend, ws) {
|
ws_close_error(ws, WS_CLOSE_GOING_AWAY);
|
}
|
NNI_LIST_FOREACH (&l->reply, ws) {
|
ws_close_error(ws, WS_CLOSE_GOING_AWAY);
|
}
|
nni_mtx_unlock(&l->mtx);
|
}
|
|
// XXX: Consider replacing this with an option.
|
void
|
nni_ws_listener_hook(
|
nni_ws_listener *l, nni_ws_listen_hook hookfn, void *hookarg)
|
{
|
nni_mtx_lock(&l->mtx);
|
l->hookfn = hookfn;
|
l->hookarg = hookarg;
|
nni_mtx_unlock(&l->mtx);
|
}
|
|
static int
|
ws_listener_listen(void *arg)
|
{
|
nni_ws_listener *l = arg;
|
int rv;
|
|
nni_mtx_lock(&l->mtx);
|
if (l->closed) {
|
nni_mtx_unlock(&l->mtx);
|
return (NNG_ECLOSED);
|
}
|
if (l->started) {
|
nni_mtx_unlock(&l->mtx);
|
return (NNG_ESTATE);
|
}
|
|
if ((rv = nni_http_server_add_handler(l->server, l->handler)) != 0) {
|
nni_http_server_fini(l->server);
|
l->server = NULL;
|
nni_mtx_unlock(&l->mtx);
|
return (rv);
|
}
|
|
if ((rv = nni_http_server_start(l->server)) != 0) {
|
nni_http_server_del_handler(l->server, l->handler);
|
nni_http_server_fini(l->server);
|
l->server = NULL;
|
nni_mtx_unlock(&l->mtx);
|
return (rv);
|
}
|
|
l->started = true;
|
|
nni_mtx_unlock(&l->mtx);
|
return (0);
|
}
|
|
static int
|
ws_listener_set_size(
|
nni_ws_listener *l, size_t *valp, const void *buf, size_t sz, nni_type t)
|
{
|
size_t val;
|
int rv;
|
|
// Max size is limited to 4 GB, but you really never want to have
|
// to have a larger value. If you think you need that, you're doing it
|
// wrong. You *can* set the size to 0 for unlimited.
|
if ((rv = nni_copyin_size(&val, buf, sz, 0, NNI_MAXSZ, t)) == 0) {
|
nni_mtx_lock(&l->mtx);
|
*valp = val;
|
nni_mtx_unlock(&l->mtx);
|
}
|
return (rv);
|
}
|
|
static int
|
ws_listener_get_size(
|
nni_ws_listener *l, size_t *valp, void *buf, size_t *szp, nni_type t)
|
{
|
size_t val;
|
nni_mtx_lock(&l->mtx);
|
val = *valp;
|
nni_mtx_unlock(&l->mtx);
|
return (nni_copyout_size(val, buf, szp, t));
|
}
|
|
static int
|
ws_listener_set_maxframe(void *arg, const void *buf, size_t sz, nni_type t)
|
{
|
nni_ws_listener *l = arg;
|
return (ws_listener_set_size(l, &l->maxframe, buf, sz, t));
|
}
|
|
static int
|
ws_listener_get_maxframe(void *arg, void *buf, size_t *szp, nni_type t)
|
{
|
nni_ws_listener *l = arg;
|
return (ws_listener_get_size(l, &l->maxframe, buf, szp, t));
|
}
|
|
static int
|
ws_listener_set_fragsize(void *arg, const void *buf, size_t sz, nni_type t)
|
{
|
nni_ws_listener *l = arg;
|
return (ws_listener_set_size(l, &l->fragsize, buf, sz, t));
|
}
|
|
static int
|
ws_listener_get_fragsize(void *arg, void *buf, size_t *szp, nni_type t)
|
{
|
nni_ws_listener *l = arg;
|
return (ws_listener_get_size(l, &l->fragsize, buf, szp, t));
|
}
|
|
static int
|
ws_listener_set_recvmax(void *arg, const void *buf, size_t sz, nni_type t)
|
{
|
nni_ws_listener *l = arg;
|
return (ws_listener_set_size(l, &l->recvmax, buf, sz, t));
|
}
|
|
static int
|
ws_listener_get_recvmax(void *arg, void *buf, size_t *szp, nni_type t)
|
{
|
nni_ws_listener *l = arg;
|
return (ws_listener_get_size(l, &l->recvmax, buf, szp, t));
|
}
|
|
static int
|
ws_listener_set_res_headers(void *arg, const void *buf, size_t sz, nni_type t)
|
{
|
nni_ws_listener *l = arg;
|
int rv;
|
|
if ((rv = ws_check_string(buf, sz, t)) == 0) {
|
nni_mtx_lock(&l->mtx);
|
rv = ws_set_headers(&l->headers, buf);
|
nni_mtx_unlock(&l->mtx);
|
}
|
return (rv);
|
}
|
|
static int
|
ws_listener_set_proto(void *arg, const void *buf, size_t sz, nni_type t)
|
{
|
nni_ws_listener *l = arg;
|
int rv;
|
|
if ((rv = ws_check_string(buf, sz, t)) == 0) {
|
char *ns;
|
if ((ns = nni_strdup(buf)) == NULL) {
|
rv = NNG_ENOMEM;
|
} else {
|
nni_mtx_lock(&l->mtx);
|
if (l->proto != NULL) {
|
nni_strfree(l->proto);
|
}
|
l->proto = ns;
|
nni_mtx_unlock(&l->mtx);
|
}
|
}
|
return (rv);
|
}
|
|
static int
|
ws_listener_get_proto(void *arg, void *buf, size_t *szp, nni_type t)
|
{
|
nni_ws_listener *l = arg;
|
int rv;
|
nni_mtx_lock(&l->mtx);
|
rv = nni_copyout_str(l->proto != NULL ? l->proto : "", buf, szp, t);
|
nni_mtx_unlock(&l->mtx);
|
return (rv);
|
}
|
|
static int
|
ws_listener_set_msgmode(void *arg, const void *buf, size_t sz, nni_type t)
|
{
|
nni_ws_listener *l = arg;
|
int rv;
|
bool b;
|
|
if ((rv = nni_copyin_bool(&b, buf, sz, t)) == 0) {
|
nni_mtx_lock(&l->mtx);
|
l->isstream = !b;
|
nni_mtx_unlock(&l->mtx);
|
}
|
return (rv);
|
}
|
|
static int
|
ws_listener_set_recv_text(void *arg, const void *buf, size_t sz, nni_type t)
|
{
|
nni_ws_listener *l = arg;
|
int rv;
|
bool b;
|
|
if ((rv = nni_copyin_bool(&b, buf, sz, t)) == 0) {
|
nni_mtx_lock(&l->mtx);
|
l->recv_text = b;
|
nni_mtx_unlock(&l->mtx);
|
}
|
return (rv);
|
}
|
|
static int
|
ws_listener_set_send_text(void *arg, const void *buf, size_t sz, nni_type t)
|
{
|
nni_ws_listener *l = arg;
|
int rv;
|
bool b;
|
|
if ((rv = nni_copyin_bool(&b, buf, sz, t)) == 0) {
|
nni_mtx_lock(&l->mtx);
|
l->send_text = b;
|
nni_mtx_unlock(&l->mtx);
|
}
|
return (rv);
|
}
|
|
static int
|
ws_listener_get_recv_text(void *arg, void *buf, size_t *szp, nni_type t)
|
{
|
nni_ws_listener *l = arg;
|
int rv;
|
nni_mtx_lock(&l->mtx);
|
rv = nni_copyout_bool(l->recv_text, buf, szp, t);
|
nni_mtx_unlock(&l->mtx);
|
return (rv);
|
}
|
|
static int
|
ws_listener_get_send_text(void *arg, void *buf, size_t *szp, nni_type t)
|
{
|
nni_ws_listener *l = arg;
|
int rv;
|
nni_mtx_lock(&l->mtx);
|
rv = nni_copyout_bool(l->send_text, buf, szp, t);
|
nni_mtx_unlock(&l->mtx);
|
return (rv);
|
}
|
|
static int
|
ws_listener_get_url(void *arg, void *buf, size_t *szp, nni_type t)
|
{
|
nni_ws_listener *l = arg;
|
int rv;
|
|
nni_mtx_lock(&l->mtx);
|
rv = nni_copyout_str(l->url->u_rawurl, buf, szp, t);
|
nni_mtx_unlock(&l->mtx);
|
return (rv);
|
}
|
|
static const nni_option ws_listener_options[] = {
|
{
|
.o_name = NNI_OPT_WS_MSGMODE,
|
.o_set = ws_listener_set_msgmode,
|
},
|
{
|
.o_name = NNG_OPT_WS_RECVMAXFRAME,
|
.o_set = ws_listener_set_maxframe,
|
.o_get = ws_listener_get_maxframe,
|
},
|
{
|
.o_name = NNG_OPT_WS_SENDMAXFRAME,
|
.o_set = ws_listener_set_fragsize,
|
.o_get = ws_listener_get_fragsize,
|
},
|
{
|
.o_name = NNG_OPT_RECVMAXSZ,
|
.o_set = ws_listener_set_recvmax,
|
.o_get = ws_listener_get_recvmax,
|
},
|
{
|
.o_name = NNG_OPT_WS_RESPONSE_HEADERS,
|
.o_set = ws_listener_set_res_headers,
|
// XXX: Get not implemented yet; likely of marginal value.
|
},
|
{
|
.o_name = NNG_OPT_WS_PROTOCOL,
|
.o_set = ws_listener_set_proto,
|
.o_get = ws_listener_get_proto,
|
},
|
{
|
.o_name = NNG_OPT_URL,
|
.o_get = ws_listener_get_url,
|
},
|
{
|
.o_name = NNG_OPT_WS_RECV_TEXT,
|
.o_set = ws_listener_set_recv_text,
|
.o_get = ws_listener_get_recv_text,
|
},
|
{
|
.o_name = NNG_OPT_WS_SEND_TEXT,
|
.o_set = ws_listener_set_send_text,
|
.o_get = ws_listener_get_send_text,
|
},
|
{
|
.o_name = NULL,
|
},
|
};
|
|
static int
|
ws_listener_set_header(nni_ws_listener *l, const char *name, const void *buf,
|
size_t sz, nni_type t)
|
{
|
int rv;
|
name += strlen(NNG_OPT_WS_RESPONSE_HEADER);
|
if ((rv = ws_check_string(buf, sz, t)) == 0) {
|
nni_mtx_lock(&l->mtx);
|
rv = ws_set_header(&l->headers, name, buf);
|
nni_mtx_unlock(&l->mtx);
|
}
|
return (rv);
|
}
|
|
static int
|
ws_listener_set(
|
void *arg, const char *name, const void *buf, size_t sz, nni_type t)
|
{
|
nni_ws_listener *l = arg;
|
int rv;
|
|
rv = nni_setopt(ws_listener_options, name, l, buf, sz, t);
|
if (rv == NNG_ENOTSUP) {
|
rv = nni_http_server_set(l->server, name, buf, sz, t);
|
}
|
|
if (rv == NNG_ENOTSUP) {
|
if (startswith(name, NNG_OPT_WS_RESPONSE_HEADER)) {
|
rv = ws_listener_set_header(l, name, buf, sz, t);
|
}
|
}
|
return (rv);
|
}
|
|
static int
|
ws_listener_get(
|
void *arg, const char *name, void *buf, size_t *szp, nni_type t)
|
{
|
nni_ws_listener *l = arg;
|
int rv;
|
|
rv = nni_getopt(ws_listener_options, name, l, buf, szp, t);
|
if (rv == NNG_ENOTSUP) {
|
rv = nni_http_server_get(l->server, name, buf, szp, t);
|
}
|
return (rv);
|
}
|
|
int
|
nni_ws_listener_alloc(nng_stream_listener **wslp, const nng_url *url)
|
{
|
nni_ws_listener *l;
|
int rv;
|
char * host;
|
|
if ((l = NNI_ALLOC_STRUCT(l)) == NULL) {
|
return (NNG_ENOMEM);
|
}
|
nni_mtx_init(&l->mtx);
|
nni_cv_init(&l->cv, &l->mtx);
|
nni_aio_list_init(&l->aios);
|
|
NNI_LIST_INIT(&l->pend, nni_ws, node);
|
NNI_LIST_INIT(&l->reply, nni_ws, node);
|
|
// make a private copy of the url structure.
|
if ((rv = nng_url_clone(&l->url, url)) != 0) {
|
ws_listener_free(l);
|
return (rv);
|
}
|
|
host = l->url->u_hostname;
|
if (strlen(host) == 0) {
|
host = NULL;
|
}
|
rv = nni_http_handler_init(&l->handler, url->u_path, ws_handler);
|
if (rv != 0) {
|
ws_listener_free(l);
|
return (rv);
|
}
|
|
if (((rv = nni_http_handler_set_host(l->handler, host)) != 0) ||
|
((rv = nni_http_handler_set_data(l->handler, l, 0)) != 0) ||
|
((rv = nni_http_server_init(&l->server, url)) != 0)) {
|
ws_listener_free(l);
|
return (rv);
|
}
|
|
l->fragsize = WS_DEF_MAXTXFRAME;
|
l->maxframe = WS_DEF_MAXRXFRAME;
|
l->recvmax = WS_DEF_RECVMAX;
|
l->isstream = true;
|
l->ops.sl_free = ws_listener_free;
|
l->ops.sl_close = ws_listener_close;
|
l->ops.sl_accept = ws_listener_accept;
|
l->ops.sl_listen = ws_listener_listen;
|
l->ops.sl_set = ws_listener_set;
|
l->ops.sl_get = ws_listener_get;
|
*wslp = (void *) l;
|
return (0);
|
}
|
|
void
|
ws_conn_cb(void *arg)
|
{
|
nni_ws_dialer *d;
|
nni_ws * ws;
|
nni_aio * uaio;
|
nni_http_conn *http;
|
nni_http_req * req = NULL;
|
int rv;
|
uint8_t raw[16];
|
char wskey[25];
|
ws_header * hdr;
|
|
ws = arg;
|
|
d = ws->dialer;
|
if ((rv = nni_aio_result(ws->connaio)) != 0) {
|
nni_mtx_lock(&ws->mtx);
|
if ((uaio = ws->useraio) != NULL) {
|
ws->useraio = NULL;
|
nni_aio_finish_error(uaio, rv);
|
}
|
nni_mtx_unlock(&ws->mtx);
|
nni_mtx_lock(&d->mtx);
|
if (nni_list_node_active(&ws->node)) {
|
nni_list_remove(&d->wspend, ws);
|
ws->dialer = NULL;
|
if (nni_list_empty(&d->wspend)) {
|
nni_cv_wake(&d->cv);
|
}
|
nni_mtx_unlock(&d->mtx);
|
ws_reap(ws);
|
} else {
|
nni_mtx_unlock(&d->mtx);
|
}
|
return;
|
}
|
|
nni_mtx_lock(&ws->mtx);
|
uaio = ws->useraio;
|
http = nni_aio_get_output(ws->connaio, 0);
|
nni_aio_set_output(ws->connaio, 0, NULL);
|
if (uaio == NULL) {
|
// This request was canceled for some reason.
|
nni_http_conn_fini(http);
|
nni_mtx_unlock(&ws->mtx);
|
ws_reap(ws);
|
return;
|
}
|
|
for (int i = 0; i < 16; i++) {
|
raw[i] = (uint8_t) nni_random();
|
}
|
nni_base64_encode(raw, 16, wskey, 24);
|
wskey[24] = '\0';
|
|
#define SETH(h, v) nni_http_req_set_header(req, h, v)
|
if ((rv != 0) || ((rv = nni_http_req_alloc(&req, d->url)) != 0) ||
|
((rv = SETH("Upgrade", "websocket")) != 0) ||
|
((rv = SETH("Connection", "Upgrade")) != 0) ||
|
((rv = SETH("Sec-WebSocket-Key", wskey)) != 0) ||
|
((rv = SETH("Sec-WebSocket-Version", "13")) != 0)) {
|
goto err;
|
}
|
|
if ((d->proto != NULL) &&
|
((rv = SETH("Sec-WebSocket-Protocol", d->proto)) != 0)) {
|
goto err;
|
}
|
|
NNI_LIST_FOREACH (&d->headers, hdr) {
|
if ((rv = SETH(hdr->name, hdr->value)) != 0) {
|
goto err;
|
}
|
}
|
#undef SETH
|
|
ws->http = http;
|
ws->req = req;
|
|
nni_http_write_req(http, req, ws->httpaio);
|
nni_mtx_unlock(&ws->mtx);
|
return;
|
|
err:
|
nni_aio_finish_error(uaio, rv);
|
nni_mtx_unlock(&ws->mtx);
|
if (http != NULL) {
|
nni_http_conn_fini(http);
|
}
|
if (req != NULL) {
|
nni_http_req_free(req);
|
}
|
ws_reap(ws);
|
}
|
|
static void
|
ws_dialer_free(void *arg)
|
{
|
nni_ws_dialer *d = arg;
|
ws_header * hdr;
|
|
nni_mtx_lock(&d->mtx);
|
while (!nni_list_empty(&d->wspend)) {
|
nni_cv_wait(&d->cv);
|
}
|
nni_mtx_unlock(&d->mtx);
|
|
nni_strfree(d->proto);
|
while ((hdr = nni_list_first(&d->headers)) != NULL) {
|
nni_list_remove(&d->headers, hdr);
|
nni_strfree(hdr->name);
|
nni_strfree(hdr->value);
|
NNI_FREE_STRUCT(hdr);
|
}
|
if (d->client) {
|
nni_http_client_fini(d->client);
|
}
|
if (d->url) {
|
nng_url_free(d->url);
|
}
|
nni_cv_fini(&d->cv);
|
nni_mtx_fini(&d->mtx);
|
NNI_FREE_STRUCT(d);
|
}
|
|
static void
|
ws_dialer_close(void *arg)
|
{
|
nni_ws_dialer *d = arg;
|
nni_ws * ws;
|
nni_mtx_lock(&d->mtx);
|
if (d->closed) {
|
nni_mtx_unlock(&d->mtx);
|
return;
|
}
|
d->closed = true;
|
NNI_LIST_FOREACH (&d->wspend, ws) {
|
nni_aio_close(ws->connaio);
|
nni_aio_close(ws->httpaio);
|
}
|
nni_mtx_unlock(&d->mtx);
|
}
|
|
static void
|
ws_dial_cancel(nni_aio *aio, void *arg, int rv)
|
{
|
nni_ws *ws = arg;
|
|
nni_mtx_lock(&ws->mtx);
|
if (aio == ws->useraio) {
|
nni_aio_abort(ws->connaio, rv);
|
nni_aio_abort(ws->httpaio, rv);
|
ws->useraio = NULL;
|
nni_aio_finish_error(aio, rv);
|
}
|
nni_mtx_unlock(&ws->mtx);
|
}
|
|
static void
|
ws_dialer_dial(void *arg, nni_aio *aio)
|
{
|
nni_ws_dialer *d = arg;
|
nni_ws * ws;
|
int rv;
|
|
if (nni_aio_begin(aio) != 0) {
|
return;
|
}
|
if ((rv = ws_init(&ws)) != 0) {
|
nni_aio_finish_error(aio, rv);
|
return;
|
}
|
nni_mtx_lock(&d->mtx);
|
if (d->closed) {
|
nni_mtx_unlock(&d->mtx);
|
nni_aio_finish_error(aio, NNG_ECLOSED);
|
ws_reap(ws);
|
return;
|
}
|
if ((rv = nni_aio_schedule(aio, ws_dial_cancel, ws)) != 0) {
|
nni_mtx_unlock(&d->mtx);
|
nni_aio_finish_error(aio, rv);
|
ws_reap(ws);
|
return;
|
}
|
ws->dialer = d;
|
ws->useraio = aio;
|
ws->server = false;
|
ws->maxframe = d->maxframe;
|
ws->isstream = d->isstream;
|
ws->recv_text = d->recv_text;
|
ws->send_text = d->send_text;
|
nni_list_append(&d->wspend, ws);
|
nni_http_client_connect(d->client, ws->connaio);
|
nni_mtx_unlock(&d->mtx);
|
}
|
|
static int
|
ws_dialer_set_msgmode(void *arg, const void *buf, size_t sz, nni_type t)
|
{
|
nni_ws_dialer *d = arg;
|
int rv;
|
bool b;
|
|
if ((rv = nni_copyin_bool(&b, buf, sz, t)) == 0) {
|
nni_mtx_lock(&d->mtx);
|
d->isstream = !b;
|
nni_mtx_unlock(&d->mtx);
|
}
|
return (rv);
|
}
|
|
static int
|
ws_dialer_set_recv_text(void *arg, const void *buf, size_t sz, nni_type t)
|
{
|
nni_ws_dialer *d = arg;
|
int rv;
|
bool b;
|
|
if ((rv = nni_copyin_bool(&b, buf, sz, t)) == 0) {
|
nni_mtx_lock(&d->mtx);
|
d->recv_text = b;
|
nni_mtx_unlock(&d->mtx);
|
}
|
return (rv);
|
}
|
|
static int
|
ws_dialer_set_send_text(void *arg, const void *buf, size_t sz, nni_type t)
|
{
|
nni_ws_dialer *d = arg;
|
int rv;
|
bool b;
|
|
if ((rv = nni_copyin_bool(&b, buf, sz, t)) == 0) {
|
nni_mtx_lock(&d->mtx);
|
d->send_text = b;
|
nni_mtx_unlock(&d->mtx);
|
}
|
return (rv);
|
}
|
|
static int
|
ws_dialer_set_size(
|
nni_ws_dialer *d, size_t *valp, const void *buf, size_t sz, nni_type t)
|
{
|
size_t val;
|
int rv;
|
|
// Max size is limited to 4 GB, but you really never want to have
|
// to have a larger value. If you think you need that, you're doing it
|
// wrong. You *can* set the size to 0 for unlimited.
|
if ((rv = nni_copyin_size(&val, buf, sz, 0, NNI_MAXSZ, t)) == 0) {
|
nni_mtx_lock(&d->mtx);
|
*valp = val;
|
nni_mtx_unlock(&d->mtx);
|
}
|
return (rv);
|
}
|
|
static int
|
ws_dialer_get_size(
|
nni_ws_dialer *d, size_t *valp, void *buf, size_t *szp, nni_type t)
|
{
|
size_t val;
|
nni_mtx_lock(&d->mtx);
|
val = *valp;
|
nni_mtx_unlock(&d->mtx);
|
return (nni_copyout_size(val, buf, szp, t));
|
}
|
|
static int
|
ws_dialer_set_maxframe(void *arg, const void *buf, size_t sz, nni_type t)
|
{
|
nni_ws_dialer *d = arg;
|
return (ws_dialer_set_size(d, &d->maxframe, buf, sz, t));
|
}
|
|
static int
|
ws_dialer_get_maxframe(void *arg, void *buf, size_t *szp, nni_type t)
|
{
|
nni_ws_dialer *d = arg;
|
return (ws_dialer_get_size(d, &d->maxframe, buf, szp, t));
|
}
|
|
static int
|
ws_dialer_set_fragsize(void *arg, const void *buf, size_t sz, nni_type t)
|
{
|
nni_ws_dialer *d = arg;
|
return (ws_dialer_set_size(d, &d->fragsize, buf, sz, t));
|
}
|
|
static int
|
ws_dialer_get_fragsize(void *arg, void *buf, size_t *szp, nni_type t)
|
{
|
nni_ws_dialer *d = arg;
|
return (ws_dialer_get_size(d, &d->fragsize, buf, szp, t));
|
}
|
|
static int
|
ws_dialer_set_recvmax(void *arg, const void *buf, size_t sz, nni_type t)
|
{
|
nni_ws_dialer *d = arg;
|
return (ws_dialer_set_size(d, &d->recvmax, buf, sz, t));
|
}
|
|
static int
|
ws_dialer_get_recvmax(void *arg, void *buf, size_t *szp, nni_type t)
|
{
|
nni_ws_dialer *d = arg;
|
return (ws_dialer_get_size(d, &d->recvmax, buf, szp, t));
|
}
|
|
static int
|
ws_dialer_set_req_headers(void *arg, const void *buf, size_t sz, nni_type t)
|
{
|
nni_ws_dialer *d = arg;
|
int rv;
|
|
if ((rv = ws_check_string(buf, sz, t)) == 0) {
|
nni_mtx_lock(&d->mtx);
|
rv = ws_set_headers(&d->headers, buf);
|
nni_mtx_unlock(&d->mtx);
|
}
|
return (rv);
|
}
|
|
static int
|
ws_dialer_set_proto(void *arg, const void *buf, size_t sz, nni_type t)
|
{
|
nni_ws_dialer *d = arg;
|
int rv;
|
|
if ((rv = ws_check_string(buf, sz, t)) == 0) {
|
char *ns;
|
if ((ns = nni_strdup(buf)) == NULL) {
|
rv = NNG_ENOMEM;
|
} else {
|
nni_mtx_lock(&d->mtx);
|
if (d->proto != NULL) {
|
nni_strfree(d->proto);
|
}
|
d->proto = ns;
|
nni_mtx_unlock(&d->mtx);
|
}
|
}
|
return (rv);
|
}
|
|
static int
|
ws_dialer_get_proto(void *arg, void *buf, size_t *szp, nni_type t)
|
{
|
nni_ws_dialer *d = arg;
|
int rv;
|
nni_mtx_lock(&d->mtx);
|
rv = nni_copyout_str(d->proto != NULL ? d->proto : "", buf, szp, t);
|
nni_mtx_unlock(&d->mtx);
|
return (rv);
|
}
|
|
static int
|
ws_dialer_get_recv_text(void *arg, void *buf, size_t *szp, nni_type t)
|
{
|
nni_ws_dialer *d = arg;
|
int rv;
|
nni_mtx_lock(&d->mtx);
|
rv = nni_copyout_bool(d->recv_text, buf, szp, t);
|
nni_mtx_unlock(&d->mtx);
|
return (rv);
|
}
|
|
static int
|
ws_dialer_get_send_text(void *arg, void *buf, size_t *szp, nni_type t)
|
{
|
nni_ws_dialer *d = arg;
|
int rv;
|
nni_mtx_lock(&d->mtx);
|
rv = nni_copyout_bool(d->send_text, buf, szp, t);
|
nni_mtx_unlock(&d->mtx);
|
return (rv);
|
}
|
|
static const nni_option ws_dialer_options[] = {
|
{
|
.o_name = NNI_OPT_WS_MSGMODE,
|
.o_set = ws_dialer_set_msgmode,
|
},
|
{
|
.o_name = NNG_OPT_WS_RECVMAXFRAME,
|
.o_set = ws_dialer_set_maxframe,
|
.o_get = ws_dialer_get_maxframe,
|
},
|
{
|
.o_name = NNG_OPT_WS_SENDMAXFRAME,
|
.o_set = ws_dialer_set_fragsize,
|
.o_get = ws_dialer_get_fragsize,
|
},
|
{
|
.o_name = NNG_OPT_RECVMAXSZ,
|
.o_set = ws_dialer_set_recvmax,
|
.o_get = ws_dialer_get_recvmax,
|
},
|
{
|
.o_name = NNG_OPT_WS_REQUEST_HEADERS,
|
.o_set = ws_dialer_set_req_headers,
|
// XXX: Get not implemented yet; likely of marginal value.
|
},
|
{
|
.o_name = NNG_OPT_WS_PROTOCOL,
|
.o_set = ws_dialer_set_proto,
|
.o_get = ws_dialer_get_proto,
|
},
|
{
|
.o_name = NNG_OPT_WS_RECV_TEXT,
|
.o_set = ws_dialer_set_recv_text,
|
.o_get = ws_dialer_get_recv_text,
|
},
|
{
|
.o_name = NNG_OPT_WS_SEND_TEXT,
|
.o_set = ws_dialer_set_send_text,
|
.o_get = ws_dialer_get_send_text,
|
},
|
|
{
|
.o_name = NULL,
|
},
|
};
|
|
static int
|
ws_dialer_set_header(
|
nni_ws_dialer *d, const char *name, const void *buf, size_t sz, nni_type t)
|
{
|
int rv;
|
name += strlen(NNG_OPT_WS_REQUEST_HEADER);
|
if ((rv = ws_check_string(buf, sz, t)) == 0) {
|
nni_mtx_lock(&d->mtx);
|
rv = ws_set_header(&d->headers, name, buf);
|
nni_mtx_unlock(&d->mtx);
|
}
|
return (rv);
|
}
|
|
static int
|
ws_dialer_set(
|
void *arg, const char *name, const void *buf, size_t sz, nni_type t)
|
{
|
nni_ws_dialer *d = arg;
|
int rv;
|
|
rv = nni_setopt(ws_dialer_options, name, d, buf, sz, t);
|
if (rv == NNG_ENOTSUP) {
|
rv = nni_http_client_set(d->client, name, buf, sz, t);
|
}
|
|
if (rv == NNG_ENOTSUP) {
|
if (startswith(name, NNG_OPT_WS_REQUEST_HEADER)) {
|
rv = ws_dialer_set_header(d, name, buf, sz, t);
|
}
|
}
|
return (rv);
|
}
|
|
static int
|
ws_dialer_get(void *arg, const char *name, void *buf, size_t *szp, nni_type t)
|
{
|
nni_ws_dialer *d = arg;
|
int rv;
|
|
rv = nni_getopt(ws_dialer_options, name, d, buf, szp, t);
|
if (rv == NNG_ENOTSUP) {
|
rv = nni_http_client_get(d->client, name, buf, szp, t);
|
}
|
return (rv);
|
}
|
|
int
|
nni_ws_dialer_alloc(nng_stream_dialer **dp, const nng_url *url)
|
{
|
nni_ws_dialer *d;
|
int rv;
|
|
if ((d = NNI_ALLOC_STRUCT(d)) == NULL) {
|
return (NNG_ENOMEM);
|
}
|
NNI_LIST_INIT(&d->headers, ws_header, node);
|
NNI_LIST_INIT(&d->wspend, nni_ws, node);
|
nni_mtx_init(&d->mtx);
|
nni_cv_init(&d->cv, &d->mtx);
|
|
if ((rv = nng_url_clone(&d->url, url)) != 0) {
|
ws_dialer_free(d);
|
return (rv);
|
}
|
|
if ((rv = nni_http_client_init(&d->client, url)) != 0) {
|
ws_dialer_free(d);
|
return (rv);
|
}
|
d->isstream = true;
|
d->recvmax = WS_DEF_RECVMAX;
|
d->maxframe = WS_DEF_MAXRXFRAME;
|
d->fragsize = WS_DEF_MAXTXFRAME;
|
|
d->ops.sd_free = ws_dialer_free;
|
d->ops.sd_close = ws_dialer_close;
|
d->ops.sd_dial = ws_dialer_dial;
|
d->ops.sd_set = ws_dialer_set;
|
d->ops.sd_get = ws_dialer_get;
|
*dp = (void *) d;
|
return (0);
|
}
|
|
// Dialer does not get a hook chance, as it can examine the request
|
// and reply after dial is done; this is not a 3-way handshake, so
|
// the dialer does not confirm the server's response at the HTTP
|
// level. (It can still issue a websocket close).
|
|
static void
|
ws_str_free(void *arg)
|
{
|
nni_ws *ws = arg;
|
ws_reap(ws);
|
}
|
|
static void
|
ws_str_close(void *arg)
|
{
|
nni_ws *ws = arg;
|
ws_close_error(ws, WS_CLOSE_NORMAL_CLOSE);
|
}
|
|
static void
|
ws_str_send(void *arg, nni_aio *aio)
|
{
|
nni_ws * ws = arg;
|
int rv;
|
ws_frame *frame;
|
|
if (nni_aio_begin(aio) != 0) {
|
return;
|
}
|
|
if (!ws->isstream) {
|
nni_msg *msg;
|
unsigned niov;
|
nni_iov iov[2];
|
if ((msg = nni_aio_get_msg(aio)) == NULL) {
|
nni_aio_finish_error(aio, NNG_EINVAL);
|
return;
|
}
|
niov = 0;
|
if (nng_msg_header_len(msg) > 0) {
|
iov[niov].iov_len = nni_msg_header_len(msg);
|
iov[niov].iov_buf = nni_msg_header(msg);
|
niov++;
|
}
|
iov[niov].iov_len = nni_msg_len(msg);
|
iov[niov].iov_buf = nni_msg_body(msg);
|
niov++;
|
|
// Scribble into the iov for now.
|
nni_aio_set_iov(aio, niov, iov);
|
}
|
|
if ((frame = NNI_ALLOC_STRUCT(frame)) == NULL) {
|
nni_aio_finish_error(aio, NNG_ENOMEM);
|
return;
|
}
|
frame->aio = aio;
|
if ((rv = ws_frame_prep_tx(ws, frame)) != 0) {
|
nni_aio_finish_error(aio, rv);
|
ws_frame_fini(frame);
|
return;
|
}
|
|
nni_mtx_lock(&ws->mtx);
|
|
if (ws->closed) {
|
nni_mtx_unlock(&ws->mtx);
|
nni_aio_finish_error(aio, NNG_ECLOSED);
|
ws_frame_fini(frame);
|
return;
|
}
|
if ((rv = nni_aio_schedule(aio, ws_write_cancel, ws)) != 0) {
|
nni_mtx_unlock(&ws->mtx);
|
nni_aio_finish_error(aio, rv);
|
ws_frame_fini(frame);
|
return;
|
}
|
nni_aio_set_prov_extra(aio, 0, frame);
|
nni_list_append(&ws->sendq, aio);
|
nni_list_append(&ws->txq, frame);
|
ws_start_write(ws);
|
nni_mtx_unlock(&ws->mtx);
|
}
|
|
static void
|
ws_str_recv(void *arg, nng_aio *aio)
|
{
|
nni_ws *ws = arg;
|
int rv;
|
|
if (nni_aio_begin(aio) != 0) {
|
return;
|
}
|
nni_mtx_lock(&ws->mtx);
|
if ((rv = nni_aio_schedule(aio, ws_read_cancel, ws)) != 0) {
|
nni_mtx_unlock(&ws->mtx);
|
nni_aio_finish_error(aio, rv);
|
return;
|
}
|
nni_list_append(&ws->recvq, aio);
|
if (nni_list_first(&ws->recvq) == aio) {
|
ws_read_finish(ws);
|
}
|
ws_start_read(ws);
|
|
nni_mtx_unlock(&ws->mtx);
|
}
|
|
static int
|
ws_get_request_headers(void *arg, void *buf, size_t *szp, nni_type t)
|
{
|
nni_ws *ws = arg;
|
nni_mtx_lock(&ws->mtx);
|
if (ws->reqhdrs == NULL) {
|
ws->reqhdrs = nni_http_req_headers(ws->req);
|
}
|
nni_mtx_unlock(&ws->mtx);
|
return (nni_copyout_str(ws->reqhdrs, buf, szp, t));
|
}
|
|
static int
|
ws_get_response_headers(void *arg, void *buf, size_t *szp, nni_type t)
|
{
|
nni_ws *ws = arg;
|
nni_mtx_lock(&ws->mtx);
|
if (ws->reshdrs == NULL) {
|
ws->reshdrs = nni_http_res_headers(ws->res);
|
}
|
nni_mtx_unlock(&ws->mtx);
|
return (nni_copyout_str(ws->reshdrs, buf, szp, t));
|
}
|
|
static int
|
ws_get_request_uri(void *arg, void *buf, size_t *szp, nni_type t)
|
{
|
nni_ws *ws = arg;
|
return (nni_copyout_str(nni_http_req_get_uri(ws->req), buf, szp, t));
|
}
|
|
static int
|
ws_get_recv_text(void *arg, void *buf, size_t *szp, nni_type t)
|
{
|
nni_ws *ws = arg;
|
bool b;
|
nni_mtx_lock(&ws->mtx);
|
b = ws->recv_text;
|
nni_mtx_unlock(&ws->mtx);
|
|
return (nni_copyout_bool(b, buf, szp, t));
|
}
|
|
static int
|
ws_get_send_text(void *arg, void *buf, size_t *szp, nni_type t)
|
{
|
nni_ws *ws = arg;
|
bool b;
|
nni_mtx_lock(&ws->mtx);
|
b = ws->send_text;
|
nni_mtx_unlock(&ws->mtx);
|
|
return (nni_copyout_bool(b, buf, szp, t));
|
}
|
|
static const nni_option ws_options[] = {
|
{
|
.o_name = NNG_OPT_WS_REQUEST_HEADERS,
|
.o_get = ws_get_request_headers,
|
},
|
{
|
.o_name = NNG_OPT_WS_RESPONSE_HEADERS,
|
.o_get = ws_get_response_headers,
|
},
|
{
|
.o_name = NNG_OPT_WS_REQUEST_URI,
|
.o_get = ws_get_request_uri,
|
},
|
{
|
.o_name = NNG_OPT_WS_RECV_TEXT,
|
.o_get = ws_get_recv_text,
|
},
|
{
|
.o_name = NNG_OPT_WS_SEND_TEXT,
|
.o_get = ws_get_send_text,
|
},
|
{
|
.o_name = NULL,
|
},
|
};
|
|
static int
|
ws_str_set(void *arg, const char *nm, const void *buf, size_t sz, nni_type t)
|
{
|
nni_ws *ws = arg;
|
int rv;
|
|
// Headers can only be set.
|
nni_mtx_lock(&ws->mtx);
|
if (ws->closed) {
|
nni_mtx_unlock(&ws->mtx);
|
return (NNG_ECLOSED);
|
}
|
nni_mtx_unlock(&ws->mtx);
|
rv = nni_http_conn_setopt(ws->http, nm, buf, sz, t);
|
if (rv == NNG_ENOTSUP) {
|
rv = nni_setopt(ws_options, nm, ws, buf, sz, t);
|
}
|
if (rv == NNG_ENOTSUP) {
|
if (startswith(nm, NNG_OPT_WS_REQUEST_HEADER) ||
|
startswith(nm, NNG_OPT_WS_RESPONSE_HEADER)) {
|
return (NNG_EREADONLY);
|
}
|
}
|
|
return (rv);
|
}
|
|
static int
|
ws_get_req_header(
|
nni_ws *ws, const char *nm, void *buf, size_t *szp, nni_type t)
|
{
|
const char *s;
|
nm += strlen(NNG_OPT_WS_REQUEST_HEADER);
|
s = nni_http_req_get_header(ws->req, nm);
|
if (s == NULL) {
|
return (NNG_ENOENT);
|
}
|
return (nni_copyout_str(s, buf, szp, t));
|
}
|
|
static int
|
ws_get_res_header(
|
nni_ws *ws, const char *nm, void *buf, size_t *szp, nni_type t)
|
{
|
const char *s;
|
nm += strlen(NNG_OPT_WS_RESPONSE_HEADER);
|
s = nni_http_res_get_header(ws->res, nm);
|
if (s == NULL) {
|
return (NNG_ENOENT);
|
}
|
return (nni_copyout_str(s, buf, szp, t));
|
}
|
|
static int
|
ws_str_get(void *arg, const char *nm, void *buf, size_t *szp, nni_type t)
|
{
|
nni_ws *ws = arg;
|
int rv;
|
|
nni_mtx_lock(&ws->mtx);
|
if (ws->closed) {
|
nni_mtx_unlock(&ws->mtx);
|
return (NNG_ECLOSED);
|
}
|
nni_mtx_unlock(&ws->mtx);
|
rv = nni_http_conn_getopt(ws->http, nm, buf, szp, t);
|
if (rv == NNG_ENOTSUP) {
|
rv = nni_getopt(ws_options, nm, ws, buf, szp, t);
|
}
|
// Check for generic headers...
|
if (rv == NNG_ENOTSUP) {
|
if (startswith(nm, NNG_OPT_WS_REQUEST_HEADER)) {
|
rv = ws_get_req_header(ws, nm, buf, szp, t);
|
} else if (startswith(nm, NNG_OPT_WS_RESPONSE_HEADER)) {
|
rv = ws_get_res_header(ws, nm, buf, szp, t);
|
}
|
}
|
return (rv);
|
}
|
