a collection of lightweight TypeScript packages for AT Protocol, the protocol powering Bluesky
atproto bluesky typescript npm
101
fork

Configure Feed

Select the types of activity you want to include in your feed.

fix(car): allow `RepoReader.fromStream` to work even if `await using` is not supported by host environment

closes https://github.com/mary-ext/atcute/issues/36

Mary bea36b2c ddd5cb56

+87 -77
+5
.changeset/icy-dryers-lick.md
··· 1 + --- 2 + '@atcute/car': patch 3 + --- 4 + 5 + allow `RepoReader.fromStream` to work even if `await using` is not supported by host environment.
+82 -77
packages/utilities/car/lib/v4/repo-reader/stream-repo-reader.ts
··· 93 93 return this.dispose(); 94 94 }, 95 95 async *[Symbol.asyncIterator]() { 96 - await using car = CarReader.fromStream(stream); 96 + // await using car = CarReader.fromStream(stream); 97 + const car = CarReader.fromStream(stream); 97 98 98 - const pending = new Map<string, EntryMeta>(); 99 - const strays = new Map<string, CarReader.CarEntry>(); 99 + try { 100 + const pending = new Map<string, EntryMeta>(); 101 + const strays = new Map<string, CarReader.CarEntry>(); 100 102 101 - const queue = new Queue<Task>(); 103 + const queue = new Queue<Task>(); 102 104 103 - const request = (cid: string, meta: EntryMeta): void => { 104 - const entry = strays.get(cid); 105 + const request = (cid: string, meta: EntryMeta): void => { 106 + const entry = strays.get(cid); 105 107 106 - if (entry !== undefined) { 107 - strays.delete(cid); 108 - queue.enqueue({ c: cid, e: entry, m: meta }); 109 - } else { 110 - pending.set(cid, meta); 111 - } 112 - }; 108 + if (entry !== undefined) { 109 + strays.delete(cid); 110 + queue.enqueue({ c: cid, e: entry, m: meta }); 111 + } else { 112 + pending.set(cid, meta); 113 + } 114 + }; 113 115 114 - { 115 - const roots = await car.roots(); 116 - assert(roots.length === 1, `expected only 1 root in the car archive; got=${roots.length}`); 116 + { 117 + const roots = await car.roots(); 118 + assert(roots.length === 1, `expected only 1 root in the car archive; got=${roots.length}`); 117 119 118 - const rootCid = roots[0].$link; 119 - request(rootCid, { t: 0 }); 120 - } 120 + const rootCid = roots[0].$link; 121 + request(rootCid, { t: 0 }); 122 + } 121 123 122 - for await (const entry of car) { 123 - const cid = CID.toString(entry.cid); 124 + for await (const entry of car) { 125 + const cid = CID.toString(entry.cid); 124 126 125 - { 126 - const meta = pending.get(cid); 127 + { 128 + const meta = pending.get(cid); 127 129 128 - if (meta !== undefined) { 129 - pending.delete(cid); 130 - queue.enqueue({ c: cid, e: entry, m: meta }); 131 - } else { 132 - strays.set(cid, entry); 130 + if (meta !== undefined) { 131 + pending.delete(cid); 132 + queue.enqueue({ c: cid, e: entry, m: meta }); 133 + } else { 134 + strays.set(cid, entry); 135 + } 133 136 } 134 - } 135 137 136 - let task: Task | undefined; 137 - while ((task = queue.dequeue())) { 138 - const { c: cid, e: entry, m: meta } = task; 138 + let task: Task | undefined; 139 + while ((task = queue.dequeue())) { 140 + const { c: cid, e: entry, m: meta } = task; 139 141 140 - switch (meta.t) { 141 - case 0: { 142 - const commit = CBOR.decode(entry.bytes); 143 - assert(isCommit(commit), `expected commit block; cid=${cid}`); 142 + switch (meta.t) { 143 + case 0: { 144 + const commit = CBOR.decode(entry.bytes); 145 + assert(isCommit(commit), `expected commit block; cid=${cid}`); 144 146 145 - request(commit.data.$link, { t: 1 }); 146 - break; 147 - } 148 - case 1: { 149 - const node = CBOR.decode(entry.bytes); 150 - assert(isMstNode(node), `expected mst node block; cid=${cid}`); 147 + request(commit.data.$link, { t: 1 }); 148 + break; 149 + } 150 + case 1: { 151 + const node = CBOR.decode(entry.bytes); 152 + assert(isMstNode(node), `expected mst node block; cid=${cid}`); 151 153 152 - const entries = node.e; 153 - const left = node.l; 154 + const entries = node.e; 155 + const left = node.l; 154 156 155 - let lastKey = ''; 157 + let lastKey = ''; 156 158 157 - if (left !== null) { 158 - request(left.$link, meta); 159 - } 159 + if (left !== null) { 160 + request(left.$link, meta); 161 + } 160 162 161 - for (let i = 0, il = entries.length; i < il; i++) { 162 - const entry = entries[i]; 163 - const next = entry.t; 163 + for (let i = 0, il = entries.length; i < il; i++) { 164 + const entry = entries[i]; 165 + const next = entry.t; 164 166 165 - const key_str = decodeUtf8From(CBOR.fromBytes(entry.k)); 166 - const key = lastKey.slice(0, entry.p) + key_str; 167 + const key_str = decodeUtf8From(CBOR.fromBytes(entry.k)); 168 + const key = lastKey.slice(0, entry.p) + key_str; 167 169 168 - lastKey = key; 170 + lastKey = key; 169 171 170 - request(entry.v.$link, { t: 2, k: key }); 172 + request(entry.v.$link, { t: 2, k: key }); 171 173 172 - if (next !== null) { 173 - request(next.$link, { t: 1 }); 174 + if (next !== null) { 175 + request(next.$link, { t: 1 }); 176 + } 174 177 } 178 + 179 + break; 175 180 } 181 + case 2: { 182 + const [collection, rkey] = meta.k.split('/'); 176 183 177 - break; 184 + yield new RepoEntry(collection, rkey, CID.toCidLink(entry.cid), entry); 185 + break; 186 + } 178 187 } 179 - case 2: { 180 - const [collection, rkey] = meta.k.split('/'); 188 + } 189 + } 181 190 182 - yield new RepoEntry(collection, rkey, CID.toCidLink(entry.cid), entry); 183 - break; 191 + missingBlocks = Array.from(pending, ([cid, meta]): MissingBlockEntry => { 192 + switch (meta.t) { 193 + case 0: { 194 + return { cid, type: 'commit' }; 195 + } 196 + case 1: { 197 + return { cid, type: 'mst-node' }; 198 + } 199 + case 2: { 200 + return { cid, type: 'record', key: meta.k }; 184 201 } 185 202 } 186 - } 203 + }); 204 + } finally { 205 + await car.dispose(); 187 206 } 188 - 189 - missingBlocks = Array.from(pending, ([cid, meta]): MissingBlockEntry => { 190 - switch (meta.t) { 191 - case 0: { 192 - return { cid, type: 'commit' }; 193 - } 194 - case 1: { 195 - return { cid, type: 'mst-node' }; 196 - } 197 - case 2: { 198 - return { cid, type: 'record', key: meta.k }; 199 - } 200 - } 201 - }); 202 207 }, 203 208 }; 204 209 };