diff --git a/core/uwsgi.c b/core/uwsgi.c index 62707a78..a26e44f4 100644 --- a/core/uwsgi.c +++ b/core/uwsgi.c @@ -495,9 +495,6 @@ static struct uwsgi_option uwsgi_base_options[] = { {"websockets-ping-freq", required_argument, 0, "set the frequency (in seconds) of websockets automatic ping packets", uwsgi_opt_set_int, &uwsgi.websockets_ping_freq, 0}, {"websocket-ping-freq", required_argument, 0, "set the frequency (in seconds) of websockets automatic ping packets", uwsgi_opt_set_int, &uwsgi.websockets_ping_freq, 0}, - {"websockets-pong-freq", required_argument, 0, "set the frequency (in seconds) of websockets automatic pong/keepalive packets", uwsgi_opt_set_int, &uwsgi.websockets_pong_freq, 0}, - {"websocket-pong-freq", required_argument, 0, "set the frequency (in seconds) of websockets automatic pong/keepalive packets", uwsgi_opt_set_int, &uwsgi.websockets_pong_freq, 0}, - {"websockets-max-size", required_argument, 0, "set the max allowed size of websocket messages (in Kbytes, default 1024)", uwsgi_opt_set_64bit, &uwsgi.websockets_max_size, 0}, {"websocket-max-size", required_argument, 0, "set the max allowed size of websocket messages (in Kbytes, default 1024)", uwsgi_opt_set_64bit, &uwsgi.websockets_max_size, 0}, diff --git a/core/websockets.c b/core/websockets.c index 15c950f6..8048d323 100644 --- a/core/websockets.c +++ b/core/websockets.c @@ -10,7 +10,7 @@ extern struct uwsgi_server uwsgi; -struct uwsgi_buffer *uwsgi_websocket_message(char *msg, size_t len) { +static struct uwsgi_buffer *uwsgi_websocket_message(char *msg, size_t len) { struct uwsgi_buffer *ub = uwsgi_buffer_new(10 + len); if (uwsgi_buffer_u8(ub, 0x81)) goto error; if (len < 126) { @@ -33,7 +33,7 @@ error: return NULL; } -int uwsgi_websockets_ping(struct wsgi_request *wsgi_req) { +static int uwsgi_websockets_ping(struct wsgi_request *wsgi_req) { if (uwsgi_response_write_body_do(wsgi_req, uwsgi.websockets_ping->buf, uwsgi.websockets_ping->pos)) { return -1; } @@ -41,28 +41,29 @@ int uwsgi_websockets_ping(struct wsgi_request *wsgi_req) { return 0; } -int uwsgi_websockets_pong(struct wsgi_request *wsgi_req) { - time_t now = uwsgi_now(); - if (wsgi_req->websocket_last_ping == 0 || - ( wsgi_req->websocket_last_ping > 0 && now - wsgi_req->websocket_last_ping > uwsgi.websockets_ping_freq)) { - - if (uwsgi_websockets_ping(wsgi_req)) return -1; - return 0; - } - // check if last pong arrived in time - else if (wsgi_req->websocket_last_ping > 0) { - if (wsgi_req->websocket_last_pong < wsgi_req->websocket_last_ping) { - if (wsgi_req->websocket_last_ping - wsgi_req->websocket_last_pong > - uwsgi.websockets_ping_freq) { - uwsgi_log("[uwsgi-websocket] no PONG received in %d seconds !!!\n", uwsgi.websockets_ping_freq); - return -1; - } - } - } +static int uwsgi_websockets_pong(struct wsgi_request *wsgi_req) { return uwsgi_response_write_body_do(wsgi_req, uwsgi.websockets_pong->buf, uwsgi.websockets_pong->pos); } -int uwsgi_websocket_send_do(struct wsgi_request *wsgi_req, char *msg, size_t len) { +static int uwsgi_websockets_check_pingpong(struct wsgi_request *wsgi_req) { + time_t now = uwsgi_now(); + if (wsgi_req->websocket_last_ping == 0 || + (now - wsgi_req->websocket_last_ping > uwsgi.websockets_ping_freq)) { + if (uwsgi_websockets_ping(wsgi_req)) return -1; + } + else if (wsgi_req->websocket_last_ping > 0) { + if (wsgi_req->websocket_last_pong < wsgi_req->websocket_last_ping) { + if (wsgi_req->websocket_last_ping - wsgi_req->websocket_last_pong > + uwsgi.websockets_ping_freq) { + uwsgi_log("[uwsgi-websocket] no PONG received in %d seconds !!!\n", uwsgi.websockets_ping_freq); + return -1; + } + } + } + return 0; +} + +static int uwsgi_websocket_send_do(struct wsgi_request *wsgi_req, char *msg, size_t len) { struct uwsgi_buffer *ub = uwsgi_websocket_message(msg, len); if (!ub) return -1; @@ -83,7 +84,7 @@ int uwsgi_websocket_send(struct wsgi_request *wsgi_req, char *msg, size_t len) { return ret; } -void uwsgi_websocket_parse_header(struct wsgi_request *wsgi_req) { +static void uwsgi_websocket_parse_header(struct wsgi_request *wsgi_req) { uint8_t byte1 = wsgi_req->websocket_buf->buf[0]; uint8_t byte2 = wsgi_req->websocket_buf->buf[1]; wsgi_req->websocket_opcode = byte1 & 0xf; @@ -91,7 +92,7 @@ void uwsgi_websocket_parse_header(struct wsgi_request *wsgi_req) { wsgi_req->websocket_size = byte2 & 0x7f; } -struct uwsgi_buffer *uwsgi_websockets_parse(struct wsgi_request *wsgi_req) { +static struct uwsgi_buffer *uwsgi_websockets_parse(struct wsgi_request *wsgi_req) { // de-mask buffer uint8_t *ptr = (uint8_t *) (wsgi_req->websocket_buf->buf + (wsgi_req->websocket_pktsize - wsgi_req->websocket_size)); size_t i; @@ -115,7 +116,7 @@ error: } -ssize_t uwsgi_websockets_recv_pkt(struct wsgi_request *wsgi_req) { +static ssize_t uwsgi_websockets_recv_pkt(struct wsgi_request *wsgi_req, int nb) { int ret = -1; @@ -125,6 +126,12 @@ ssize_t uwsgi_websockets_recv_pkt(struct wsgi_request *wsgi_req) { if (rlen == 0) return -1; if (rlen < 0) { if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINPROGRESS) { + if (nb) { + if (uwsgi_websockets_check_pingpong(wsgi_req)) { + return -1; + } + return 0; + } goto wait; } uwsgi_error("uwsgi_websockets_recv_pkt()"); @@ -132,7 +139,7 @@ ssize_t uwsgi_websockets_recv_pkt(struct wsgi_request *wsgi_req) { } wait: - ret = uwsgi.wait_read_hook(wsgi_req->fd, uwsgi.websockets_pong_freq); + ret = uwsgi.wait_read_hook(wsgi_req->fd, uwsgi.websockets_ping_freq); if (ret > 0) { rlen = wsgi_req->socket->proto_read_body(wsgi_req, wsgi_req->websocket_buf->buf + wsgi_req->websocket_buf->pos, wsgi_req->websocket_buf->len - wsgi_req->websocket_buf->pos); if (rlen > 0) return rlen; @@ -143,7 +150,7 @@ wait: return -1; } // send unsolicited pong - if (uwsgi_websockets_pong(wsgi_req)) { + if (uwsgi_websockets_check_pingpong(wsgi_req)) { return -1; } } @@ -152,7 +159,7 @@ wait: } -struct uwsgi_buffer *uwsgi_websocket_recv_do(struct wsgi_request *wsgi_req) { +static struct uwsgi_buffer *uwsgi_websocket_recv_do(struct wsgi_request *wsgi_req, int nb) { if (!wsgi_req->websocket_buf) { // this buffer will be destroyed on connection close wsgi_req->websocket_buf = uwsgi_buffer_new(uwsgi.page_size); @@ -257,8 +264,12 @@ struct uwsgi_buffer *uwsgi_websocket_recv_do(struct wsgi_request *wsgi_req) { // need more data else { if (uwsgi_buffer_ensure(wsgi_req->websocket_buf, uwsgi.page_size)) return NULL; - ssize_t len = uwsgi_websockets_recv_pkt(wsgi_req); + ssize_t len = uwsgi_websockets_recv_pkt(wsgi_req, nb); if (len <= 0) { + if (nb == 1 && len == 0) { + // return an empty buffer to signal blocking event + return uwsgi_buffer_new(0); + } return NULL; } // update buffer size @@ -273,13 +284,25 @@ struct uwsgi_buffer *uwsgi_websocket_recv(struct wsgi_request *wsgi_req) { if (wsgi_req->websocket_closed) { return NULL; } - struct uwsgi_buffer *ub = uwsgi_websocket_recv_do(wsgi_req); + struct uwsgi_buffer *ub = uwsgi_websocket_recv_do(wsgi_req, 0); if (!ub) { wsgi_req->websocket_closed = 1; } return ub; } +struct uwsgi_buffer *uwsgi_websocket_recv_nb(struct wsgi_request *wsgi_req) { + if (wsgi_req->websocket_closed) { + return NULL; + } + struct uwsgi_buffer *ub = uwsgi_websocket_recv_do(wsgi_req, 1); + if (!ub) { + wsgi_req->websocket_closed = 1; + } + return ub; +} + + ssize_t uwsgi_websockets_simple_send(struct wsgi_request *wsgi_req, struct uwsgi_buffer *ub) { ssize_t len = wsgi_req->socket->proto_write(wsgi_req, ub->buf, ub->pos); @@ -313,6 +336,8 @@ int uwsgi_websocket_handshake(struct wsgi_request *wsgi_req, char *key, uint16_t } free(b64); + wsgi_req->websocket_last_pong = uwsgi_now(); + return uwsgi_response_write_headers_do(wsgi_req); #else uwsgi_log("you need to build uWSGI with SSL support to use the websocket handshake api function !!!\n"); @@ -326,6 +351,5 @@ void uwsgi_websockets_init() { uwsgi.websockets_ping = uwsgi_buffer_new(2); uwsgi_buffer_append(uwsgi.websockets_ping, "\x89\0", 2); uwsgi.websockets_ping_freq = 30; - uwsgi.websockets_pong_freq = 10; uwsgi.websockets_max_size = 1024; } diff --git a/plugins/gevent/gevent.c b/plugins/gevent/gevent.c index 96ca774b..2af8e404 100644 --- a/plugins/gevent/gevent.c +++ b/plugins/gevent/gevent.c @@ -74,15 +74,17 @@ static void uwsgi_gevent_gbcw() { } struct wsgi_request *uwsgi_gevent_current_wsgi_req(void) { + struct wsgi_request *wsgi_req = NULL; PyObject *current_greenlet = GET_CURRENT_GREENLET; PyObject *py_wsgi_req = PyObject_GetAttrString(current_greenlet, "uwsgi_wsgi_req"); // not in greenlet if (!py_wsgi_req) { uwsgi_log("[BUG] current_wsgi_req NOT FOUND !!!\n"); - exit(1); + goto end; } - struct wsgi_request *wsgi_req = (struct wsgi_request*) PyLong_AsLong(py_wsgi_req); + wsgi_req = (struct wsgi_request*) PyLong_AsLong(py_wsgi_req); Py_DECREF(py_wsgi_req); +end: Py_DECREF(current_greenlet); return wsgi_req; } @@ -408,8 +410,13 @@ static void gevent_loop() { python_call(ugevent.signal, ge_signal_tuple, 0, NULL); - if (!PyObject_CallMethod(ugevent.hub, "join", NULL)) { - PyErr_Print(); + for(;;) { + if (!PyObject_CallMethod(ugevent.hub, "join", NULL)) { + PyErr_Print(); + } + else { + break; + } } if (uwsgi.workers[uwsgi.mywid].manage_next_request == 0) { diff --git a/plugins/psgi/uwsgi_plmodule.c b/plugins/psgi/uwsgi_plmodule.c index bcfd143f..07a1a77a 100644 --- a/plugins/psgi/uwsgi_plmodule.c +++ b/plugins/psgi/uwsgi_plmodule.c @@ -378,6 +378,25 @@ XS(XS_websocket_recv) { XSRETURN(1); } +XS(XS_websocket_recv_nb) { + dXSARGS; + + psgi_check_args(0); + + struct wsgi_request *wsgi_req = current_wsgi_req(); + struct uwsgi_buffer *ub = uwsgi_websocket_recv_nb(wsgi_req); + if (!ub) { + croak("unable to receive websocket message"); + XSRETURN_UNDEF; + } + + ST(0) = newSVpv(ub->buf, ub->pos); + uwsgi_buffer_destroy(ub); + sv_2mortal(ST(0)); + + XSRETURN(1); +} + void init_perl_embedded_module() { @@ -400,6 +419,7 @@ void init_perl_embedded_module() { psgi_xs(alarm); psgi_xs(websocket_handshake); psgi_xs(websocket_recv); + psgi_xs(websocket_recv_nb); psgi_xs(websocket_send); psgi_xs(postfork); psgi_xs(atexit); diff --git a/plugins/python/uwsgi_pymodule.c b/plugins/python/uwsgi_pymodule.c index a3f63e13..a98bb242 100644 --- a/plugins/python/uwsgi_pymodule.c +++ b/plugins/python/uwsgi_pymodule.c @@ -5,7 +5,7 @@ extern struct uwsgi_python up; PyObject *py_uwsgi_signal_wait(PyObject * self, PyObject * args) { - struct wsgi_request *wsgi_req = current_wsgi_req(); + struct wsgi_request *wsgi_req = py_current_wsgi_req(); int wait_for_specific_signal = 0; uint8_t uwsgi_signal = 0; int received_signal; @@ -42,7 +42,7 @@ PyObject *py_uwsgi_signal_wait(PyObject * self, PyObject * args) { PyObject *py_uwsgi_signal_received(PyObject * self, PyObject * args) { - struct wsgi_request *wsgi_req = current_wsgi_req(); + struct wsgi_request *wsgi_req = py_current_wsgi_req(); return PyInt_FromLong(wsgi_req->signal_received); } @@ -506,7 +506,7 @@ clear: PyObject *py_uwsgi_log_this(PyObject * self, PyObject * args) { - struct wsgi_request *wsgi_req = current_wsgi_req(); + struct wsgi_request *wsgi_req = py_current_wsgi_req(); wsgi_req->log_this = 1; @@ -532,7 +532,7 @@ PyObject *py_uwsgi_get_logvar(PyObject * self, PyObject * args) { char *key = NULL; Py_ssize_t keylen = 0; - struct wsgi_request *wsgi_req = current_wsgi_req(); + struct wsgi_request *wsgi_req = py_current_wsgi_req(); if (!PyArg_ParseTuple(args, "s#:get_logvar", &key, &keylen)) { return NULL; @@ -554,7 +554,7 @@ PyObject *py_uwsgi_set_logvar(PyObject * self, PyObject * args) { Py_ssize_t keylen = 0; char *val = NULL; Py_ssize_t vallen = 0; - struct wsgi_request *wsgi_req = current_wsgi_req(); + struct wsgi_request *wsgi_req = py_current_wsgi_req(); if (!PyArg_ParseTuple(args, "s#s#:set_logvar", &key, &keylen, &val, &vallen)) { return NULL; @@ -677,7 +677,7 @@ PyObject *py_uwsgi_send(PyObject * self, PyObject * args) { PyObject *data; PyObject *arg1, *arg2; - struct wsgi_request *wsgi_req = current_wsgi_req(); + struct wsgi_request *wsgi_req = py_current_wsgi_req(); int uwsgi_fd = wsgi_req->fd; @@ -708,7 +708,7 @@ PyObject *py_uwsgi_offload(PyObject * self, PyObject * args) { /* size_t len = 0; char *filename = NULL; - struct wsgi_request *wsgi_req = current_wsgi_req(); + struct wsgi_request *wsgi_req = py_current_wsgi_req(); if (!PyArg_ParseTuple(args, "s|i:offload_transfer", &filename, &len)) { return NULL; @@ -747,7 +747,7 @@ PyObject *py_uwsgi_advanced_sendfile(PyObject * self, PyObject * args) { size_t chunk = 0; off_t pos = 0; size_t filesize = 0; - struct wsgi_request *wsgi_req = current_wsgi_req(); + struct wsgi_request *wsgi_req = py_current_wsgi_req(); int fd = -1; @@ -964,7 +964,7 @@ PyObject *py_uwsgi_unlock(PyObject * self, PyObject * args) { } PyObject *py_uwsgi_connection_fd(PyObject * self, PyObject * args) { - struct wsgi_request *wsgi_req = current_wsgi_req(); + struct wsgi_request *wsgi_req = py_current_wsgi_req(); return PyInt_FromLong(wsgi_req->fd); } @@ -979,7 +979,7 @@ PyObject *py_uwsgi_websocket_handshake(PyObject * self, PyObject * args) { return NULL; } - struct wsgi_request *wsgi_req = current_wsgi_req(); + struct wsgi_request *wsgi_req = py_current_wsgi_req(); UWSGI_RELEASE_GIL int ret = uwsgi_websocket_handshake(wsgi_req, key, key_len, origin, origin_len); @@ -1001,7 +1001,7 @@ PyObject *py_uwsgi_websocket_send(PyObject * self, PyObject * args) { return NULL; } - struct wsgi_request *wsgi_req = current_wsgi_req(); + struct wsgi_request *wsgi_req = py_current_wsgi_req(); UWSGI_RELEASE_GIL int ret = uwsgi_websocket_send(wsgi_req, message, message_len); @@ -1014,7 +1014,7 @@ PyObject *py_uwsgi_websocket_send(PyObject * self, PyObject * args) { } PyObject *py_uwsgi_websocket_recv(PyObject * self, PyObject * args) { - struct wsgi_request *wsgi_req = current_wsgi_req(); + struct wsgi_request *wsgi_req = py_current_wsgi_req(); UWSGI_RELEASE_GIL struct uwsgi_buffer *ub = uwsgi_websocket_recv(wsgi_req); UWSGI_GET_GIL @@ -1027,6 +1027,20 @@ PyObject *py_uwsgi_websocket_recv(PyObject * self, PyObject * args) { return ret; } +PyObject *py_uwsgi_websocket_recv_nb(PyObject * self, PyObject * args) { + struct wsgi_request *wsgi_req = py_current_wsgi_req(); + UWSGI_RELEASE_GIL + struct uwsgi_buffer *ub = uwsgi_websocket_recv_nb(wsgi_req); + UWSGI_GET_GIL + if (!ub) { + return PyErr_Format(PyExc_IOError, "unable to receive websocket message"); + } + + PyObject *ret = PyString_FromStringAndSize(ub->buf, ub->pos); + uwsgi_buffer_destroy(ub); + return ret; +} + PyObject *py_uwsgi_embedded_data(PyObject * self, PyObject * args) { @@ -1634,7 +1648,7 @@ PyObject *py_uwsgi_send_spool(PyObject * self, PyObject * args, PyObject *kw) { char *cur_buf; int i; char spool_filename[1024]; - struct wsgi_request *wsgi_req = current_wsgi_req(); + struct wsgi_request *wsgi_req = py_current_wsgi_req(); char *priority = NULL; long numprio = 0; time_t at = 0; @@ -2212,7 +2226,7 @@ PyObject *py_uwsgi_mem(PyObject * self, PyObject * args) { PyObject *py_uwsgi_cl(PyObject * self, PyObject * args) { - struct wsgi_request *wsgi_req = current_wsgi_req(); + struct wsgi_request *wsgi_req = py_current_wsgi_req(); return PyLong_FromUnsignedLongLong(wsgi_req->post_cl); @@ -2220,7 +2234,7 @@ PyObject *py_uwsgi_cl(PyObject * self, PyObject * args) { PyObject *py_uwsgi_disconnect(PyObject * self, PyObject * args) { - struct wsgi_request *wsgi_req = current_wsgi_req(); + struct wsgi_request *wsgi_req = py_current_wsgi_req(); #ifdef UWSGI_DEBUG uwsgi_log("disconnecting worker %d (pid :%d) from session...\n", uwsgi.mywid, uwsgi.mypid); @@ -2354,7 +2368,7 @@ PyObject *py_uwsgi_parse_file(PyObject * self, PyObject * args) { PyObject *py_uwsgi_grunt(PyObject * self, PyObject * args) { pid_t grunt_pid; - struct wsgi_request *wsgi_req = current_wsgi_req(); + struct wsgi_request *wsgi_req = py_current_wsgi_req(); if (uwsgi.grunt) { uwsgi_log("spawning a grunt from worker %d (pid :%d)...\n", uwsgi.mywid, uwsgi.mypid); @@ -2425,7 +2439,7 @@ static PyMethodDef uwsgi_spooler_methods[] = { PyObject *py_uwsgi_suspend(PyObject * self, PyObject * args) { - struct wsgi_request *wsgi_req = current_wsgi_req(); + struct wsgi_request *wsgi_req = py_current_wsgi_req(); if (uwsgi.schedule_to_main) uwsgi.schedule_to_main(wsgi_req); @@ -2522,6 +2536,7 @@ static PyMethodDef uwsgi_advanced_methods[] = { //{"call_hook", py_uwsgi_call_hook, METH_VARARGS, ""}, {"websocket_recv", py_uwsgi_websocket_recv, METH_VARARGS, ""}, + {"websocket_recv_nb", py_uwsgi_websocket_recv_nb, METH_VARARGS, ""}, {"websocket_send", py_uwsgi_websocket_send, METH_VARARGS, ""}, {"websocket_handshake", py_uwsgi_websocket_handshake, METH_VARARGS, ""}, diff --git a/plugins/python/uwsgi_python.h b/plugins/python/uwsgi_python.h index 92e64bfb..21a30f98 100644 --- a/plugins/python/uwsgi_python.h +++ b/plugins/python/uwsgi_python.h @@ -267,6 +267,11 @@ PyObject *uwsgi_python_setup_thread(char *); #undef UWSGI_MINTERPRETERS #endif +#define py_current_wsgi_req() current_wsgi_req();\ + if (!wsgi_req) {\ + return PyErr_Format(PyExc_SystemError, "you can call uwsgi api function only from the main callable");\ + } + #define uwsgi_pyexit {PyErr_Print();exit(1);} #ifdef __linux__ diff --git a/plugins/python/wsgi_handlers.c b/plugins/python/wsgi_handlers.c index cb63570a..9489e6e6 100644 --- a/plugins/python/wsgi_handlers.c +++ b/plugins/python/wsgi_handlers.c @@ -187,7 +187,7 @@ PyObject *py_uwsgi_write(PyObject * self, PyObject * args) { char *content; size_t content_len; - struct wsgi_request *wsgi_req = current_wsgi_req(); + struct wsgi_request *wsgi_req = py_current_wsgi_req(); data = PyTuple_GetItem(args, 0); if (PyString_Check(data)) { @@ -211,7 +211,7 @@ PyObject *py_uwsgi_write(PyObject * self, PyObject * args) { PyObject *py_eventfd_read(PyObject * self, PyObject * args) { int fd, timeout = 0; - struct wsgi_request *wsgi_req = current_wsgi_req(); + struct wsgi_request *wsgi_req = py_current_wsgi_req(); if (!PyArg_ParseTuple(args, "i|i", &fd, &timeout)) { return NULL; @@ -228,7 +228,7 @@ PyObject *py_eventfd_read(PyObject * self, PyObject * args) { PyObject *py_eventfd_write(PyObject * self, PyObject * args) { int fd, timeout = 0; - struct wsgi_request *wsgi_req = current_wsgi_req(); + struct wsgi_request *wsgi_req = py_current_wsgi_req(); if (!PyArg_ParseTuple(args, "i|i", &fd, &timeout)) { return NULL; @@ -458,7 +458,7 @@ void uwsgi_after_request_wsgi(struct wsgi_request *wsgi_req) { PyObject *py_uwsgi_sendfile(PyObject * self, PyObject * args) { - struct wsgi_request *wsgi_req = current_wsgi_req(); + struct wsgi_request *wsgi_req = py_current_wsgi_req(); if (!PyArg_ParseTuple(args, "O|i:uwsgi_sendfile", &wsgi_req->async_sendfile, &wsgi_req->sendfile_fd_chunk)) { return NULL; diff --git a/plugins/python/wsgi_headers.c b/plugins/python/wsgi_headers.c index ef0bd414..be1ba396 100644 --- a/plugins/python/wsgi_headers.c +++ b/plugins/python/wsgi_headers.c @@ -11,7 +11,7 @@ PyObject *py_uwsgi_spit(PyObject * self, PyObject * args) { PyObject *exc_info = NULL; size_t i; - struct wsgi_request *wsgi_req = current_wsgi_req(); + struct wsgi_request *wsgi_req = py_current_wsgi_req(); // avoid double sending of headers if (wsgi_req->headers_sent) { diff --git a/plugins/rack/rack_api.c b/plugins/rack/rack_api.c index 32282d72..fa238ce3 100644 --- a/plugins/rack/rack_api.c +++ b/plugins/rack/rack_api.c @@ -930,6 +930,21 @@ VALUE uwsgi_ruby_websocket_recv(VALUE *class) { } +VALUE uwsgi_ruby_websocket_recv_nb(VALUE *class) { + + struct wsgi_request *wsgi_req = current_wsgi_req(); + struct uwsgi_buffer *ub = uwsgi_websocket_recv_nb(wsgi_req); + if (!ub) { + rb_raise(rb_eRuntimeError, "unable to receive websocket message"); + return Qnil; + } + VALUE ret = rb_str_new(ub->buf, ub->pos); + uwsgi_buffer_destroy(ub); + return ret; + +} + + void uwsgi_rack_init_api() { @@ -957,6 +972,7 @@ void uwsgi_rack_init_api() { uwsgi_rack_api("websocket_handshake", uwsgi_ruby_websocket_handshake, -1); uwsgi_rack_api("websocket_send", uwsgi_ruby_websocket_send, 1); uwsgi_rack_api("websocket_recv", uwsgi_ruby_websocket_recv, 0); + uwsgi_rack_api("websocket_recv_nb", uwsgi_ruby_websocket_recv_nb, 0); uwsgi_rack_api("setprocname", rack_uwsgi_setprocname, 1); uwsgi_rack_api("mem", rack_uwsgi_mem, 0); diff --git a/tests/websockets.py b/tests/websockets.py index 45fb5b36..8f57e1aa 100644 --- a/tests/websockets.py +++ b/tests/websockets.py @@ -1,6 +1,10 @@ -#!./uwsgi --https :8443,foobar.crt,foobar.key --http-raw-body --gevent 100 --channel room001 --processes 4 --module tests.websocket +#!./uwsgi --https :8443,foobar.crt,foobar.key --http-websockets --gevent 100 --module tests.websocket import uwsgi -import time +import gevent +from gevent.queue import JoinableQueue +from gevent.socket import wait_read + +queue = JoinableQueue() def application(env, sr): @@ -51,9 +55,17 @@ def application(env, sr): elif env['PATH_INFO'] == '/foobar/': uwsgi.websocket_handshake(env['HTTP_SEC_WEBSOCKET_KEY'], env.get('HTTP_ORIGIN', '')) print "websockets..." - uwsgi.websocket_channel_join('room001') while True: - msg = uwsgi.websocket_recv() - print len(msg) - #uwsgi.websocket_send("hello %s = %s" % (time.time(), msg)) - uwsgi.channel_send('room001', "channel %s = %s" % (time.time(), msg)) + msg = uwsgi.websocket_recv_nb() + if msg: + queue.put(msg) + else: + try: + wait_read(uwsgi.connection_fd(), 0.1) + except gevent.socket.timeout: + try: + msg = queue.get_nowait() + uwsgi.websocket_send(msg) + except: + pass + return "" diff --git a/uwsgi.h b/uwsgi.h index 70bc2f8a..149860a3 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -2187,7 +2187,6 @@ struct uwsgi_server { struct uwsgi_buffer *websockets_ping; struct uwsgi_buffer *websockets_pong; int websockets_ping_freq; - int websockets_pong_freq; uint64_t websockets_max_size; int (*wait_write_hook) (int, int); @@ -3724,13 +3723,12 @@ void uwsgi_subscribe_all(uint8_t, int); void uwsgi_websockets_init(void); int uwsgi_websocket_send(struct wsgi_request *, char *, size_t); struct uwsgi_buffer *uwsgi_websocket_recv(struct wsgi_request *); +struct uwsgi_buffer *uwsgi_websocket_recv_nb(struct wsgi_request *); uint16_t uwsgi_be16(char *); uint32_t uwsgi_be32(char *); uint64_t uwsgi_be64(char *); -int uwsgi_websockets_pong(struct wsgi_request *); - int uwsgi_websocket_handshake(struct wsgi_request *, char *, uint16_t, char *, uint16_t); int uwsgi_response_prepare_headers(struct wsgi_request *, char *, uint16_t);