280 lines
9.4 KiB
JavaScript
280 lines
9.4 KiB
JavaScript
|
"use strict";
|
||
|
|
Object.defineProperty(exports, "__esModule", { value: true });
|
||
|
|
exports.uServer = void 0;
|
||
|
|
const debug_1 = require("debug");
|
||
|
|
const server_1 = require("./server");
|
||
|
|
const transports_uws_1 = require("./transports-uws");
|
||
|
|
const debug = (0, debug_1.default)("engine:uws");
|
||
|
|
/**
|
||
|
|
* An Engine.IO server based on the `uWebSockets.js` package.
|
||
|
|
*/
|
||
|
|
// TODO export it into its own package
|
||
|
|
class uServer extends server_1.BaseServer {
|
||
|
|
init() { }
|
||
|
|
cleanup() { }
|
||
|
|
/**
|
||
|
|
* Prepares a request by processing the query string.
|
||
|
|
*
|
||
|
|
* @private
|
||
|
|
*/
|
||
|
|
prepare(req, res) {
|
||
|
|
req.method = req.getMethod().toUpperCase();
|
||
|
|
req.url = req.getUrl();
|
||
|
|
const params = new URLSearchParams(req.getQuery());
|
||
|
|
req._query = Object.fromEntries(params.entries());
|
||
|
|
req.headers = {};
|
||
|
|
req.forEach((key, value) => {
|
||
|
|
req.headers[key] = value;
|
||
|
|
});
|
||
|
|
req.connection = {
|
||
|
|
remoteAddress: Buffer.from(res.getRemoteAddressAsText()).toString(),
|
||
|
|
};
|
||
|
|
res.onAborted(() => {
|
||
|
|
debug("response has been aborted");
|
||
|
|
});
|
||
|
|
}
|
||
|
|
createTransport(transportName, req) {
|
||
|
|
return new transports_uws_1.default[transportName](req);
|
||
|
|
}
|
||
|
|
/**
|
||
|
|
* Attach the engine to a µWebSockets.js server
|
||
|
|
* @param app
|
||
|
|
* @param options
|
||
|
|
*/
|
||
|
|
attach(app /* : TemplatedApp */, options = {}) {
|
||
|
|
const path = this._computePath(options);
|
||
|
|
app
|
||
|
|
.any(path, this.handleRequest.bind(this))
|
||
|
|
//
|
||
|
|
.ws(path, {
|
||
|
|
compression: options.compression,
|
||
|
|
idleTimeout: options.idleTimeout,
|
||
|
|
maxBackpressure: options.maxBackpressure,
|
||
|
|
maxPayloadLength: this.opts.maxHttpBufferSize,
|
||
|
|
upgrade: this.handleUpgrade.bind(this),
|
||
|
|
open: (ws) => {
|
||
|
|
const transport = ws.getUserData().transport;
|
||
|
|
transport.socket = ws;
|
||
|
|
transport.writable = true;
|
||
|
|
transport.emit("ready");
|
||
|
|
},
|
||
|
|
message: (ws, message, isBinary) => {
|
||
|
|
ws.getUserData().transport.onData(isBinary ? message : Buffer.from(message).toString());
|
||
|
|
},
|
||
|
|
close: (ws, code, message) => {
|
||
|
|
ws.getUserData().transport.onClose(code, message);
|
||
|
|
},
|
||
|
|
});
|
||
|
|
}
|
||
|
|
_applyMiddlewares(req, res, callback) {
|
||
|
|
if (this.middlewares.length === 0) {
|
||
|
|
return callback();
|
||
|
|
}
|
||
|
|
// needed to buffer headers until the status is computed
|
||
|
|
req.res = new ResponseWrapper(res);
|
||
|
|
super._applyMiddlewares(req, req.res, (err) => {
|
||
|
|
// some middlewares (like express-session) wait for the writeHead() call to flush their headers
|
||
|
|
// see https://github.com/expressjs/session/blob/1010fadc2f071ddf2add94235d72224cf65159c6/index.js#L220-L244
|
||
|
|
req.res.writeHead();
|
||
|
|
callback(err);
|
||
|
|
});
|
||
|
|
}
|
||
|
|
handleRequest(res, req) {
|
||
|
|
debug('handling "%s" http request "%s"', req.getMethod(), req.getUrl());
|
||
|
|
this.prepare(req, res);
|
||
|
|
req.res = res;
|
||
|
|
const callback = (errorCode, errorContext) => {
|
||
|
|
if (errorCode !== undefined) {
|
||
|
|
this.emit("connection_error", {
|
||
|
|
req,
|
||
|
|
code: errorCode,
|
||
|
|
message: server_1.Server.errorMessages[errorCode],
|
||
|
|
context: errorContext,
|
||
|
|
});
|
||
|
|
this.abortRequest(req.res, errorCode, errorContext);
|
||
|
|
return;
|
||
|
|
}
|
||
|
|
if (req._query.sid) {
|
||
|
|
debug("setting new request for existing client");
|
||
|
|
// @ts-ignore
|
||
|
|
this.clients[req._query.sid].transport.onRequest(req);
|
||
|
|
}
|
||
|
|
else {
|
||
|
|
const closeConnection = (errorCode, errorContext) => this.abortRequest(res, errorCode, errorContext);
|
||
|
|
this.handshake(req._query.transport, req, closeConnection);
|
||
|
|
}
|
||
|
|
};
|
||
|
|
this._applyMiddlewares(req, res, (err) => {
|
||
|
|
if (err) {
|
||
|
|
callback(server_1.Server.errors.BAD_REQUEST, { name: "MIDDLEWARE_FAILURE" });
|
||
|
|
}
|
||
|
|
else {
|
||
|
|
this.verify(req, false, callback);
|
||
|
|
}
|
||
|
|
});
|
||
|
|
}
|
||
|
|
handleUpgrade(res, req, context) {
|
||
|
|
debug("on upgrade");
|
||
|
|
this.prepare(req, res);
|
||
|
|
req.res = res;
|
||
|
|
const callback = async (errorCode, errorContext) => {
|
||
|
|
if (errorCode !== undefined) {
|
||
|
|
this.emit("connection_error", {
|
||
|
|
req,
|
||
|
|
code: errorCode,
|
||
|
|
message: server_1.Server.errorMessages[errorCode],
|
||
|
|
context: errorContext,
|
||
|
|
});
|
||
|
|
this.abortRequest(res, errorCode, errorContext);
|
||
|
|
return;
|
||
|
|
}
|
||
|
|
const id = req._query.sid;
|
||
|
|
let transport;
|
||
|
|
if (id) {
|
||
|
|
const client = this.clients[id];
|
||
|
|
if (!client) {
|
||
|
|
debug("upgrade attempt for closed client");
|
||
|
|
return res.close();
|
||
|
|
}
|
||
|
|
else if (client.upgrading) {
|
||
|
|
debug("transport has already been trying to upgrade");
|
||
|
|
return res.close();
|
||
|
|
}
|
||
|
|
else if (client.upgraded) {
|
||
|
|
debug("transport had already been upgraded");
|
||
|
|
return res.close();
|
||
|
|
}
|
||
|
|
else {
|
||
|
|
debug("upgrading existing transport");
|
||
|
|
transport = this.createTransport(req._query.transport, req);
|
||
|
|
client._maybeUpgrade(transport);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
else {
|
||
|
|
transport = await this.handshake(req._query.transport, req, (errorCode, errorContext) => this.abortRequest(res, errorCode, errorContext));
|
||
|
|
if (!transport) {
|
||
|
|
return;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
// calling writeStatus() triggers the flushing of any header added in a middleware
|
||
|
|
req.res.writeStatus("101 Switching Protocols");
|
||
|
|
res.upgrade({
|
||
|
|
transport,
|
||
|
|
}, req.getHeader("sec-websocket-key"), req.getHeader("sec-websocket-protocol"), req.getHeader("sec-websocket-extensions"), context);
|
||
|
|
};
|
||
|
|
this._applyMiddlewares(req, res, (err) => {
|
||
|
|
if (err) {
|
||
|
|
callback(server_1.Server.errors.BAD_REQUEST, { name: "MIDDLEWARE_FAILURE" });
|
||
|
|
}
|
||
|
|
else {
|
||
|
|
this.verify(req, true, callback);
|
||
|
|
}
|
||
|
|
});
|
||
|
|
}
|
||
|
|
abortRequest(res, errorCode, errorContext) {
|
||
|
|
const statusCode = errorCode === server_1.Server.errors.FORBIDDEN
|
||
|
|
? "403 Forbidden"
|
||
|
|
: "400 Bad Request";
|
||
|
|
const message = errorContext && errorContext.message
|
||
|
|
? errorContext.message
|
||
|
|
: server_1.Server.errorMessages[errorCode];
|
||
|
|
res.writeStatus(statusCode);
|
||
|
|
res.writeHeader("Content-Type", "application/json");
|
||
|
|
res.end(JSON.stringify({
|
||
|
|
code: errorCode,
|
||
|
|
message,
|
||
|
|
}));
|
||
|
|
}
|
||
|
|
}
|
||
|
|
exports.uServer = uServer;
|
||
|
|
class ResponseWrapper {
|
||
|
|
constructor(res) {
|
||
|
|
this.res = res;
|
||
|
|
this.statusWritten = false;
|
||
|
|
this.headers = [];
|
||
|
|
this.isAborted = false;
|
||
|
|
}
|
||
|
|
set statusCode(status) {
|
||
|
|
if (!status) {
|
||
|
|
return;
|
||
|
|
}
|
||
|
|
// FIXME: handle all status codes?
|
||
|
|
this.writeStatus(status === 200 ? "200 OK" : "204 No Content");
|
||
|
|
}
|
||
|
|
writeHead(status) {
|
||
|
|
this.statusCode = status;
|
||
|
|
}
|
||
|
|
setHeader(key, value) {
|
||
|
|
if (Array.isArray(value)) {
|
||
|
|
value.forEach((val) => {
|
||
|
|
this.writeHeader(key, val);
|
||
|
|
});
|
||
|
|
}
|
||
|
|
else {
|
||
|
|
this.writeHeader(key, value);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
removeHeader() {
|
||
|
|
// FIXME: not implemented
|
||
|
|
}
|
||
|
|
// needed by vary: https://github.com/jshttp/vary/blob/5d725d059b3871025cf753e9dfa08924d0bcfa8f/index.js#L134
|
||
|
|
getHeader() { }
|
||
|
|
writeStatus(status) {
|
||
|
|
if (this.isAborted)
|
||
|
|
return;
|
||
|
|
this.res.writeStatus(status);
|
||
|
|
this.statusWritten = true;
|
||
|
|
this.writeBufferedHeaders();
|
||
|
|
return this;
|
||
|
|
}
|
||
|
|
writeHeader(key, value) {
|
||
|
|
if (this.isAborted)
|
||
|
|
return;
|
||
|
|
if (key === "Content-Length") {
|
||
|
|
// the content length is automatically added by uWebSockets.js
|
||
|
|
return;
|
||
|
|
}
|
||
|
|
if (this.statusWritten) {
|
||
|
|
this.res.writeHeader(key, value);
|
||
|
|
}
|
||
|
|
else {
|
||
|
|
this.headers.push([key, value]);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
writeBufferedHeaders() {
|
||
|
|
this.headers.forEach(([key, value]) => {
|
||
|
|
this.res.writeHeader(key, value);
|
||
|
|
});
|
||
|
|
}
|
||
|
|
end(data) {
|
||
|
|
if (this.isAborted)
|
||
|
|
return;
|
||
|
|
this.res.cork(() => {
|
||
|
|
if (!this.statusWritten) {
|
||
|
|
// status will be inferred as "200 OK"
|
||
|
|
this.writeBufferedHeaders();
|
||
|
|
}
|
||
|
|
this.res.end(data);
|
||
|
|
});
|
||
|
|
}
|
||
|
|
onData(fn) {
|
||
|
|
if (this.isAborted)
|
||
|
|
return;
|
||
|
|
this.res.onData(fn);
|
||
|
|
}
|
||
|
|
onAborted(fn) {
|
||
|
|
if (this.isAborted)
|
||
|
|
return;
|
||
|
|
this.res.onAborted(() => {
|
||
|
|
// Any attempt to use the UWS response object after abort will throw!
|
||
|
|
this.isAborted = true;
|
||
|
|
fn();
|
||
|
|
});
|
||
|
|
}
|
||
|
|
cork(fn) {
|
||
|
|
if (this.isAborted)
|
||
|
|
return;
|
||
|
|
this.res.cork(fn);
|
||
|
|
}
|
||
|
|
}
|