return null
}
return this._subs[e] || null
}
_isServerSub(e) {
return void 0 !== this._serverSubs[e]
}
_sendSubscribeCommands() {
const e = [];
for (const t in this._subs) {
if (!this._subs.hasOwnProperty(t))
continue;
const n = this._subs[t];
if (!0 !== n._inflight && n.state === SubscriptionState.Subscribing) {
const t = n._subscribe();
t && e.push(t)
}
}
return e
}
_connectResponse(e) {
if (this._transportIsOpen = !0,
this._transportWasOpen = !0,
this._reconnectAttempts = 0,
this._refreshRequired = !1,
this._isConnected())
return;
this._client = e.client,
this._setState(State$1.Connected),
this._refreshTimeout && clearTimeout(this._refreshTimeout),
e.expires && (this._refreshTimeout = setTimeout(( () => this._refresh()), ttlMilliseconds(e.ttl))),
this._session = e.session,
this._node = e.node,
this.startBatching(),
this._sendSubscribeCommands(),
this.stopBatching();
const t = {
client: e.client,
transport: this._transport.subName()
};
e.data && (t.data = e.data),
this.emit("connected", t),
this._resolvePromises(),
this._processServerSubs(e.subs || {}),
e.ping && e.ping > 0 ? (this._serverPing = 1e3 * e.ping,
this._sendPong = !0 === e.pong,
this._waitServerPing()) : this._serverPing = 0
}
_processServerSubs(e) {
for (const t in e) {
if (!e.hasOwnProperty(t))
continue;
const n = e[t];
this._serverSubs[t] = {
offset: n.offset,
epoch: n.epoch,
recoverable: n.recoverable || !1
};
const i = this._getSubscribeContext(t, n);
this.emit("subscribed", i)
}
for (const t in e) {
if (!e.hasOwnProperty(t))
continue;
const n = e[t];
if (n.recovered) {
const e = n.publications;
if (e && e.length > 0)
for (const n in e)
e.hasOwnProperty(n) && this._handlePublication(t, e[n])
}
}
for (const t in this._serverSubs)
this._serverSubs.hasOwnProperty(t) && (e[t] || (this.emit("unsubscribed", {
channel: t
}),
delete this._serverSubs[t]))
}
_clearRefreshTimeout() {
null !== this._refreshTimeout && (clearTimeout(this._refreshTimeout),
this._refreshTimeout = null)
}
_clearReconnectTimeout() {
null !== this._reconnectTimeout && (clearTimeout(this._reconnectTimeout),
this._reconnectTimeout = null)
}
_clearServerPingTimeout() {
null !== this._serverPingTimeout && (clearTimeout(this._serverPingTimeout),
this._serverPingTimeout = null)
}
_waitServerPing() {
0 !== this._config.maxServerPingDelay && this._isConnected() && (this._clearServerPingTimeout(),
this._serverPingTimeout = setTimeout(( () => {
this._isConnected() && this._disconnect(connectingCodes.noPing, "no ping", !0)
}
), this._serverPing + this._config.maxServerPingDelay))
}
_getSubscribeContext(e, t) {
const n = {
channel: e,
positioned: !1,
recoverable: !1,
wasRecovering: !1,
recovered: !1,
hasRecoveredPublications: !1
};
t.recovered && (n.recovered = !0),
t.positioned && (n.positioned = !0),
t.recoverable && (n.recoverable = !0),
t.was_recovering && (n.wasRecovering = !0);
let i = "";
"epoch"in t && (i = t.epoch);
let r = 0;
return "offset"in t && (r = t.offset),
(n.positioned || n.recoverable) && (n.streamPosition = {
offset: r,
epoch: i
}),
Array.isArray(t.publications) && t.publications.length > 0 && (n.hasRecoveredPublications = !0),
t.data && (n.data = t.data),
n
}
_handleReply(e, t) {
const n = e.id;
if (!(n in this._callbacks))
return void t();
const i = this._callbacks[n];
if (clearTimeout(this._callbacks[n].timeout),
delete this._callbacks[n],
errorExists(e)) {
const n = i.errback;
if (!n)
return void t();
n({
error: {
code: e.error.code,
message: e.error.message || "",
temporary: e.error.temporary || !1
},
next: t
})
} else {
const n = i.callback;
if (!n)
return;
|
