-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
refact: Manage across multiple attached streams
Used sets for managing the known keys because they should be unique and finding the difference between the known keys and new announced keys is easy. Might want to consider basing keys entirely on the autobase properties however. Having a separate collection of keys seems more prone to error. Given there are multiple streams now, `announceAll()` was added.
- Loading branch information
1 parent
4ef3d8a
commit a7aa8ca
Showing
3 changed files
with
147 additions
and
43 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,66 +1,116 @@ | ||
| import Protomux from 'protomux' | ||
| import c from 'compact-encoding' | ||
| import b4a from 'b4a' | ||
| import { difference } from './utils/set-operations.js' | ||
|
|
||
| export class AutobaseManager { | ||
| constructor (stream, base, allow, get) { | ||
| constructor (base, allow, get) { | ||
| this.base = base | ||
| this.allow = allow | ||
| this.get = get | ||
|
|
||
| this._inputKeys = new Set() | ||
| this._outputKeys = new Set() | ||
| this._streams = [] | ||
|
|
||
| if (this.base.localInput) { | ||
| this._addKeys([this.base.localInput.key.toString('hex')], 'input') | ||
| } | ||
| if (this.base.localOutput) { | ||
| this._addKeys([this.base.localOutput.key.toString('hex')], 'output') | ||
| } | ||
| } | ||
|
|
||
| attachStream (stream) { | ||
| const self = this | ||
|
|
||
| const mux = Protomux.from(stream) | ||
|
|
||
| const channel = mux.createChannel({ protocol: 'autobase-manager' }) | ||
| channel.open() | ||
|
|
||
| this.inputAnnouncer = channel.addMessage({ | ||
| const inputAnnouncer = channel.addMessage({ | ||
| encoding: c.array(c.string), | ||
| async onmessage (msgs, session) { | ||
| const allowedKeys = msgs.filter((msg) => allow(msg, 'input', session)) | ||
| const allowedKeys = msgs.filter((msg) => self.allow(msg, 'input', session)) | ||
| if (allowedKeys.length) { | ||
| for (const key of allowedKeys) { | ||
| const core = get(b4a.from(key, 'hex')) | ||
| await core.ready() // seems necessary for autobase id setup | ||
|
|
||
| await base.addInput(core) | ||
| // Check if any are new | ||
| const newKeys = difference(allowedKeys, self._inputKeys) | ||
| if (newKeys.size > 0) { | ||
| await self._addKeys(newKeys, 'input') | ||
| } | ||
| } | ||
| } | ||
| }) | ||
|
|
||
| this.outputAnnouncer = channel.addMessage({ | ||
| const outputAnnouncer = channel.addMessage({ | ||
| encoding: c.array(c.string), | ||
| async onmessage (msgs, session) { | ||
| const allowedKeys = msgs.filter((msg) => allow(msg, 'output', session)) | ||
| const allowedKeys = msgs.filter((msg) => self.allow(msg, 'output', session)) | ||
| if (allowedKeys.length) { | ||
| for (const key of allowedKeys) { | ||
| const core = get(b4a.from(key, 'hex')) | ||
|
|
||
| // Necessary for autobase id (aka the core's id) setup | ||
| await core.ready() | ||
|
|
||
| // Update output to ensure up to date before adding | ||
| // Get a 'Batch is out-of-date.' error otherwise | ||
| if (base.started) { | ||
| await base.view.update() | ||
| } | ||
|
|
||
| await base.addOutput(core) | ||
| // Check if any are new | ||
| const newKeys = difference(allowedKeys, self._outputKeys) | ||
| if (newKeys.size > 0) { | ||
| await self._addKeys(newKeys, 'output') | ||
| } | ||
| } | ||
| } | ||
| }) | ||
|
|
||
| if (this.base.localInput || this.base.inputs || this.base.outputs || this.base.localOutput) this.announce() | ||
| const streamRecord = { stream, inputAnnouncer, outputAnnouncer } | ||
| this._streams.push(streamRecord) | ||
| stream.once('close', () => { | ||
| this._streams.slice(this._streams.indexOf(streamRecord), 1) | ||
| }) | ||
|
|
||
| if (this.base.localInput || this.base.inputs || this.base.outputs || this.base.localOutput) this.announce(streamRecord) | ||
| } | ||
|
|
||
| announce () { | ||
| async announce (stream) { | ||
| const keys = this.base.inputs.map((core) => core.key.toString('hex')) | ||
| if (keys.length) { | ||
| this.inputAnnouncer.send(keys) | ||
| // console.log('[' + this.base.localOutput.key.toString('hex').slice(-6) + | ||
| // '] announce keys', keys.map((key) => key.slice(-6))) | ||
| stream.inputAnnouncer.send(keys) | ||
| } | ||
|
|
||
| const outputKeys = this.base.outputs.map((core) => core.key.toString('hex')) | ||
| if (outputKeys.length) { | ||
| this.outputAnnouncer.send(outputKeys) | ||
| // console.log('[' + this.base.localOutput.key.toString('hex').slice(-6) + | ||
| // '] announce outputKeys', outputKeys.map((key) => key.slice(-6))) | ||
| stream.outputAnnouncer.send(outputKeys) | ||
| } | ||
| } | ||
|
|
||
| async announceAll () { | ||
| for (const stream of this._streams) { | ||
| await this.announce(stream) | ||
| } | ||
| } | ||
|
|
||
| async _addKeys (keys, destination) { | ||
| // Get & Ready Cores | ||
| const cores = await Promise.all(Array.from(keys).map(async (key) => { | ||
| const core = this.get(b4a.from(key, 'hex')) | ||
| // Necessary for autobase id (aka the core's id) setup | ||
| await core.ready() | ||
| return core | ||
| })) | ||
|
|
||
| // Add to the corresponding place in autobase | ||
| for (const core of cores) { | ||
| if (destination === 'output') { | ||
| this._outputKeys.add(core.key.toString('hex')) | ||
|
|
||
| // Update output to ensure up to date before adding | ||
| // Get a 'Batch is out-of-date.' error otherwise | ||
| if (this.base.started) await this.base.view.update() | ||
|
|
||
| await this.base.addOutput(core) | ||
| } else { | ||
| this._inputKeys.add(core.key.toString('hex')) | ||
| await this.base.addInput(core) | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,8 @@ | ||
| // One-way set difference | ||
| export function difference (setA, setB) { | ||
| const _difference = new Set(setA) | ||
| for (const elem of setB) { | ||
| _difference.delete(elem) | ||
| } | ||
| return _difference | ||
| } |