You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 

419 lines
12 KiB

import { Transport } from "../transport.js";
import debugModule from "debug"; // debug()
import { yeast } from "../contrib/yeast.js";
import { encode } from "../contrib/parseqs.js";
import { encodePayload, decodePayload } from "engine.io-parser";
import { XHR as XMLHttpRequest } from "./xmlhttprequest.js";
import { Emitter } from "@socket.io/component-emitter";
import { installTimerFunctions, pick } from "../util.js";
import { globalThisShim as globalThis } from "../globalThis.js";
const debug = debugModule("engine.io-client:polling"); // debug()
function empty() { }
const hasXHR2 = (function () {
const xhr = new XMLHttpRequest({
xdomain: false
});
return null != xhr.responseType;
})();
export class Polling extends Transport {
/**
* XHR Polling constructor.
*
* @param {Object} opts
* @api public
*/
constructor(opts) {
super(opts);
this.polling = false;
if (typeof location !== "undefined") {
const isSSL = "https:" === location.protocol;
let port = location.port;
// some user agents have empty `location.port`
if (!port) {
port = isSSL ? "443" : "80";
}
this.xd =
(typeof location !== "undefined" &&
opts.hostname !== location.hostname) ||
port !== opts.port;
this.xs = opts.secure !== isSSL;
}
/**
* XHR supports binary
*/
const forceBase64 = opts && opts.forceBase64;
this.supportsBinary = hasXHR2 && !forceBase64;
}
/**
* Transport name.
*/
get name() {
return "polling";
}
/**
* Opens the socket (triggers polling). We write a PING message to determine
* when the transport is open.
*
* @api private
*/
doOpen() {
this.poll();
}
/**
* Pauses polling.
*
* @param {Function} callback upon buffers are flushed and transport is paused
* @api private
*/
pause(onPause) {
this.readyState = "pausing";
const pause = () => {
debug("paused");
this.readyState = "paused";
onPause();
};
if (this.polling || !this.writable) {
let total = 0;
if (this.polling) {
debug("we are currently polling - waiting to pause");
total++;
this.once("pollComplete", function () {
debug("pre-pause polling complete");
--total || pause();
});
}
if (!this.writable) {
debug("we are currently writing - waiting to pause");
total++;
this.once("drain", function () {
debug("pre-pause writing complete");
--total || pause();
});
}
}
else {
pause();
}
}
/**
* Starts polling cycle.
*
* @api public
*/
poll() {
debug("polling");
this.polling = true;
this.doPoll();
this.emitReserved("poll");
}
/**
* Overloads onData to detect payloads.
*
* @api private
*/
onData(data) {
debug("polling got data %s", data);
const callback = packet => {
// if its the first message we consider the transport open
if ("opening" === this.readyState && packet.type === "open") {
this.onOpen();
}
// if its a close packet, we close the ongoing requests
if ("close" === packet.type) {
this.onClose({ description: "transport closed by the server" });
return false;
}
// otherwise bypass onData and handle the message
this.onPacket(packet);
};
// decode payload
decodePayload(data, this.socket.binaryType).forEach(callback);
// if an event did not trigger closing
if ("closed" !== this.readyState) {
// if we got data we're not polling
this.polling = false;
this.emitReserved("pollComplete");
if ("open" === this.readyState) {
this.poll();
}
else {
debug('ignoring poll - transport state "%s"', this.readyState);
}
}
}
/**
* For polling, send a close packet.
*
* @api private
*/
doClose() {
const close = () => {
debug("writing close packet");
this.write([{ type: "close" }]);
};
if ("open" === this.readyState) {
debug("transport open - closing");
close();
}
else {
// in case we're trying to close while
// handshaking is in progress (GH-164)
debug("transport not open - deferring close");
this.once("open", close);
}
}
/**
* Writes a packets payload.
*
* @param {Array} data packets
* @param {Function} drain callback
* @api private
*/
write(packets) {
this.writable = false;
encodePayload(packets, data => {
this.doWrite(data, () => {
this.writable = true;
this.emitReserved("drain");
});
});
}
/**
* Generates uri for connection.
*
* @api private
*/
uri() {
let query = this.query || {};
const schema = this.opts.secure ? "https" : "http";
let port = "";
// cache busting is forced
if (false !== this.opts.timestampRequests) {
query[this.opts.timestampParam] = yeast();
}
if (!this.supportsBinary && !query.sid) {
query.b64 = 1;
}
// avoid port if default for schema
if (this.opts.port &&
(("https" === schema && Number(this.opts.port) !== 443) ||
("http" === schema && Number(this.opts.port) !== 80))) {
port = ":" + this.opts.port;
}
const encodedQuery = encode(query);
const ipv6 = this.opts.hostname.indexOf(":") !== -1;
return (schema +
"://" +
(ipv6 ? "[" + this.opts.hostname + "]" : this.opts.hostname) +
port +
this.opts.path +
(encodedQuery.length ? "?" + encodedQuery : ""));
}
/**
* Creates a request.
*
* @param {String} method
* @api private
*/
request(opts = {}) {
Object.assign(opts, { xd: this.xd, xs: this.xs }, this.opts);
return new Request(this.uri(), opts);
}
/**
* Sends data.
*
* @param {String} data to send.
* @param {Function} called upon flush.
* @api private
*/
doWrite(data, fn) {
const req = this.request({
method: "POST",
data: data
});
req.on("success", fn);
req.on("error", (xhrStatus, context) => {
this.onError("xhr post error", xhrStatus, context);
});
}
/**
* Starts a poll cycle.
*
* @api private
*/
doPoll() {
debug("xhr poll");
const req = this.request();
req.on("data", this.onData.bind(this));
req.on("error", (xhrStatus, context) => {
this.onError("xhr poll error", xhrStatus, context);
});
this.pollXhr = req;
}
}
export class Request extends Emitter {
/**
* Request constructor
*
* @param {Object} options
* @api public
*/
constructor(uri, opts) {
super();
installTimerFunctions(this, opts);
this.opts = opts;
this.method = opts.method || "GET";
this.uri = uri;
this.async = false !== opts.async;
this.data = undefined !== opts.data ? opts.data : null;
this.create();
}
/**
* Creates the XHR object and sends the request.
*
* @api private
*/
create() {
const opts = pick(this.opts, "agent", "pfx", "key", "passphrase", "cert", "ca", "ciphers", "rejectUnauthorized", "autoUnref");
opts.xdomain = !!this.opts.xd;
opts.xscheme = !!this.opts.xs;
const xhr = (this.xhr = new XMLHttpRequest(opts));
try {
debug("xhr open %s: %s", this.method, this.uri);
xhr.open(this.method, this.uri, this.async);
try {
if (this.opts.extraHeaders) {
xhr.setDisableHeaderCheck && xhr.setDisableHeaderCheck(true);
for (let i in this.opts.extraHeaders) {
if (this.opts.extraHeaders.hasOwnProperty(i)) {
xhr.setRequestHeader(i, this.opts.extraHeaders[i]);
}
}
}
}
catch (e) { }
if ("POST" === this.method) {
try {
xhr.setRequestHeader("Content-type", "text/plain;charset=UTF-8");
}
catch (e) { }
}
try {
xhr.setRequestHeader("Accept", "*/*");
}
catch (e) { }
// ie6 check
if ("withCredentials" in xhr) {
xhr.withCredentials = this.opts.withCredentials;
}
if (this.opts.requestTimeout) {
xhr.timeout = this.opts.requestTimeout;
}
xhr.onreadystatechange = () => {
if (4 !== xhr.readyState)
return;
if (200 === xhr.status || 1223 === xhr.status) {
this.onLoad();
}
else {
// make sure the `error` event handler that's user-set
// does not throw in the same tick and gets caught here
this.setTimeoutFn(() => {
this.onError(typeof xhr.status === "number" ? xhr.status : 0);
}, 0);
}
};
debug("xhr data %s", this.data);
xhr.send(this.data);
}
catch (e) {
// Need to defer since .create() is called directly from the constructor
// and thus the 'error' event can only be only bound *after* this exception
// occurs. Therefore, also, we cannot throw here at all.
this.setTimeoutFn(() => {
this.onError(e);
}, 0);
return;
}
if (typeof document !== "undefined") {
this.index = Request.requestsCount++;
Request.requests[this.index] = this;
}
}
/**
* Called upon error.
*
* @api private
*/
onError(err) {
this.emitReserved("error", err, this.xhr);
this.cleanup(true);
}
/**
* Cleans up house.
*
* @api private
*/
cleanup(fromError) {
if ("undefined" === typeof this.xhr || null === this.xhr) {
return;
}
this.xhr.onreadystatechange = empty;
if (fromError) {
try {
this.xhr.abort();
}
catch (e) { }
}
if (typeof document !== "undefined") {
delete Request.requests[this.index];
}
this.xhr = null;
}
/**
* Called upon load.
*
* @api private
*/
onLoad() {
const data = this.xhr.responseText;
if (data !== null) {
this.emitReserved("data", data);
this.emitReserved("success");
this.cleanup();
}
}
/**
* Aborts the request.
*
* @api public
*/
abort() {
this.cleanup();
}
}
Request.requestsCount = 0;
Request.requests = {};
/**
* Aborts pending requests when unloading the window. This is needed to prevent
* memory leaks (e.g. when using IE) and to ensure that no spurious error is
* emitted.
*/
if (typeof document !== "undefined") {
// @ts-ignore
if (typeof attachEvent === "function") {
// @ts-ignore
attachEvent("onunload", unloadHandler);
}
else if (typeof addEventListener === "function") {
const terminationEvent = "onpagehide" in globalThis ? "pagehide" : "unload";
addEventListener(terminationEvent, unloadHandler, false);
}
}
function unloadHandler() {
for (let i in Request.requests) {
if (Request.requests.hasOwnProperty(i)) {
Request.requests[i].abort();
}
}
}