| … | … | @@ -80,6 +80,7 @@ export class DidPeer { |
| 80 | 80 | // consumer's job (the gateway lens), exactly as it verifies commits. |
| 81 | 81 | this._labelKey = null |
| 82 | 82 | this.labels = null |
| 83 | this._onLabels = null // set by DidPeerRouter: (peer) => void once a label core is open |
| 83 | 84 | this._channels = new Set() |
| 84 | 85 | this.head = null |
| 85 | 86 | this._opening = null |
| … | … | @@ -158,6 +159,7 @@ export class DidPeer { |
| 158 | 159 | core.on('append', () => this._announceAll()) |
| 159 | 160 | this.labels = core |
| 160 | 161 | this._announceAll() |
| 162 | if (this._onLabels) { try { this._onLabels(this) } catch {} } |
| 161 | 163 | return true |
| 162 | 164 | } finally { this._labelOpening = false } |
| 163 | 165 | } |
| … | … | @@ -641,6 +643,20 @@ export class DidPeerRouter { |
| 641 | 643 | |
| 642 | 644 | _isCared (did) { return this._fronted.has(did) || this._peers.has(did) || this._opts.has(did) } |
| 643 | 645 | _joinDid (did) { if (this._join) { try { this._join(did) } catch {} } } |
| 646 | // A label core opening is not a firehose event, so a consumer serving labels (the gateway) has no other |
| 647 | // way to learn that a front gained one — the common case is the labeler upgrading while we run, which |
| 648 | // the router learns of from an announce on an already-open front. → unsubscribe(). |
| 649 | onLabels (fn) { |
| 650 | if (!this._labelListeners) this._labelListeners = new Set() |
| 651 | this._labelListeners.add(fn) |
| 652 | return () => this._labelListeners.delete(fn) |
| 653 | } |
| 654 | |
| 655 | _tellLabels (did, peer) { |
| 656 | if (!this._labelListeners) return |
| 657 | for (const fn of this._labelListeners) { try { fn(did, peer) } catch {} } |
| 658 | } |
| 659 | |
| 644 | 660 | _wake (did) { const ws = this._waiters.get(did); if (ws) { this._waiters.delete(did); for (const fn of ws) fn() } } |
| 645 | 661 | |
| 646 | 662 | _setHandle (handle, did) { |
| … | … | @@ -797,6 +813,7 @@ export class DidPeerRouter { |
| 797 | 813 | if (entry.coords.handle) peer.handle = entry.coords.handle |
| 798 | 814 | // EVERY peer announcing this (did, firehoseKey) is a candidate; _call tries them until one answers. |
| 799 | 815 | if (this._useRpc) for (const r of entry.conns.values()) if (r) peer.addRpc(r) |
| 816 | peer._onLabels = (p) => this._tellLabels(did, p) |
| 800 | 817 | await peer.provide({ repoKey: entry.coords.repoKey, firehoseKey: entry.coords.firehoseKey, labelKey: entry.coords.labelKey || null }) |
| 801 | 818 | // Re-sync: more peers may have announced this same (did, firehoseKey) during the open above. |
| 802 | 819 | if (this._useRpc) for (const r of entry.conns.values()) if (r) peer.addRpc(r) |