Compare commits
No commits in common. "e0170f9dfcde4e279d78d9e4554cb4592e577612" and "1387b060a4e37ad5f21db2863d43d0aa33cdd619" have entirely different histories.
e0170f9dfc
...
1387b060a4
155
index.js
155
index.js
|
|
@ -51,9 +51,10 @@ class Channel {
|
|||
}
|
||||
|
||||
async open(handshake) {
|
||||
await this._mux.pullLocal();
|
||||
await this._mux.pullFree();
|
||||
|
||||
if (this._slave) {
|
||||
await this._mux.pullLocal();
|
||||
await this._mux.pullFree();
|
||||
}
|
||||
const freeLen = this._mux._free.length;
|
||||
const id =
|
||||
this._mux._free.length > 0
|
||||
|
|
@ -64,15 +65,16 @@ class Channel {
|
|||
this._localId = id + 1;
|
||||
this._mux._local[id] = this;
|
||||
|
||||
if (freeLen > 0) {
|
||||
await this._mux.popFree(id);
|
||||
} else {
|
||||
await this._mux.pushLocal(id);
|
||||
if (this._slave) {
|
||||
if (freeLen > 0) {
|
||||
await this._mux.popFree(id);
|
||||
} else {
|
||||
await this._mux.pushLocal(id);
|
||||
}
|
||||
}
|
||||
|
||||
if (this._remoteId === 0) {
|
||||
this._info.outgoing.push(this._localId);
|
||||
await this._mux.pushInfos();
|
||||
}
|
||||
|
||||
const state = { buffer: null, start: 2, end: 2 };
|
||||
|
|
@ -146,27 +148,27 @@ class Channel {
|
|||
if (this.closed === true) return;
|
||||
this.closed = true;
|
||||
|
||||
await this._mux.pullInfos();
|
||||
|
||||
this._info.opened--;
|
||||
|
||||
await this._mux.pushInfos();
|
||||
|
||||
if (this._remoteId > 0) {
|
||||
this._mux._remote[this._remoteId - 1] = null;
|
||||
this._remoteId = 0;
|
||||
await this._mux.pullFree();
|
||||
if (this._slave) {
|
||||
await this._mux.pullFree();
|
||||
}
|
||||
// If remote has acked, we can reuse the local id now
|
||||
// otherwise, we need to wait for the "ack" to arrive
|
||||
this._mux._free.push(this._localId - 1);
|
||||
|
||||
await this._mux.pushFree(this._localId - 1);
|
||||
if (this._slave) {
|
||||
await this._mux.pushFree(this._localId - 1);
|
||||
}
|
||||
}
|
||||
|
||||
this._mux._local[this._localId - 1] = null;
|
||||
this._localId = 0;
|
||||
|
||||
await this._mux._gc(this._info);
|
||||
this._mux._gc(this._info);
|
||||
this._track(this.onclose(isRemote, this));
|
||||
|
||||
if (this._active === 0) this._destroy();
|
||||
|
|
@ -223,34 +225,34 @@ class Channel {
|
|||
type,
|
||||
encoding,
|
||||
onmessage,
|
||||
async recv(state, session) {
|
||||
session._track(m.onmessage(await encoding.decode(state), session));
|
||||
recv(state, session) {
|
||||
session._track(m.onmessage(encoding.decode(state), session));
|
||||
},
|
||||
async send(m, session = s) {
|
||||
send(m, session = s) {
|
||||
if (session.closed === true) return false;
|
||||
|
||||
const mux = session._mux;
|
||||
const state = { buffer: null, start: 0, end: typeLen };
|
||||
|
||||
if (mux._batch !== null) {
|
||||
await encoding.preencode(state, m);
|
||||
encoding.preencode(state, m);
|
||||
state.buffer = mux._alloc(state.end);
|
||||
|
||||
c.uint.encode(state, type);
|
||||
await encoding.encode(state, m);
|
||||
encoding.encode(state, m);
|
||||
|
||||
mux._pushBatch(session._localId, state.buffer);
|
||||
return true;
|
||||
}
|
||||
|
||||
c.uint.preencode(state, session._localId);
|
||||
await encoding.preencode(state, m);
|
||||
encoding.preencode(state, m);
|
||||
|
||||
state.buffer = mux._alloc(state.end);
|
||||
|
||||
c.uint.encode(state, session._localId);
|
||||
c.uint.encode(state, type);
|
||||
await encoding.encode(state, m);
|
||||
encoding.encode(state, m);
|
||||
|
||||
return mux.stream.write(state.buffer);
|
||||
},
|
||||
|
|
@ -278,11 +280,7 @@ class Channel {
|
|||
|
||||
module.exports = class Protomux {
|
||||
constructor(stream, { alloc, slave = false } = {}) {
|
||||
if (stream.userData) {
|
||||
throw new Error("Mux already set for this stream");
|
||||
}
|
||||
|
||||
stream.userData = this;
|
||||
if (stream.userData === null) stream.userData = this;
|
||||
|
||||
this.isProtomux = true;
|
||||
this.stream = stream;
|
||||
|
|
@ -306,6 +304,7 @@ module.exports = class Protomux {
|
|||
|
||||
this._infos = new Map();
|
||||
this._notify = new Map();
|
||||
this._slave = slave;
|
||||
|
||||
this.stream.on("data", this._ondata.bind(this));
|
||||
this.stream.on("end", this._onend.bind(this));
|
||||
|
|
@ -357,7 +356,8 @@ module.exports = class Protomux {
|
|||
const info = this._infos.get(key);
|
||||
return info ? info.opened > 0 : false;
|
||||
}
|
||||
async createChannel({
|
||||
|
||||
createChannel({
|
||||
userData = null,
|
||||
protocol,
|
||||
aliases = [],
|
||||
|
|
@ -371,7 +371,7 @@ module.exports = class Protomux {
|
|||
}) {
|
||||
if (this.stream.destroyed) return null;
|
||||
|
||||
const info = await this._get(protocol, id, aliases);
|
||||
const info = this._get(protocol, id, aliases);
|
||||
if (unique && info.opened > 0) return null;
|
||||
|
||||
if (info.incoming.length === 0) {
|
||||
|
|
@ -392,12 +392,7 @@ module.exports = class Protomux {
|
|||
|
||||
this._remoteBacklog--;
|
||||
|
||||
await this.pullInfos();
|
||||
|
||||
const remoteId = info.incoming.shift();
|
||||
|
||||
await this.pushInfos();
|
||||
|
||||
const r = this._remote[remoteId - 1];
|
||||
if (r === null) return null;
|
||||
|
||||
|
|
@ -462,11 +457,9 @@ module.exports = class Protomux {
|
|||
this.stream.write(state.buffer);
|
||||
}
|
||||
|
||||
async _get(protocol, id, aliases = []) {
|
||||
_get(protocol, id, aliases = []) {
|
||||
const key = toKey(protocol, id);
|
||||
|
||||
await this.pullInfos();
|
||||
|
||||
let info = this._infos.get(key);
|
||||
if (info) return info;
|
||||
|
||||
|
|
@ -489,22 +482,18 @@ module.exports = class Protomux {
|
|||
this._infos.set(key, info);
|
||||
}
|
||||
|
||||
await this.pushInfos();
|
||||
|
||||
return info;
|
||||
}
|
||||
|
||||
async _gc(info) {
|
||||
_gc(info) {
|
||||
if (
|
||||
info.opened === 0 &&
|
||||
info.outgoing.length === 0 &&
|
||||
info.incoming.length === 0
|
||||
) {
|
||||
await this.pullInfos();
|
||||
this._infos.delete(info.key);
|
||||
|
||||
for (const alias of info.aliases) this._infos.delete(alias);
|
||||
await this.pullInfos();
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -613,15 +602,18 @@ module.exports = class Protomux {
|
|||
}
|
||||
|
||||
const rid = remoteId - 1;
|
||||
const info = await this._get(protocol, id);
|
||||
const info = this._get(protocol, id);
|
||||
|
||||
await this.pullRemote();
|
||||
if (this._slave) {
|
||||
await this.pullRemote();
|
||||
}
|
||||
|
||||
// allow the remote to grow the ids by one
|
||||
if (this._remote.length === rid) {
|
||||
this._remote.push(null);
|
||||
|
||||
await this.pushRemote(this._remote.length - 1);
|
||||
if (this._slave) {
|
||||
await this.pushRemote(this._remote.length - 1);
|
||||
}
|
||||
}
|
||||
|
||||
if (rid >= this._remote.length || this._remote[rid] !== null) {
|
||||
|
|
@ -629,25 +621,26 @@ module.exports = class Protomux {
|
|||
}
|
||||
|
||||
if (info.outgoing.length > 0) {
|
||||
await this.pullInfos();
|
||||
const localId = info.outgoing.shift();
|
||||
const session = this._local[localId - 1];
|
||||
|
||||
await this.pushInfos();
|
||||
|
||||
if (session === null) {
|
||||
await this.pullFree();
|
||||
if (this._slave) {
|
||||
await this.pullFree();
|
||||
}
|
||||
// we already closed the channel - ignore
|
||||
this._free.push(localId - 1);
|
||||
|
||||
await this.pushFree(localId - 1);
|
||||
|
||||
if (this._slave) {
|
||||
await this.pushFree(localId - 1);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
this._remote[rid] = { state, pending: null, session: null };
|
||||
|
||||
await this.pushRemote(remoteId);
|
||||
if (this._slave) {
|
||||
await this.pushRemote(remoteId);
|
||||
}
|
||||
|
||||
session._remoteId = remoteId;
|
||||
session._fullyOpen();
|
||||
|
|
@ -665,13 +658,12 @@ module.exports = class Protomux {
|
|||
throw new Error("Remote exceeded backlog");
|
||||
}
|
||||
|
||||
await this.pullInfos();
|
||||
|
||||
info.pairing++;
|
||||
info.incoming.push(remoteId);
|
||||
|
||||
await this.pushInfos();
|
||||
await this.pushRemote(remoteId);
|
||||
if (this._slave) {
|
||||
await this.pushRemote(remoteId);
|
||||
}
|
||||
|
||||
this._requestSession(protocol, id, info).catch(this._safeDestroyBound);
|
||||
}
|
||||
|
|
@ -684,23 +676,22 @@ module.exports = class Protomux {
|
|||
const i = info.outgoing.indexOf(localId);
|
||||
if (i === -1) continue;
|
||||
|
||||
await this.pullInfos();
|
||||
|
||||
info.outgoing.splice(i, 1);
|
||||
|
||||
await this.pushInfos();
|
||||
|
||||
const session = this._local[localId - 1];
|
||||
|
||||
await this.pullFree(localId - 1);
|
||||
if (this._slave) {
|
||||
await this.pullFree(localId - 1);
|
||||
}
|
||||
|
||||
this._free.push(localId - 1);
|
||||
|
||||
await this.pushFree(localId - 1);
|
||||
if (this._slave) {
|
||||
await this.pushFree(localId - 1);
|
||||
}
|
||||
|
||||
if (session !== null) session._close(true);
|
||||
|
||||
await this._gc(info);
|
||||
this._gc(info);
|
||||
return;
|
||||
}
|
||||
|
||||
|
|
@ -727,25 +718,13 @@ module.exports = class Protomux {
|
|||
|
||||
if (notify) await notify(id);
|
||||
|
||||
await this.pullInfos();
|
||||
|
||||
let p = --info.pairing;
|
||||
|
||||
await this.pushInfos();
|
||||
|
||||
if (p > 0) return;
|
||||
if (--info.pairing > 0) return;
|
||||
|
||||
while (info.incoming.length > 0) {
|
||||
await this.pullInfos();
|
||||
|
||||
const id = info.incoming.shift();
|
||||
|
||||
await this.pushInfos();
|
||||
|
||||
this._rejectSession(info, id);
|
||||
this._rejectSession(info, info.incoming.shift());
|
||||
}
|
||||
|
||||
await this._gc(info);
|
||||
this._gc(info);
|
||||
}
|
||||
|
||||
_rejectSession(info, remoteId) {
|
||||
|
|
@ -811,12 +790,6 @@ module.exports = class Protomux {
|
|||
async pushRemote(id) {
|
||||
await this.stream.syncProtomux("pushRemote", id);
|
||||
}
|
||||
async pushInfos() {
|
||||
await this.stream.syncProtomux(
|
||||
"pushInfos",
|
||||
Array.from(this._infos.entries())
|
||||
);
|
||||
}
|
||||
|
||||
async pullLocal() {
|
||||
const ids = await this.stream.syncProtomux("pullLocal");
|
||||
|
|
@ -844,14 +817,6 @@ module.exports = class Protomux {
|
|||
}
|
||||
});
|
||||
}
|
||||
async pullInfos() {
|
||||
const info = await this.stream.syncProtomux(
|
||||
"pullInfos",
|
||||
Array.from(this._infos.entries())
|
||||
);
|
||||
|
||||
this._infos = new Map(info);
|
||||
}
|
||||
};
|
||||
|
||||
function noop() {}
|
||||
|
|
|
|||
Reference in New Issue