mirror of
https://github.com/rails/rails.git
synced 2022-11-09 12:12:34 -05:00
6d7c12274e
A SubscriptionGuarantor maintains a set of pending subscriptions, resending the subscribe command unless and until the subscription is confirmed or rejected by the server or cancelled client-side. A race condition in the ActionCable server - where an unsubscribe is sent, followed rapidly by a subscribe, but handled in the reverse order - necessitates this enhancement. Indeed, the subscriptions created and torn down by Turbo Streams amplifies the existence of this race condition.
489 lines
15 KiB
JavaScript
489 lines
15 KiB
JavaScript
(function(global, factory) {
|
|
typeof exports === "object" && typeof module !== "undefined" ? factory(exports) : typeof define === "function" && define.amd ? define([ "exports" ], factory) : (global = typeof globalThis !== "undefined" ? globalThis : global || self,
|
|
factory(global.ActionCable = {}));
|
|
})(this, (function(exports) {
|
|
"use strict";
|
|
var adapters = {
|
|
logger: self.console,
|
|
WebSocket: self.WebSocket
|
|
};
|
|
var logger = {
|
|
log(...messages) {
|
|
if (this.enabled) {
|
|
messages.push(Date.now());
|
|
adapters.logger.log("[ActionCable]", ...messages);
|
|
}
|
|
}
|
|
};
|
|
const now = () => (new Date).getTime();
|
|
const secondsSince = time => (now() - time) / 1e3;
|
|
class ConnectionMonitor {
|
|
constructor(connection) {
|
|
this.visibilityDidChange = this.visibilityDidChange.bind(this);
|
|
this.connection = connection;
|
|
this.reconnectAttempts = 0;
|
|
}
|
|
start() {
|
|
if (!this.isRunning()) {
|
|
this.startedAt = now();
|
|
delete this.stoppedAt;
|
|
this.startPolling();
|
|
addEventListener("visibilitychange", this.visibilityDidChange);
|
|
logger.log(`ConnectionMonitor started. stale threshold = ${this.constructor.staleThreshold} s`);
|
|
}
|
|
}
|
|
stop() {
|
|
if (this.isRunning()) {
|
|
this.stoppedAt = now();
|
|
this.stopPolling();
|
|
removeEventListener("visibilitychange", this.visibilityDidChange);
|
|
logger.log("ConnectionMonitor stopped");
|
|
}
|
|
}
|
|
isRunning() {
|
|
return this.startedAt && !this.stoppedAt;
|
|
}
|
|
recordPing() {
|
|
this.pingedAt = now();
|
|
}
|
|
recordConnect() {
|
|
this.reconnectAttempts = 0;
|
|
this.recordPing();
|
|
delete this.disconnectedAt;
|
|
logger.log("ConnectionMonitor recorded connect");
|
|
}
|
|
recordDisconnect() {
|
|
this.disconnectedAt = now();
|
|
logger.log("ConnectionMonitor recorded disconnect");
|
|
}
|
|
startPolling() {
|
|
this.stopPolling();
|
|
this.poll();
|
|
}
|
|
stopPolling() {
|
|
clearTimeout(this.pollTimeout);
|
|
}
|
|
poll() {
|
|
this.pollTimeout = setTimeout((() => {
|
|
this.reconnectIfStale();
|
|
this.poll();
|
|
}), this.getPollInterval());
|
|
}
|
|
getPollInterval() {
|
|
const {staleThreshold: staleThreshold, reconnectionBackoffRate: reconnectionBackoffRate} = this.constructor;
|
|
const backoff = Math.pow(1 + reconnectionBackoffRate, Math.min(this.reconnectAttempts, 10));
|
|
const jitterMax = this.reconnectAttempts === 0 ? 1 : reconnectionBackoffRate;
|
|
const jitter = jitterMax * Math.random();
|
|
return staleThreshold * 1e3 * backoff * (1 + jitter);
|
|
}
|
|
reconnectIfStale() {
|
|
if (this.connectionIsStale()) {
|
|
logger.log(`ConnectionMonitor detected stale connection. reconnectAttempts = ${this.reconnectAttempts}, time stale = ${secondsSince(this.refreshedAt)} s, stale threshold = ${this.constructor.staleThreshold} s`);
|
|
this.reconnectAttempts++;
|
|
if (this.disconnectedRecently()) {
|
|
logger.log(`ConnectionMonitor skipping reopening recent disconnect. time disconnected = ${secondsSince(this.disconnectedAt)} s`);
|
|
} else {
|
|
logger.log("ConnectionMonitor reopening");
|
|
this.connection.reopen();
|
|
}
|
|
}
|
|
}
|
|
get refreshedAt() {
|
|
return this.pingedAt ? this.pingedAt : this.startedAt;
|
|
}
|
|
connectionIsStale() {
|
|
return secondsSince(this.refreshedAt) > this.constructor.staleThreshold;
|
|
}
|
|
disconnectedRecently() {
|
|
return this.disconnectedAt && secondsSince(this.disconnectedAt) < this.constructor.staleThreshold;
|
|
}
|
|
visibilityDidChange() {
|
|
if (document.visibilityState === "visible") {
|
|
setTimeout((() => {
|
|
if (this.connectionIsStale() || !this.connection.isOpen()) {
|
|
logger.log(`ConnectionMonitor reopening stale connection on visibilitychange. visibilityState = ${document.visibilityState}`);
|
|
this.connection.reopen();
|
|
}
|
|
}), 200);
|
|
}
|
|
}
|
|
}
|
|
ConnectionMonitor.staleThreshold = 6;
|
|
ConnectionMonitor.reconnectionBackoffRate = .15;
|
|
var INTERNAL = {
|
|
message_types: {
|
|
welcome: "welcome",
|
|
disconnect: "disconnect",
|
|
ping: "ping",
|
|
confirmation: "confirm_subscription",
|
|
rejection: "reject_subscription"
|
|
},
|
|
disconnect_reasons: {
|
|
unauthorized: "unauthorized",
|
|
invalid_request: "invalid_request",
|
|
server_restart: "server_restart"
|
|
},
|
|
default_mount_path: "/cable",
|
|
protocols: [ "actioncable-v1-json", "actioncable-unsupported" ]
|
|
};
|
|
const {message_types: message_types, protocols: protocols} = INTERNAL;
|
|
const supportedProtocols = protocols.slice(0, protocols.length - 1);
|
|
const indexOf = [].indexOf;
|
|
class Connection {
|
|
constructor(consumer) {
|
|
this.open = this.open.bind(this);
|
|
this.consumer = consumer;
|
|
this.subscriptions = this.consumer.subscriptions;
|
|
this.monitor = new ConnectionMonitor(this);
|
|
this.disconnected = true;
|
|
}
|
|
send(data) {
|
|
if (this.isOpen()) {
|
|
this.webSocket.send(JSON.stringify(data));
|
|
return true;
|
|
} else {
|
|
return false;
|
|
}
|
|
}
|
|
open() {
|
|
if (this.isActive()) {
|
|
logger.log(`Attempted to open WebSocket, but existing socket is ${this.getState()}`);
|
|
return false;
|
|
} else {
|
|
logger.log(`Opening WebSocket, current state is ${this.getState()}, subprotocols: ${protocols}`);
|
|
if (this.webSocket) {
|
|
this.uninstallEventHandlers();
|
|
}
|
|
this.webSocket = new adapters.WebSocket(this.consumer.url, protocols);
|
|
this.installEventHandlers();
|
|
this.monitor.start();
|
|
return true;
|
|
}
|
|
}
|
|
close({allowReconnect: allowReconnect} = {
|
|
allowReconnect: true
|
|
}) {
|
|
if (!allowReconnect) {
|
|
this.monitor.stop();
|
|
}
|
|
if (this.isActive()) {
|
|
return this.webSocket.close();
|
|
}
|
|
}
|
|
reopen() {
|
|
logger.log(`Reopening WebSocket, current state is ${this.getState()}`);
|
|
if (this.isActive()) {
|
|
try {
|
|
return this.close();
|
|
} catch (error) {
|
|
logger.log("Failed to reopen WebSocket", error);
|
|
} finally {
|
|
logger.log(`Reopening WebSocket in ${this.constructor.reopenDelay}ms`);
|
|
setTimeout(this.open, this.constructor.reopenDelay);
|
|
}
|
|
} else {
|
|
return this.open();
|
|
}
|
|
}
|
|
getProtocol() {
|
|
if (this.webSocket) {
|
|
return this.webSocket.protocol;
|
|
}
|
|
}
|
|
isOpen() {
|
|
return this.isState("open");
|
|
}
|
|
isActive() {
|
|
return this.isState("open", "connecting");
|
|
}
|
|
isProtocolSupported() {
|
|
return indexOf.call(supportedProtocols, this.getProtocol()) >= 0;
|
|
}
|
|
isState(...states) {
|
|
return indexOf.call(states, this.getState()) >= 0;
|
|
}
|
|
getState() {
|
|
if (this.webSocket) {
|
|
for (let state in adapters.WebSocket) {
|
|
if (adapters.WebSocket[state] === this.webSocket.readyState) {
|
|
return state.toLowerCase();
|
|
}
|
|
}
|
|
}
|
|
return null;
|
|
}
|
|
installEventHandlers() {
|
|
for (let eventName in this.events) {
|
|
const handler = this.events[eventName].bind(this);
|
|
this.webSocket[`on${eventName}`] = handler;
|
|
}
|
|
}
|
|
uninstallEventHandlers() {
|
|
for (let eventName in this.events) {
|
|
this.webSocket[`on${eventName}`] = function() {};
|
|
}
|
|
}
|
|
}
|
|
Connection.reopenDelay = 500;
|
|
Connection.prototype.events = {
|
|
message(event) {
|
|
if (!this.isProtocolSupported()) {
|
|
return;
|
|
}
|
|
const {identifier: identifier, message: message, reason: reason, reconnect: reconnect, type: type} = JSON.parse(event.data);
|
|
switch (type) {
|
|
case message_types.welcome:
|
|
this.monitor.recordConnect();
|
|
return this.subscriptions.reload();
|
|
|
|
case message_types.disconnect:
|
|
logger.log(`Disconnecting. Reason: ${reason}`);
|
|
return this.close({
|
|
allowReconnect: reconnect
|
|
});
|
|
|
|
case message_types.ping:
|
|
return this.monitor.recordPing();
|
|
|
|
case message_types.confirmation:
|
|
this.subscriptions.confirmSubscription(identifier);
|
|
return this.subscriptions.notify(identifier, "connected");
|
|
|
|
case message_types.rejection:
|
|
return this.subscriptions.reject(identifier);
|
|
|
|
default:
|
|
return this.subscriptions.notify(identifier, "received", message);
|
|
}
|
|
},
|
|
open() {
|
|
logger.log(`WebSocket onopen event, using '${this.getProtocol()}' subprotocol`);
|
|
this.disconnected = false;
|
|
if (!this.isProtocolSupported()) {
|
|
logger.log("Protocol is unsupported. Stopping monitor and disconnecting.");
|
|
return this.close({
|
|
allowReconnect: false
|
|
});
|
|
}
|
|
},
|
|
close(event) {
|
|
logger.log("WebSocket onclose event");
|
|
if (this.disconnected) {
|
|
return;
|
|
}
|
|
this.disconnected = true;
|
|
this.monitor.recordDisconnect();
|
|
return this.subscriptions.notifyAll("disconnected", {
|
|
willAttemptReconnect: this.monitor.isRunning()
|
|
});
|
|
},
|
|
error() {
|
|
logger.log("WebSocket onerror event");
|
|
}
|
|
};
|
|
const extend = function(object, properties) {
|
|
if (properties != null) {
|
|
for (let key in properties) {
|
|
const value = properties[key];
|
|
object[key] = value;
|
|
}
|
|
}
|
|
return object;
|
|
};
|
|
class Subscription {
|
|
constructor(consumer, params = {}, mixin) {
|
|
this.consumer = consumer;
|
|
this.identifier = JSON.stringify(params);
|
|
extend(this, mixin);
|
|
}
|
|
perform(action, data = {}) {
|
|
data.action = action;
|
|
return this.send(data);
|
|
}
|
|
send(data) {
|
|
return this.consumer.send({
|
|
command: "message",
|
|
identifier: this.identifier,
|
|
data: JSON.stringify(data)
|
|
});
|
|
}
|
|
unsubscribe() {
|
|
return this.consumer.subscriptions.remove(this);
|
|
}
|
|
}
|
|
class SubscriptionGuarantor {
|
|
constructor(subscriptions) {
|
|
this.subscriptions = subscriptions;
|
|
this.pendingSubscriptions = [];
|
|
}
|
|
guarantee(subscription) {
|
|
if (this.pendingSubscriptions.indexOf(subscription) == -1) {
|
|
logger.log(`SubscriptionGuarantor guaranteeing ${subscription.identifier}`);
|
|
this.pendingSubscriptions.push(subscription);
|
|
} else {
|
|
logger.log(`SubscriptionGuarantor already guaranteeing ${subscription.identifier}`);
|
|
}
|
|
this.startGuaranteeing();
|
|
}
|
|
forget(subscription) {
|
|
logger.log(`SubscriptionGuarantor forgetting ${subscription.identifier}`);
|
|
this.pendingSubscriptions = this.pendingSubscriptions.filter((s => s !== subscription));
|
|
}
|
|
startGuaranteeing() {
|
|
this.stopGuaranteeing();
|
|
this.retrySubscribing();
|
|
}
|
|
stopGuaranteeing() {
|
|
clearTimeout(this.retryTimeout);
|
|
}
|
|
retrySubscribing() {
|
|
this.retryTimeout = setTimeout((() => {
|
|
if (this.subscriptions && typeof this.subscriptions.subscribe === "function") {
|
|
this.pendingSubscriptions.map((subscription => {
|
|
logger.log(`SubscriptionGuarantor resubscribing ${subscription.identifier}`);
|
|
this.subscriptions.subscribe(subscription);
|
|
}));
|
|
}
|
|
}), 500);
|
|
}
|
|
}
|
|
class Subscriptions {
|
|
constructor(consumer) {
|
|
this.consumer = consumer;
|
|
this.guarantor = new SubscriptionGuarantor(this);
|
|
this.subscriptions = [];
|
|
}
|
|
create(channelName, mixin) {
|
|
const channel = channelName;
|
|
const params = typeof channel === "object" ? channel : {
|
|
channel: channel
|
|
};
|
|
const subscription = new Subscription(this.consumer, params, mixin);
|
|
return this.add(subscription);
|
|
}
|
|
add(subscription) {
|
|
this.subscriptions.push(subscription);
|
|
this.consumer.ensureActiveConnection();
|
|
this.notify(subscription, "initialized");
|
|
this.subscribe(subscription);
|
|
return subscription;
|
|
}
|
|
remove(subscription) {
|
|
this.forget(subscription);
|
|
if (!this.findAll(subscription.identifier).length) {
|
|
this.sendCommand(subscription, "unsubscribe");
|
|
}
|
|
return subscription;
|
|
}
|
|
reject(identifier) {
|
|
return this.findAll(identifier).map((subscription => {
|
|
this.forget(subscription);
|
|
this.notify(subscription, "rejected");
|
|
return subscription;
|
|
}));
|
|
}
|
|
forget(subscription) {
|
|
this.guarantor.forget(subscription);
|
|
this.subscriptions = this.subscriptions.filter((s => s !== subscription));
|
|
return subscription;
|
|
}
|
|
findAll(identifier) {
|
|
return this.subscriptions.filter((s => s.identifier === identifier));
|
|
}
|
|
reload() {
|
|
return this.subscriptions.map((subscription => this.subscribe(subscription)));
|
|
}
|
|
notifyAll(callbackName, ...args) {
|
|
return this.subscriptions.map((subscription => this.notify(subscription, callbackName, ...args)));
|
|
}
|
|
notify(subscription, callbackName, ...args) {
|
|
let subscriptions;
|
|
if (typeof subscription === "string") {
|
|
subscriptions = this.findAll(subscription);
|
|
} else {
|
|
subscriptions = [ subscription ];
|
|
}
|
|
return subscriptions.map((subscription => typeof subscription[callbackName] === "function" ? subscription[callbackName](...args) : undefined));
|
|
}
|
|
subscribe(subscription) {
|
|
if (this.sendCommand(subscription, "subscribe")) {
|
|
this.guarantor.guarantee(subscription);
|
|
}
|
|
}
|
|
confirmSubscription(identifier) {
|
|
logger.log(`Subscription confirmed ${identifier}`);
|
|
this.findAll(identifier).map((subscription => this.guarantor.forget(subscription)));
|
|
}
|
|
sendCommand(subscription, command) {
|
|
const {identifier: identifier} = subscription;
|
|
return this.consumer.send({
|
|
command: command,
|
|
identifier: identifier
|
|
});
|
|
}
|
|
}
|
|
class Consumer {
|
|
constructor(url) {
|
|
this._url = url;
|
|
this.subscriptions = new Subscriptions(this);
|
|
this.connection = new Connection(this);
|
|
}
|
|
get url() {
|
|
return createWebSocketURL(this._url);
|
|
}
|
|
send(data) {
|
|
return this.connection.send(data);
|
|
}
|
|
connect() {
|
|
return this.connection.open();
|
|
}
|
|
disconnect() {
|
|
return this.connection.close({
|
|
allowReconnect: false
|
|
});
|
|
}
|
|
ensureActiveConnection() {
|
|
if (!this.connection.isActive()) {
|
|
return this.connection.open();
|
|
}
|
|
}
|
|
}
|
|
function createWebSocketURL(url) {
|
|
if (typeof url === "function") {
|
|
url = url();
|
|
}
|
|
if (url && !/^wss?:/i.test(url)) {
|
|
const a = document.createElement("a");
|
|
a.href = url;
|
|
a.href = a.href;
|
|
a.protocol = a.protocol.replace("http", "ws");
|
|
return a.href;
|
|
} else {
|
|
return url;
|
|
}
|
|
}
|
|
function createConsumer(url = getConfig("url") || INTERNAL.default_mount_path) {
|
|
return new Consumer(url);
|
|
}
|
|
function getConfig(name) {
|
|
const element = document.head.querySelector(`meta[name='action-cable-${name}']`);
|
|
if (element) {
|
|
return element.getAttribute("content");
|
|
}
|
|
}
|
|
exports.Connection = Connection;
|
|
exports.ConnectionMonitor = ConnectionMonitor;
|
|
exports.Consumer = Consumer;
|
|
exports.INTERNAL = INTERNAL;
|
|
exports.Subscription = Subscription;
|
|
exports.SubscriptionGuarantor = SubscriptionGuarantor;
|
|
exports.Subscriptions = Subscriptions;
|
|
exports.adapters = adapters;
|
|
exports.createConsumer = createConsumer;
|
|
exports.createWebSocketURL = createWebSocketURL;
|
|
exports.getConfig = getConfig;
|
|
exports.logger = logger;
|
|
Object.defineProperty(exports, "__esModule", {
|
|
value: true
|
|
});
|
|
}));
|