diff options
Diffstat (limited to '')
| -rw-r--r-- | src/nodejs/unit-http/http_server.js | 50 | ||||
| -rw-r--r-- | src/nodejs/unit-http/unit.cpp | 75 | ||||
| -rw-r--r-- | src/nodejs/unit-http/unit.h | 6 |
3 files changed, 87 insertions, 44 deletions
diff --git a/src/nodejs/unit-http/http_server.js b/src/nodejs/unit-http/http_server.js index d378e410..e59296ae 100644 --- a/src/nodejs/unit-http/http_server.js +++ b/src/nodejs/unit-http/http_server.js @@ -11,6 +11,7 @@ const util = require('util'); const unit_lib = require('./build/Release/unit-http'); const Socket = require('./socket'); const WebSocketFrame = require('./websocket_frame'); +const Readable = require('stream').Readable; function ServerResponse(req) { @@ -23,6 +24,7 @@ function ServerResponse(req) { req._response = this; this.socket = req.socket; this.connection = req.connection; + this.writable = true; } util.inherits(ServerResponse, EventEmitter); @@ -268,6 +270,7 @@ ServerResponse.prototype._writeBody = function(chunk, encoding, callback) { res = this._write(chunk, 0, contentLength); if (res < contentLength) { this.socket.writable = false; + this.writable = false; o = new BufferedOutput(this, res, chunk, encoding, callback); this.server._output.push(o); @@ -328,6 +331,8 @@ ServerResponse.prototype.end = function end(chunk, encoding, callback) { if (typeof callback === 'function') { callback(); } + + this.emit("finish"); }); this.finished = true; @@ -337,15 +342,14 @@ ServerResponse.prototype.end = function end(chunk, encoding, callback) { }; function ServerRequest(server, socket) { - EventEmitter.call(this); + Readable.call(this); this.server = server; this.socket = socket; this.connection = socket; + this._pushed_eofchunk = false; } -util.inherits(ServerRequest, EventEmitter); - -ServerRequest.prototype.unpipe = undefined; +util.inherits(ServerRequest, Readable); ServerRequest.prototype.setTimeout = function setTimeout(msecs, callback) { this.timeout = msecs; @@ -377,35 +381,21 @@ ServerRequest.prototype.STATUS_CODES = function STATUS_CODES() { return http.STATUS_CODES; }; -ServerRequest.prototype.listeners = function listeners() { - return []; -}; - -ServerRequest.prototype.resume = function resume() { - return []; -}; +ServerRequest.prototype._request_read = unit_lib.request_read; -/* - * The "on" method is overridden to defer reading data until user code is - * ready, that is (ev === "data"). This can occur after req.emit("end") is - * executed, since the user code can be scheduled asynchronously by Promises - * and so on. Passing the data is postponed by process.nextTick() until - * the "on" method caller completes. - */ -ServerRequest.prototype.on = function on(ev, fn) { - Server.prototype.on.call(this, ev, fn); +ServerRequest.prototype._read = function _read(n) { + const b = this._request_read(n); - if (ev === "data") { - process.nextTick(function () { - if (this._data.length !== 0) { - this.emit("data", this._data); - } + if (b != null) { + this.push(b); + } - }.bind(this)); + if (!this._pushed_eofchunk && (b == null || b.length < n)) { + this._pushed_eofchunk = true; + this.push(null); } }; -ServerRequest.prototype.addListener = ServerRequest.prototype.on; function Server(requestListener) { EventEmitter.call(this); @@ -472,11 +462,6 @@ Server.prototype.emit_request = function (req, res) { } else { this.emit("request", req, res); } - - process.nextTick(() => { - req.emit("finish"); - req.emit("end"); - }); }; Server.prototype.emit_close = function () { @@ -523,6 +508,7 @@ Server.prototype.emit_drain = function () { } resp.socket.writable = true; + resp.writable = true; process.nextTick(() => { resp.emit("drain"); diff --git a/src/nodejs/unit-http/unit.cpp b/src/nodejs/unit-http/unit.cpp index 0e049865..67b4377d 100644 --- a/src/nodejs/unit-http/unit.cpp +++ b/src/nodejs/unit-http/unit.cpp @@ -27,6 +27,7 @@ struct port_data_t { struct req_data_t { napi_ref sock_ref; + napi_ref req_ref; napi_ref resp_ref; napi_ref conn_ref; }; @@ -65,6 +66,7 @@ Unit::init(napi_env env, napi_value exports) constructor_ = napi.create_reference(ctor); napi.set_named_property(exports, "Unit", ctor); + napi.set_named_property(exports, "request_read", request_read); napi.set_named_property(exports, "response_send_headers", response_send_headers); napi.set_named_property(exports, "response_write", response_write); @@ -206,7 +208,7 @@ Unit::request_handler(nxt_unit_request_info_t *req) server_obj = get_server_object(); socket = create_socket(server_obj, req); - request = create_request(server_obj, socket); + request = create_request(server_obj, socket, req); response = create_response(server_obj, request, req); create_headers(req, request); @@ -301,6 +303,7 @@ Unit::close_handler(nxt_unit_request_info_t *req) nxt_napi::create(0)); remove_wrap(req_data->sock_ref); + remove_wrap(req_data->req_ref); remove_wrap(req_data->resp_ref); remove_wrap(req_data->conn_ref); @@ -488,9 +491,8 @@ Unit::get_server_object() void Unit::create_headers(nxt_unit_request_info_t *req, napi_value request) { - void *data; uint32_t i; - napi_value headers, raw_headers, buffer; + napi_value headers, raw_headers; napi_status status; nxt_unit_request_t *r; @@ -515,11 +517,6 @@ Unit::create_headers(nxt_unit_request_info_t *req, napi_value request) set_named_property(request, "url", r->target, r->target_length); set_named_property(request, "_websocket_handshake", r->websocket_handshake); - - buffer = create_buffer((size_t) req->content_length, &data); - nxt_unit_request_read(req, data, req->content_length); - - set_named_property(request, "_data", buffer); } @@ -577,13 +574,20 @@ Unit::create_socket(napi_value server_obj, nxt_unit_request_info_t *req) napi_value -Unit::create_request(napi_value server_obj, napi_value socket) +Unit::create_request(napi_value server_obj, napi_value socket, + nxt_unit_request_info_t *req) { - napi_value constructor; + napi_value constructor, res; + req_data_t *req_data; constructor = get_named_property(server_obj, "ServerRequest"); - return new_instance(constructor, server_obj, socket); + res = new_instance(constructor, server_obj, socket); + + req_data = (req_data_t *) req->data; + req_data->req_ref = wrap(res, req, req_destroy); + + return res; } @@ -643,6 +647,47 @@ Unit::create_websocket_frame(napi_value server_obj, napi_value +Unit::request_read(napi_env env, napi_callback_info info) +{ + void *data; + uint32_t wm; + nxt_napi napi(env); + napi_value this_arg, argv, buffer; + nxt_unit_request_info_t *req; + + try { + this_arg = napi.get_cb_info(info, argv); + + try { + req = napi.get_request_info(this_arg); + + } catch (exception &e) { + return nullptr; + } + + if (req->content_length == 0) { + return nullptr; + } + + wm = napi.get_value_uint32(argv); + + if (wm > req->content_length) { + wm = req->content_length; + } + + buffer = napi.create_buffer((size_t) wm, &data); + nxt_unit_request_read(req, data, wm); + + } catch (exception &e) { + napi.throw_error(e); + return nullptr; + } + + return buffer; +} + + +napi_value Unit::response_send_headers(napi_env env, napi_callback_info info) { int ret; @@ -884,6 +929,7 @@ Unit::response_end(napi_env env, napi_callback_info info) req_data = (req_data_t *) req->data; napi.remove_wrap(req_data->sock_ref); + napi.remove_wrap(req_data->req_ref); napi.remove_wrap(req_data->resp_ref); napi.remove_wrap(req_data->conn_ref); @@ -1012,6 +1058,13 @@ Unit::sock_destroy(napi_env env, void *r, void *finalize_hint) void +Unit::req_destroy(napi_env env, void *r, void *finalize_hint) +{ + nxt_unit_req_debug(NULL, "req_destroy: %p", r); +} + + +void Unit::resp_destroy(napi_env env, void *r, void *finalize_hint) { nxt_unit_req_debug(NULL, "resp_destroy: %p", r); diff --git a/src/nodejs/unit-http/unit.h b/src/nodejs/unit-http/unit.h index 07823c26..4ef40d45 100644 --- a/src/nodejs/unit-http/unit.h +++ b/src/nodejs/unit-http/unit.h @@ -21,6 +21,7 @@ private: static void destroy(napi_env env, void *nativeObject, void *finalize_hint); static void conn_destroy(napi_env env, void *nativeObject, void *finalize_hint); static void sock_destroy(napi_env env, void *nativeObject, void *finalize_hint); + static void req_destroy(napi_env env, void *nativeObject, void *finalize_hint); static void resp_destroy(napi_env env, void *nativeObject, void *finalize_hint); static napi_value create_server(napi_env env, napi_callback_info info); @@ -50,7 +51,8 @@ private: napi_value create_socket(napi_value server_obj, nxt_unit_request_info_t *req); - napi_value create_request(napi_value server_obj, napi_value socket); + napi_value create_request(napi_value server_obj, napi_value socket, + nxt_unit_request_info_t *req); napi_value create_response(napi_value server_obj, napi_value request, nxt_unit_request_info_t *req); @@ -58,6 +60,8 @@ private: napi_value create_websocket_frame(napi_value server_obj, nxt_unit_websocket_frame_t *ws); + static napi_value request_read(napi_env env, napi_callback_info info); + static napi_value response_send_headers(napi_env env, napi_callback_info info); |