Compare commits
3 Commits
98bb74d8b1
...
05c91000e1
Author | SHA1 | Date |
---|---|---|
Derrick Hammer | 05c91000e1 | |
Derrick Hammer | 2b1789f963 | |
Derrick Hammer | 5094ec4b26 |
128
index.js
128
index.js
|
@ -51,10 +51,9 @@ class Channel {
|
||||||
}
|
}
|
||||||
|
|
||||||
async open(handshake) {
|
async open(handshake) {
|
||||||
if (this._mux._slave) {
|
|
||||||
await this._mux.pullLocal();
|
await this._mux.pullLocal();
|
||||||
await this._mux.pullFree();
|
await this._mux.pullFree();
|
||||||
}
|
|
||||||
const freeLen = this._mux._free.length;
|
const freeLen = this._mux._free.length;
|
||||||
const id =
|
const id =
|
||||||
this._mux._free.length > 0
|
this._mux._free.length > 0
|
||||||
|
@ -65,13 +64,11 @@ class Channel {
|
||||||
this._localId = id + 1;
|
this._localId = id + 1;
|
||||||
this._mux._local[id] = this;
|
this._mux._local[id] = this;
|
||||||
|
|
||||||
if (this._mux._slave) {
|
|
||||||
if (freeLen > 0) {
|
if (freeLen > 0) {
|
||||||
await this._mux.popFree(id);
|
await this._mux.popFree(id);
|
||||||
} else {
|
} else {
|
||||||
await this._mux.pushLocal(id);
|
await this._mux.pushLocal(id);
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
|
||||||
if (this._remoteId === 0) {
|
if (this._remoteId === 0) {
|
||||||
this._info.outgoing.push(this._localId);
|
this._info.outgoing.push(this._localId);
|
||||||
|
@ -148,27 +145,27 @@ class Channel {
|
||||||
if (this.closed === true) return;
|
if (this.closed === true) return;
|
||||||
this.closed = true;
|
this.closed = true;
|
||||||
|
|
||||||
|
await this._mux.pullInfos();
|
||||||
|
|
||||||
this._info.opened--;
|
this._info.opened--;
|
||||||
|
|
||||||
|
await this._mux.pushInfos();
|
||||||
|
|
||||||
if (this._remoteId > 0) {
|
if (this._remoteId > 0) {
|
||||||
this._mux._remote[this._remoteId - 1] = null;
|
this._mux._remote[this._remoteId - 1] = null;
|
||||||
this._remoteId = 0;
|
this._remoteId = 0;
|
||||||
if (this._mux._slave) {
|
|
||||||
await this._mux.pullFree();
|
await this._mux.pullFree();
|
||||||
}
|
|
||||||
// If remote has acked, we can reuse the local id now
|
// If remote has acked, we can reuse the local id now
|
||||||
// otherwise, we need to wait for the "ack" to arrive
|
// otherwise, we need to wait for the "ack" to arrive
|
||||||
this._mux._free.push(this._localId - 1);
|
this._mux._free.push(this._localId - 1);
|
||||||
|
|
||||||
if (this._mux._slave) {
|
|
||||||
await this._mux.pushFree(this._localId - 1);
|
await this._mux.pushFree(this._localId - 1);
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
|
||||||
this._mux._local[this._localId - 1] = null;
|
this._mux._local[this._localId - 1] = null;
|
||||||
this._localId = 0;
|
this._localId = 0;
|
||||||
|
|
||||||
this._mux._gc(this._info);
|
await this._mux._gc(this._info);
|
||||||
this._track(this.onclose(isRemote, this));
|
this._track(this.onclose(isRemote, this));
|
||||||
|
|
||||||
if (this._active === 0) this._destroy();
|
if (this._active === 0) this._destroy();
|
||||||
|
@ -361,7 +358,7 @@ module.exports = class Protomux {
|
||||||
return info ? info.opened > 0 : false;
|
return info ? info.opened > 0 : false;
|
||||||
}
|
}
|
||||||
|
|
||||||
createChannel({
|
async createChannel({
|
||||||
userData = null,
|
userData = null,
|
||||||
protocol,
|
protocol,
|
||||||
aliases = [],
|
aliases = [],
|
||||||
|
@ -375,7 +372,7 @@ module.exports = class Protomux {
|
||||||
}) {
|
}) {
|
||||||
if (this.stream.destroyed) return null;
|
if (this.stream.destroyed) return null;
|
||||||
|
|
||||||
const info = this._get(protocol, id, aliases);
|
const info = await this._get(protocol, id, aliases);
|
||||||
if (unique && info.opened > 0) return null;
|
if (unique && info.opened > 0) return null;
|
||||||
|
|
||||||
if (info.incoming.length === 0) {
|
if (info.incoming.length === 0) {
|
||||||
|
@ -396,7 +393,12 @@ module.exports = class Protomux {
|
||||||
|
|
||||||
this._remoteBacklog--;
|
this._remoteBacklog--;
|
||||||
|
|
||||||
|
await this.pullInfos();
|
||||||
|
|
||||||
const remoteId = info.incoming.shift();
|
const remoteId = info.incoming.shift();
|
||||||
|
|
||||||
|
await this.pushInfos();
|
||||||
|
|
||||||
const r = this._remote[remoteId - 1];
|
const r = this._remote[remoteId - 1];
|
||||||
if (r === null) return null;
|
if (r === null) return null;
|
||||||
|
|
||||||
|
@ -461,9 +463,11 @@ module.exports = class Protomux {
|
||||||
this.stream.write(state.buffer);
|
this.stream.write(state.buffer);
|
||||||
}
|
}
|
||||||
|
|
||||||
_get(protocol, id, aliases = []) {
|
async _get(protocol, id, aliases = []) {
|
||||||
const key = toKey(protocol, id);
|
const key = toKey(protocol, id);
|
||||||
|
|
||||||
|
await this.pullInfos();
|
||||||
|
|
||||||
let info = this._infos.get(key);
|
let info = this._infos.get(key);
|
||||||
if (info) return info;
|
if (info) return info;
|
||||||
|
|
||||||
|
@ -486,18 +490,22 @@ module.exports = class Protomux {
|
||||||
this._infos.set(key, info);
|
this._infos.set(key, info);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
await this.pushInfos();
|
||||||
|
|
||||||
return info;
|
return info;
|
||||||
}
|
}
|
||||||
|
|
||||||
_gc(info) {
|
async _gc(info) {
|
||||||
if (
|
if (
|
||||||
info.opened === 0 &&
|
info.opened === 0 &&
|
||||||
info.outgoing.length === 0 &&
|
info.outgoing.length === 0 &&
|
||||||
info.incoming.length === 0
|
info.incoming.length === 0
|
||||||
) {
|
) {
|
||||||
|
await this.pullInfos();
|
||||||
this._infos.delete(info.key);
|
this._infos.delete(info.key);
|
||||||
|
|
||||||
for (const alias of info.aliases) this._infos.delete(alias);
|
for (const alias of info.aliases) this._infos.delete(alias);
|
||||||
|
await this.pullInfos();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -606,45 +614,41 @@ module.exports = class Protomux {
|
||||||
}
|
}
|
||||||
|
|
||||||
const rid = remoteId - 1;
|
const rid = remoteId - 1;
|
||||||
const info = this._get(protocol, id);
|
const info = await this._get(protocol, id);
|
||||||
|
|
||||||
if (this._slave) {
|
|
||||||
await this.pullRemote();
|
await this.pullRemote();
|
||||||
}
|
|
||||||
|
|
||||||
// allow the remote to grow the ids by one
|
// allow the remote to grow the ids by one
|
||||||
if (this._remote.length === rid) {
|
if (this._remote.length === rid) {
|
||||||
this._remote.push(null);
|
this._remote.push(null);
|
||||||
if (this._slave) {
|
|
||||||
await this.pushRemote(this._remote.length - 1);
|
await this.pushRemote(this._remote.length - 1);
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
|
||||||
if (rid >= this._remote.length || this._remote[rid] !== null) {
|
if (rid >= this._remote.length || this._remote[rid] !== null) {
|
||||||
throw new Error("Invalid open message");
|
throw new Error("Invalid open message");
|
||||||
}
|
}
|
||||||
|
|
||||||
if (info.outgoing.length > 0) {
|
if (info.outgoing.length > 0) {
|
||||||
|
await this.pullInfos();
|
||||||
const localId = info.outgoing.shift();
|
const localId = info.outgoing.shift();
|
||||||
const session = this._local[localId - 1];
|
const session = this._local[localId - 1];
|
||||||
|
|
||||||
|
await this.pushInfos();
|
||||||
|
|
||||||
if (session === null) {
|
if (session === null) {
|
||||||
if (this._slave) {
|
|
||||||
await this.pullFree();
|
await this.pullFree();
|
||||||
}
|
|
||||||
// we already closed the channel - ignore
|
// we already closed the channel - ignore
|
||||||
this._free.push(localId - 1);
|
this._free.push(localId - 1);
|
||||||
if (this._slave) {
|
|
||||||
await this.pushFree(localId - 1);
|
await this.pushFree(localId - 1);
|
||||||
}
|
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
this._remote[rid] = { state, pending: null, session: null };
|
this._remote[rid] = { state, pending: null, session: null };
|
||||||
|
|
||||||
if (this._slave) {
|
|
||||||
await this.pushRemote(remoteId);
|
await this.pushRemote(remoteId);
|
||||||
}
|
|
||||||
|
|
||||||
session._remoteId = remoteId;
|
session._remoteId = remoteId;
|
||||||
session._fullyOpen();
|
session._fullyOpen();
|
||||||
|
@ -662,12 +666,13 @@ module.exports = class Protomux {
|
||||||
throw new Error("Remote exceeded backlog");
|
throw new Error("Remote exceeded backlog");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
await this.pullInfos();
|
||||||
|
|
||||||
info.pairing++;
|
info.pairing++;
|
||||||
info.incoming.push(remoteId);
|
info.incoming.push(remoteId);
|
||||||
|
|
||||||
if (this._slave) {
|
await this.pushInfos();
|
||||||
await this.pushRemote(remoteId);
|
await this.pushRemote(remoteId);
|
||||||
}
|
|
||||||
|
|
||||||
this._requestSession(protocol, id, info).catch(this._safeDestroyBound);
|
this._requestSession(protocol, id, info).catch(this._safeDestroyBound);
|
||||||
}
|
}
|
||||||
|
@ -680,22 +685,23 @@ module.exports = class Protomux {
|
||||||
const i = info.outgoing.indexOf(localId);
|
const i = info.outgoing.indexOf(localId);
|
||||||
if (i === -1) continue;
|
if (i === -1) continue;
|
||||||
|
|
||||||
|
await this.pullInfos();
|
||||||
|
|
||||||
info.outgoing.splice(i, 1);
|
info.outgoing.splice(i, 1);
|
||||||
|
|
||||||
|
await this.pushInfos();
|
||||||
|
|
||||||
const session = this._local[localId - 1];
|
const session = this._local[localId - 1];
|
||||||
|
|
||||||
if (this._slave) {
|
|
||||||
await this.pullFree(localId - 1);
|
await this.pullFree(localId - 1);
|
||||||
}
|
|
||||||
|
|
||||||
this._free.push(localId - 1);
|
this._free.push(localId - 1);
|
||||||
if (this._slave) {
|
|
||||||
await this.pushFree(localId - 1);
|
await this.pushFree(localId - 1);
|
||||||
}
|
|
||||||
|
|
||||||
if (session !== null) session._close(true);
|
if (session !== null) session._close(true);
|
||||||
|
|
||||||
this._gc(info);
|
await this._gc(info);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -722,13 +728,25 @@ module.exports = class Protomux {
|
||||||
|
|
||||||
if (notify) await notify(id);
|
if (notify) await notify(id);
|
||||||
|
|
||||||
if (--info.pairing > 0) return;
|
await this.pullInfos();
|
||||||
|
|
||||||
|
let p = --info.pairing;
|
||||||
|
|
||||||
|
await this.pushInfos();
|
||||||
|
|
||||||
|
if (p > 0) return;
|
||||||
|
|
||||||
while (info.incoming.length > 0) {
|
while (info.incoming.length > 0) {
|
||||||
this._rejectSession(info, info.incoming.shift());
|
await this.pullInfos();
|
||||||
|
|
||||||
|
const id = info.incoming.shift();
|
||||||
|
|
||||||
|
await this.pushInfos();
|
||||||
|
|
||||||
|
this._rejectSession(info, id);
|
||||||
}
|
}
|
||||||
|
|
||||||
this._gc(info);
|
await this._gc(info);
|
||||||
}
|
}
|
||||||
|
|
||||||
_rejectSession(info, remoteId) {
|
_rejectSession(info, remoteId) {
|
||||||
|
@ -783,19 +801,43 @@ module.exports = class Protomux {
|
||||||
}
|
}
|
||||||
|
|
||||||
async popFree(id) {
|
async popFree(id) {
|
||||||
|
if (!this._slave) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
await this.stream.syncProtomux("popFree", id);
|
await this.stream.syncProtomux("popFree", id);
|
||||||
}
|
}
|
||||||
async pushFree(id) {
|
async pushFree(id) {
|
||||||
|
if (!this._slave) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
await this.stream.syncProtomux("pushFree", id);
|
await this.stream.syncProtomux("pushFree", id);
|
||||||
}
|
}
|
||||||
async pushLocal(id) {
|
async pushLocal(id) {
|
||||||
|
if (!this._slave) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
await this.stream.syncProtomux("pushLocal", id);
|
await this.stream.syncProtomux("pushLocal", id);
|
||||||
}
|
}
|
||||||
async pushRemote(id) {
|
async pushRemote(id) {
|
||||||
|
if (!this._slave) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
await this.stream.syncProtomux("pushRemote", id);
|
await this.stream.syncProtomux("pushRemote", id);
|
||||||
}
|
}
|
||||||
|
async pushInfos() {
|
||||||
|
if (!this._slave) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
await this.stream.syncProtomux(
|
||||||
|
"pushInfos",
|
||||||
|
Array.from(this._infos.entries())
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
async pullLocal() {
|
async pullLocal() {
|
||||||
|
if (!this._slave) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
const ids = await this.stream.syncProtomux("pullLocal");
|
const ids = await this.stream.syncProtomux("pullLocal");
|
||||||
|
|
||||||
ids.forEach((item) => {
|
ids.forEach((item) => {
|
||||||
|
@ -806,12 +848,18 @@ module.exports = class Protomux {
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
async pullFree() {
|
async pullFree() {
|
||||||
|
if (!this._slave) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
const ids = await this.stream.syncProtomux("pullFree");
|
const ids = await this.stream.syncProtomux("pullFree");
|
||||||
|
|
||||||
this._free = Array.from(new Set([...this._free, ...ids]));
|
this._free = Array.from(new Set([...this._free, ...ids]));
|
||||||
this._free = this._free.filter((item) => item !== null);
|
this._free = this._free.filter((item) => item !== null);
|
||||||
}
|
}
|
||||||
async pullRemote() {
|
async pullRemote() {
|
||||||
|
if (!this._slave) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
const ids = await this.stream.syncProtomux("pullRemote");
|
const ids = await this.stream.syncProtomux("pullRemote");
|
||||||
|
|
||||||
ids.forEach((item) => {
|
ids.forEach((item) => {
|
||||||
|
@ -821,6 +869,18 @@ module.exports = class Protomux {
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
async pullInfos() {
|
||||||
|
if (!this._slave) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
const info = await this.stream.syncProtomux(
|
||||||
|
"pullInfos",
|
||||||
|
Array.from(this._infos.entries())
|
||||||
|
);
|
||||||
|
|
||||||
|
this._infos = new Map(info);
|
||||||
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
function noop() {}
|
function noop() {}
|
||||||
|
|
Reference in New Issue