fix: dag.import to support ReadableStream

src/dag/index.ts

@@ -222,7 +222,7 @@ export interface DAGAPI {
    * Import all blocks from one or more CARs and optionally recursively pin the roots identified
    * within the CARs.
    */
-  import(sources: Iterable<Uint8Array> | AsyncIterable<Uint8Array> | AsyncIterable<AsyncIterable<Uint8Array>> | Iterable<AsyncIterable<Uint8Array>>, options?: DAGImportOptions): AsyncIterable<DAGImportResult>
+  import(sources: ReadableStream<Uint8Array> | Iterable<Uint8Array> | AsyncIterable<Uint8Array> | AsyncIterable<AsyncIterable<Uint8Array>> | Iterable<AsyncIterable<Uint8Array>>, options?: DAGImportOptions): AsyncIterable<DAGImportResult>
 }
 
 export function createDAG (client: HTTPRPCClient, codecs: Codecs): DAGAPI {

test/interface-tests/src/dag/import.ts

@@ -44,6 +44,23 @@ async function createCar (blocks: Array<{ cid: CID, bytes: Uint8Array }>): Promi
   return out
 }
 
+async function createReadableStreamFromCar (car: AsyncIterable<Uint8Array>): Promise<ReadableStream> {
+  const stream = new ReadableStream({
+    async start (controller) {
+      try {
+        for await (const chunk of car) {
+          controller.enqueue(chunk)
+        }
+        controller.close()
+      } catch (err) {
+        controller.error(err)
+      }
+    }
+  })
+
+  return stream
+}
+
 /**
  * @typedef {import('ipfsd-ctl').Factory} Factory
  */
@@ -179,5 +196,23 @@ export function testImport (factory: Factory<KuboNode>, options: MochaConfig): v
       const result = await all(ipfs.dag.import(async function * () { yield input }()))
       expect(result).to.have.deep.nested.property('[0].root.cid', cids[0])
     })
+
+    it('should be able to import car file as a ReadableStream', async () => {
+      const blocks = await createBlocks(5)
+      const car = await createCar(blocks)
+
+      const stream = await createReadableStreamFromCar(car)
+
+      const result = await all(ipfs.dag.import(stream))
+      expect(result).to.have.lengthOf(1)
+      expect(result).to.have.deep.nested.property('[0].root.cid', blocks[0].cid)
+
+      for (const { cid } of blocks) {
+        await expect(ipfs.block.get(cid)).to.eventually.be.ok()
+      }
+
+      await expect(all(ipfs.pin.ls({ paths: blocks[0].cid }))).to.eventually.have.lengthOf(1)
+        .and.have.nested.property('[0].type', 'recursive')
+    })
   })
 }