// // Copyright 2020 Staysail Systems, Inc. // Copyright 2018 Capitar IT Group BV // Copyright 2019 Devolutions // // 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 #include #include #include "core/nng_impl.h" #include "supplemental/websocket/websocket.h" #include #include typedef struct ws_dialer ws_dialer; typedef struct ws_listener ws_listener; typedef struct ws_pipe ws_pipe; struct ws_dialer { uint16_t peer; // remote protocol nni_list aios; nni_mtx mtx; nni_aio * connaio; nng_stream_dialer *dialer; bool started; }; struct ws_listener { uint16_t peer; // remote protocol nni_list aios; nni_mtx mtx; nni_aio * accaio; nng_stream_listener *listener; bool started; }; struct ws_pipe { nni_mtx mtx; bool closed; uint16_t peer; nni_aio * user_txaio; nni_aio * user_rxaio; nni_aio * txaio; nni_aio * rxaio; nng_stream *ws; }; static void wstran_pipe_send_cb(void *arg) { ws_pipe *p = arg; nni_aio *taio; nni_aio *uaio; nni_mtx_lock(&p->mtx); taio = p->txaio; uaio = p->user_txaio; p->user_txaio = NULL; if (uaio != NULL) { int rv; if ((rv = nni_aio_result(taio)) != 0) { nni_aio_finish_error(uaio, rv); } else { nni_aio_finish(uaio, 0, 0); } } nni_mtx_unlock(&p->mtx); } static void wstran_pipe_recv_cb(void *arg) { ws_pipe *p = arg; nni_aio *raio = p->rxaio; nni_aio *uaio; int rv; nni_mtx_lock(&p->mtx); uaio = p->user_rxaio; p->user_rxaio = NULL; if ((rv = nni_aio_result(raio)) != 0) { if (uaio != NULL) { nni_aio_finish_error(uaio, rv); } } else { nni_msg *msg = nni_aio_get_msg(raio); if (uaio != NULL) { nni_aio_finish_msg(uaio, msg); } else { nni_msg_free(msg); } } nni_mtx_unlock(&p->mtx); } static void wstran_pipe_recv_cancel(nni_aio *aio, void *arg, int rv) { ws_pipe *p = arg; nni_mtx_lock(&p->mtx); if (p->user_rxaio != aio) { nni_mtx_unlock(&p->mtx); return; } p->user_rxaio = NULL; nni_aio_abort(p->rxaio, rv); nni_aio_finish_error(aio, rv); nni_mtx_unlock(&p->mtx); } static void wstran_pipe_recv(void *arg, nni_aio *aio) { ws_pipe *p = arg; int rv; if (nni_aio_begin(aio) != 0) { return; } nni_mtx_lock(&p->mtx); if ((rv = nni_aio_schedule(aio, wstran_pipe_recv_cancel, p)) != 0) { nni_mtx_unlock(&p->mtx); nni_aio_finish_error(aio, rv); return; } p->user_rxaio = aio; nng_stream_recv(p->ws, p->rxaio); nni_mtx_unlock(&p->mtx); } static void wstran_pipe_send_cancel(nni_aio *aio, void *arg, int rv) { ws_pipe *p = arg; nni_mtx_lock(&p->mtx); if (p->user_txaio != aio) { nni_mtx_unlock(&p->mtx); return; } p->user_txaio = NULL; nni_aio_abort(p->txaio, rv); nni_aio_finish_error(aio, rv); nni_mtx_unlock(&p->mtx); } static void wstran_pipe_send(void *arg, nni_aio *aio) { ws_pipe *p = arg; int rv; if (nni_aio_begin(aio) != 0) { return; } nni_mtx_lock(&p->mtx); if ((rv = nni_aio_schedule(aio, wstran_pipe_send_cancel, p)) != 0) { nni_mtx_unlock(&p->mtx); nni_aio_finish_error(aio, rv); return; } p->user_txaio = aio; nni_aio_set_msg(p->txaio, nni_aio_get_msg(aio)); nni_aio_set_msg(aio, NULL); nng_stream_send(p->ws, p->txaio); nni_mtx_unlock(&p->mtx); } static void wstran_pipe_stop(void *arg) { ws_pipe *p = arg; nni_aio_stop(p->rxaio); nni_aio_stop(p->txaio); } static int wstran_pipe_init(void *arg, nni_pipe *pipe) { NNI_ARG_UNUSED(arg); NNI_ARG_UNUSED(pipe); return (0); } static void wstran_pipe_fini(void *arg) { ws_pipe *p = arg; nni_aio_free(p->rxaio); nni_aio_free(p->txaio); nng_stream_free(p->ws); nni_mtx_fini(&p->mtx); NNI_FREE_STRUCT(p); } static void wstran_pipe_close(void *arg) { ws_pipe *p = arg; nni_aio_close(p->rxaio); nni_aio_close(p->txaio); nni_mtx_lock(&p->mtx); nng_stream_close(p->ws); nni_mtx_unlock(&p->mtx); } static int wstran_pipe_alloc(ws_pipe **pipep, void *ws) { ws_pipe *p; int rv; if ((p = NNI_ALLOC_STRUCT(p)) == NULL) { return (NNG_ENOMEM); } nni_mtx_init(&p->mtx); // Initialize AIOs. if (((rv = nni_aio_alloc(&p->txaio, wstran_pipe_send_cb, p)) != 0) || ((rv = nni_aio_alloc(&p->rxaio, wstran_pipe_recv_cb, p)) != 0)) { wstran_pipe_fini(p); return (rv); } p->ws = ws; *pipep = p; return (0); } static uint16_t wstran_pipe_peer(void *arg) { ws_pipe *p = arg; return (p->peer); } static int ws_listener_bind(void *arg) { ws_listener *l = arg; int rv; if ((rv = nng_stream_listener_listen(l->listener)) == 0) { l->started = true; } return (rv); } static void ws_listener_cancel(nni_aio *aio, void *arg, int rv) { 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 wstran_listener_accept(void *arg, nni_aio *aio) { ws_listener *l = arg; int rv; // We already bound, so we just need to look for an available // pipe (created by the handler), and match it. // Otherwise we stick the AIO in the accept list. if (nni_aio_begin(aio) != 0) { return; } nni_mtx_lock(&l->mtx); if ((rv = nni_aio_schedule(aio, ws_listener_cancel, l)) != 0) { nni_mtx_unlock(&l->mtx); nni_aio_finish_error(aio, rv); return; } nni_list_append(&l->aios, aio); if (aio == nni_list_first(&l->aios)) { nng_stream_listener_accept(l->listener, l->accaio); } nni_mtx_unlock(&l->mtx); } static void wstran_dialer_cancel(nni_aio *aio, void *arg, int rv) { ws_dialer *d = arg; nni_mtx_lock(&d->mtx); if (nni_aio_list_active(aio)) { nni_aio_list_remove(aio); nni_aio_finish_error(aio, rv); } nni_mtx_unlock(&d->mtx); } static void wstran_dialer_connect(void *arg, nni_aio *aio) { ws_dialer *d = arg; int rv; if (nni_aio_begin(aio) != 0) { return; } nni_mtx_lock(&d->mtx); if ((rv = nni_aio_schedule(aio, wstran_dialer_cancel, d)) != 0) { nni_mtx_unlock(&d->mtx); nni_aio_finish_error(aio, rv); return; } NNI_ASSERT(nni_list_empty(&d->aios)); d->started = true; nni_list_append(&d->aios, aio); nng_stream_dialer_dial(d->dialer, d->connaio); nni_mtx_unlock(&d->mtx); } static const nni_option ws_pipe_options[] = { // terminate list { .o_name = NULL, } }; static int wstran_pipe_getopt( void *arg, const char *name, void *buf, size_t *szp, nni_type t) { ws_pipe *p = arg; int rv; if ((rv = nni_stream_get(p->ws, name, buf, szp, t)) == NNG_ENOTSUP) { rv = nni_getopt(ws_pipe_options, name, p, buf, szp, t); } return (rv); } static nni_sp_pipe_ops ws_pipe_ops = { .p_init = wstran_pipe_init, .p_fini = wstran_pipe_fini, .p_stop = wstran_pipe_stop, .p_send = wstran_pipe_send, .p_recv = wstran_pipe_recv, .p_close = wstran_pipe_close, .p_peer = wstran_pipe_peer, .p_getopt = wstran_pipe_getopt, }; static void wstran_dialer_fini(void *arg) { ws_dialer *d = arg; nni_aio_stop(d->connaio); nng_stream_dialer_free(d->dialer); nni_aio_free(d->connaio); nni_mtx_fini(&d->mtx); NNI_FREE_STRUCT(d); } static void wstran_listener_fini(void *arg) { ws_listener *l = arg; nni_aio_stop(l->accaio); nng_stream_listener_free(l->listener); nni_aio_free(l->accaio); nni_mtx_fini(&l->mtx); NNI_FREE_STRUCT(l); } static void wstran_connect_cb(void *arg) { ws_dialer * d = arg; ws_pipe * p; nni_aio * caio = d->connaio; nni_aio * uaio; int rv; nng_stream *ws = NULL; nni_mtx_lock(&d->mtx); if (nni_aio_result(caio) == 0) { ws = nni_aio_get_output(caio, 0); } if ((uaio = nni_list_first(&d->aios)) == NULL) { // The client stopped caring about this! nng_stream_free(ws); nni_mtx_unlock(&d->mtx); return; } nni_aio_list_remove(uaio); NNI_ASSERT(nni_list_empty(&d->aios)); if ((rv = nni_aio_result(caio)) != 0) { nni_aio_finish_error(uaio, rv); } else if ((rv = wstran_pipe_alloc(&p, ws)) != 0) { nng_stream_free(ws); nni_aio_finish_error(uaio, rv); } else { p->peer = d->peer; nni_aio_set_output(uaio, 0, p); nni_aio_finish(uaio, 0, 0); } nni_mtx_unlock(&d->mtx); } static void wstran_dialer_close(void *arg) { ws_dialer *d = arg; nni_aio_close(d->connaio); nng_stream_dialer_close(d->dialer); } static void wstran_listener_close(void *arg) { ws_listener *l = arg; nni_aio_close(l->accaio); nng_stream_listener_close(l->listener); } static void wstran_accept_cb(void *arg) { ws_listener *l = arg; nni_aio * aaio = l->accaio; nni_aio * uaio; int rv; nni_mtx_lock(&l->mtx); uaio = nni_list_first(&l->aios); if ((rv = nni_aio_result(aaio)) != 0) { if (uaio != NULL) { nni_aio_list_remove(uaio); nni_aio_finish_error(uaio, rv); } } else { nng_stream *ws = nni_aio_get_output(aaio, 0); if (uaio != NULL) { ws_pipe *p; // Make a pipe nni_aio_list_remove(uaio); if ((rv = wstran_pipe_alloc(&p, ws)) != 0) { nng_stream_close(ws); nni_aio_finish_error(uaio, rv); } else { p->peer = l->peer; nni_aio_set_output(uaio, 0, p); nni_aio_finish(uaio, 0, 0); } } } if (!nni_list_empty(&l->aios)) { nng_stream_listener_accept(l->listener, aaio); } nni_mtx_unlock(&l->mtx); } static int wstran_dialer_init(void **dp, nng_url *url, nni_dialer *ndialer) { ws_dialer *d; nni_sock * s = nni_dialer_sock(ndialer); int rv; char name[64]; if ((d = NNI_ALLOC_STRUCT(d)) == NULL) { return (NNG_ENOMEM); } nni_mtx_init(&d->mtx); nni_aio_list_init(&d->aios); d->peer = nni_sock_peer_id(s); snprintf( name, sizeof(name), "%s.sp.nanomsg.org", nni_sock_peer_name(s)); if (((rv = nni_ws_dialer_alloc(&d->dialer, url)) != 0) || ((rv = nni_aio_alloc(&d->connaio, wstran_connect_cb, d)) != 0) || ((rv = nng_stream_dialer_set_bool( d->dialer, NNI_OPT_WS_MSGMODE, true)) != 0) || ((rv = nng_stream_dialer_set_string( d->dialer, NNG_OPT_WS_PROTOCOL, name)) != 0)) { wstran_dialer_fini(d); return (rv); } *dp = d; return (0); } static int wstran_listener_init(void **lp, nng_url *url, nni_listener *listener) { ws_listener *l; int rv; nni_sock * s = nni_listener_sock(listener); char name[64]; if ((l = NNI_ALLOC_STRUCT(l)) == NULL) { return (NNG_ENOMEM); } nni_mtx_init(&l->mtx); nni_aio_list_init(&l->aios); l->peer = nni_sock_peer_id(s); snprintf( name, sizeof(name), "%s.sp.nanomsg.org", nni_sock_proto_name(s)); if (((rv = nni_ws_listener_alloc(&l->listener, url)) != 0) || ((rv = nni_aio_alloc(&l->accaio, wstran_accept_cb, l)) != 0) || ((rv = nng_stream_listener_set_bool( l->listener, NNI_OPT_WS_MSGMODE, true)) != 0) || ((rv = nng_stream_listener_set_string( l->listener, NNG_OPT_WS_PROTOCOL, name)) != 0)) { wstran_listener_fini(l); return (rv); } *lp = l; return (0); } static void wstran_init(void) { } static void wstran_fini(void) { } static const nni_option wstran_ep_opts[] = { // terminate list { .o_name = NULL, }, }; static int wstran_dialer_getopt( void *arg, const char *name, void *buf, size_t *szp, nni_type t) { ws_dialer *d = arg; int rv; rv = nni_stream_dialer_get(d->dialer, name, buf, szp, t); if (rv == NNG_ENOTSUP) { rv = nni_getopt(wstran_ep_opts, name, d, buf, szp, t); } return (rv); } static int wstran_dialer_setopt( void *arg, const char *name, const void *buf, size_t sz, nni_type t) { ws_dialer *d = arg; int rv; rv = nni_stream_dialer_set(d->dialer, name, buf, sz, t); if (rv == NNG_ENOTSUP) { rv = nni_setopt(wstran_ep_opts, name, d, buf, sz, t); } return (rv); } static int wstran_listener_get( void *arg, const char *name, void *buf, size_t *szp, nni_type t) { ws_listener *l = arg; int rv; rv = nni_stream_listener_get(l->listener, name, buf, szp, t); if (rv == NNG_ENOTSUP) { rv = nni_getopt(wstran_ep_opts, name, l, buf, szp, t); } return (rv); } static int wstran_listener_set( void *arg, const char *name, const void *buf, size_t sz, nni_type t) { ws_listener *l = arg; int rv; rv = nni_stream_listener_set(l->listener, name, buf, sz, t); if (rv == NNG_ENOTSUP) { rv = nni_setopt(wstran_ep_opts, name, l, buf, sz, t); } return (rv); } static nni_sp_dialer_ops ws_dialer_ops = { .d_init = wstran_dialer_init, .d_fini = wstran_dialer_fini, .d_connect = wstran_dialer_connect, .d_close = wstran_dialer_close, .d_setopt = wstran_dialer_setopt, .d_getopt = wstran_dialer_getopt, }; static nni_sp_listener_ops ws_listener_ops = { .l_init = wstran_listener_init, .l_fini = wstran_listener_fini, .l_bind = ws_listener_bind, .l_accept = wstran_listener_accept, .l_close = wstran_listener_close, .l_setopt = wstran_listener_set, .l_getopt = wstran_listener_get, }; static nni_sp_tran ws_tran = { .tran_scheme = "ws", .tran_dialer = &ws_dialer_ops, .tran_listener = &ws_listener_ops, .tran_pipe = &ws_pipe_ops, .tran_init = wstran_init, .tran_fini = wstran_fini, }; static nni_sp_tran ws4_tran = { .tran_scheme = "ws4", .tran_dialer = &ws_dialer_ops, .tran_listener = &ws_listener_ops, .tran_pipe = &ws_pipe_ops, .tran_init = wstran_init, .tran_fini = wstran_fini, }; static nni_sp_tran ws6_tran = { .tran_scheme = "ws6", .tran_dialer = &ws_dialer_ops, .tran_listener = &ws_listener_ops, .tran_pipe = &ws_pipe_ops, .tran_init = wstran_init, .tran_fini = wstran_fini, }; #ifndef NNG_ELIDE_DEPRECATED int nng_ws_register(void) { return (nni_init()); } int nng_wss_register(void) { return (nni_init()); } #endif void nni_sp_ws_register(void) { nni_sp_tran_register(&ws_tran); nni_sp_tran_register(&ws4_tran); nni_sp_tran_register(&ws6_tran); } #ifdef NNG_TRANSPORT_WSS static nni_sp_tran wss_tran = { .tran_scheme = "wss", .tran_dialer = &ws_dialer_ops, .tran_listener = &ws_listener_ops, .tran_pipe = &ws_pipe_ops, .tran_init = wstran_init, .tran_fini = wstran_fini, }; static nni_sp_tran wss4_tran = { .tran_scheme = "wss4", .tran_dialer = &ws_dialer_ops, .tran_listener = &ws_listener_ops, .tran_pipe = &ws_pipe_ops, .tran_init = wstran_init, .tran_fini = wstran_fini, }; static nni_sp_tran wss6_tran = { .tran_scheme = "wss6", .tran_dialer = &ws_dialer_ops, .tran_listener = &ws_listener_ops, .tran_pipe = &ws_pipe_ops, .tran_init = wstran_init, .tran_fini = wstran_fini, }; void nni_sp_wss_register(void) { nni_sp_tran_register(&wss_tran); nni_sp_tran_register(&wss4_tran); nni_sp_tran_register(&wss6_tran); } #endif // NNG_TRANSPORT_WSS