diff --git a/plugins/coroae/coroae.c b/plugins/coroae/coroae.c index 0e80261c..31e33acb 100644 --- a/plugins/coroae/coroae.c +++ b/plugins/coroae/coroae.c @@ -4,8 +4,23 @@ extern struct uwsgi_server uwsgi; extern struct uwsgi_perl uperl; +MGVTBL uwsgi_coroae_vtbl = { 0, 0, 0, 0, 0 }; + #define free_req_queue uwsgi.async_queue_unused_ptr++; uwsgi.async_queue_unused[uwsgi.async_queue_unused_ptr] = wsgi_req +static struct wsgi_request *coroae_current_wsgi_req(void) { + MAGIC *mg; + SV *current = CORO_CURRENT; + for (mg = SvMAGIC (current); mg; mg = mg->mg_moremagic) { + if (mg->mg_type == PERL_MAGIC_ext + 1 && mg->mg_virtual == &uwsgi_coroae_vtbl) { + return (struct wsgi_request *) mg->mg_ptr; + } + } + uwsgi_log("[BUG] current_wsgi_req NOT FOUND !!!\n"); + exit(1); +} + + SV * coroae_coro_new(CV *block) { SV *newobj = NULL; dSP; @@ -18,7 +33,7 @@ SV * coroae_coro_new(CV *block) { call_method("new", G_SCALAR); SPAGAIN; if(SvTRUE(ERRSV)) { - uwsgi_log("[uwsgi-perl error] %s\n", SvPV_nolen(ERRSV)); + uwsgi_log("[uwsgi-perl error] %s", SvPV_nolen(ERRSV)); } else { newobj = SvREFCNT_inc(POPs); @@ -41,7 +56,7 @@ static int coroae_wait_fd_read(int fd, int timeout) { call_pv("Coro::AnyEvent::readable", G_SCALAR); SPAGAIN; if(SvTRUE(ERRSV)) { - uwsgi_log("[uwsgi-perl error] %s\n", SvPV_nolen(ERRSV)); + uwsgi_log("[uwsgi-perl error] %s", SvPV_nolen(ERRSV)); } else { SV *p_ret = POPs; @@ -68,7 +83,7 @@ static int coroae_wait_fd_write(int fd, int timeout) { call_pv("Coro::AnyEvent::writable", G_SCALAR); SPAGAIN; if(SvTRUE(ERRSV)) { - uwsgi_log("[uwsgi-perl error] %s\n", SvPV_nolen(ERRSV)); + uwsgi_log("[uwsgi-perl error] %s", SvPV_nolen(ERRSV)); } else { if (SvTRUE(POPs)) { @@ -176,6 +191,7 @@ edge: CV *async_xs_call = newXS(NULL, XS_coroae_accept_request, "uwsgi::coroae"); CvXSUBANY(async_xs_call).any_ptr = wsgi_req; SV *coro_req = coroae_coro_new(async_xs_call); + sv_magicext(SvRV(coro_req), 0, PERL_MAGIC_ext + 1, &uwsgi_coroae_vtbl, (const char *)wsgi_req, 0); CORO_READY(coro_req); if (uwsgi_sock->edge_trigger) { @@ -221,8 +237,9 @@ static SV *coroae_add_watcher(int fd, SV *cb) { SPAGAIN; if(SvTRUE(ERRSV)) { - uwsgi_log("[uwsgi-perl error] %s\n", SvPV_nolen(ERRSV)); - newobj = NULL; + // no need to continue... + uwsgi_log("[uwsgi-perl error] %s", SvPV_nolen(ERRSV)); + exit(1); } else { newobj = SvREFCNT_inc(POPs); @@ -251,7 +268,7 @@ static SV *coroae_condvar_new() { SPAGAIN; if(SvTRUE(ERRSV)) { - uwsgi_log("[uwsgi-perl error] %s\n", SvPV_nolen(ERRSV)); + uwsgi_log("[uwsgi-perl error] %s", SvPV_nolen(ERRSV)); newobj = NULL; } else { @@ -277,7 +294,7 @@ static void coroae_wait_condvar(SV *cv) { SPAGAIN; if(SvTRUE(ERRSV)) { - uwsgi_log("[uwsgi-perl error] %s\n", SvPV_nolen(ERRSV)); + uwsgi_log("[uwsgi-perl error] %s", SvPV_nolen(ERRSV)); } PUTBACK; FREETMPS; @@ -294,7 +311,7 @@ static void coroae_loop() { exit(1); } - if (!uperl.psgi) { + if (!uperl.loaded) { uwsgi_log("no perl/PSGI code loaded (with --psgi), unable to initialize Coro::AnyEvent\n"); exit(1); } @@ -315,6 +332,7 @@ static void coroae_loop() { exit(1); } + uwsgi.current_wsgi_req = coroae_current_wsgi_req; uwsgi.wait_write_hook = coroae_wait_fd_write; uwsgi.wait_read_hook = coroae_wait_fd_read; diff --git a/plugins/gevent/gevent.c b/plugins/gevent/gevent.c index ae37b081..ec3a0728 100644 --- a/plugins/gevent/gevent.c +++ b/plugins/gevent/gevent.c @@ -63,8 +63,8 @@ struct wsgi_request *uwsgi_gevent_current_wsgi_req(void) { PyObject *py_wsgi_req = PyObject_GetAttrString(current_greenlet, "uwsgi_wsgi_req"); // not in greenlet if (!py_wsgi_req) { - PyErr_Clear(); - return NULL; + uwsgi_log("[BUG] current_wsgi_req NOT FOUND !!!\n"); + exit(1); } struct wsgi_request *wsgi_req = (struct wsgi_request*) PyLong_AsLong(py_wsgi_req); Py_DECREF(py_wsgi_req); diff --git a/plugins/psgi/psgi.h b/plugins/psgi/psgi.h index 0ffdbd7e..873bfaba 100644 --- a/plugins/psgi/psgi.h +++ b/plugins/psgi/psgi.h @@ -45,6 +45,8 @@ struct uwsgi_perl { SV *postfork; SV *atexit; + + int loaded; }; void init_perl_embedded_module(void); diff --git a/plugins/psgi/psgi_loader.c b/plugins/psgi/psgi_loader.c index 1f19165e..9afdfbc2 100644 --- a/plugins/psgi/psgi_loader.c +++ b/plugins/psgi/psgi_loader.c @@ -437,6 +437,8 @@ int init_psgi_app(struct wsgi_request *wsgi_req, char *app, uint16_t app_len, Pe PERL_SET_CONTEXT(uperl.main[0]); } + uperl.loaded = 1; + return id; clear: diff --git a/plugins/psgi/uwsgi_plmodule.c b/plugins/psgi/uwsgi_plmodule.c index 5f77656c..3e7969a7 100644 --- a/plugins/psgi/uwsgi_plmodule.c +++ b/plugins/psgi/uwsgi_plmodule.c @@ -300,6 +300,72 @@ XS(XS_i_am_the_lord) { #endif +XS(XS_websocket_handshake) { + + dXSARGS; + + char *key = NULL; + STRLEN key_len = 0; + + char *origin = NULL; + STRLEN origin_len = 0; + + psgi_check_args(1); + + key = SvPV(ST(0), key_len); + + if (items > 1) { + origin = SvPV(ST(0), origin_len); + } + struct wsgi_request *wsgi_req = current_wsgi_req(); + + if (uwsgi_websocket_handshake(wsgi_req, key, key_len, origin, origin_len)) { + croak("unable to complete websocket handshake"); + } + + XSRETURN_UNDEF; +} + +XS(XS_websocket_send) { + dXSARGS; + + char *message = NULL; + STRLEN message_len = 0; + + psgi_check_args(1); + + message = SvPV(ST(0), message_len); + + struct wsgi_request *wsgi_req = current_wsgi_req(); + + if (uwsgi_websocket_send(wsgi_req, message, message_len)) { + croak("unable to send websocket message"); + } + + XSRETURN_UNDEF; +} + +XS(XS_websocket_recv) { + dXSARGS; + + psgi_check_args(0); + + struct wsgi_request *wsgi_req = current_wsgi_req(); + struct uwsgi_buffer *ub = uwsgi_websocket_recv(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() { psgi_xs(reload); psgi_xs(cache_set); @@ -317,6 +383,9 @@ void init_perl_embedded_module() { #ifdef UWSGI_SSL psgi_xs(i_am_the_lord); #endif + psgi_xs(websocket_handshake); + psgi_xs(websocket_recv); + psgi_xs(websocket_send); psgi_xs(postfork); psgi_xs(atexit); } diff --git a/tests/websockets_echo.pl b/tests/websockets_echo.pl new file mode 100644 index 00000000..352cad67 --- /dev/null +++ b/tests/websockets_echo.pl @@ -0,0 +1,63 @@ +#!./uwsgi --https :8443,foobar.crt,foobar.key --http-raw-body --coroae 100 --psgi tests/websocket_echo.pl + +my $app = sub { + + my $env = shift; + + my $ws_scheme = 'ws'; + if (exists($env->{HTTPS}) || $env['psgi.url_scheme'] eq 'https') { + $ws_scheme = 'wss'; + } + + if ($env->{PATH_INFO} eq '/') { +my $body = < + + + + +

WebSocket

+ + +
+
+ + + +EOF + return ['200', ['Content-Type' => 'text/html'], [$body]]; + } + elsif ($env->{PATH_INFO} eq '/foobar/') { + uwsgi::websocket_handshake($env->{HTTP_SEC_WEBSOCKET_KEY}, $env->{HTTP_ORIGIN}); + print "websockets...\n"; + while(1) { + my $msg = uwsgi::websocket_recv; + uwsgi::websocket_send('['.time().'] '.$msg); + } + } + +}