Update file download pipeline #8

Author: vlad-ignatov

index.d.ts

@@ -222,6 +222,13 @@ export declare namespace BulkDataClient {
          * useful information, for example which site imported this data.
          */
         metadata?: Record<string, any>
+
+        /**
+         * Path to log file. Absolute, or relative to process CWD. If not
+         * provided, the file will be called log.ndjson and will be stored in
+         * the downloads folder.
+         */
+        file?: string
     }
 
     interface CLIOptions {
@@ -448,6 +455,8 @@ export declare namespace BulkDataClient {
          * Only needed if `destination` points to S3
          */
         awsSecretAccessKey: string
+
+        log: LoggingOptions
     }
 
     interface JWK {

src/lib/BulkDataClient.ts

@@ -674,12 +674,15 @@ class BulkDataClient extends EventEmitter
         // Just "remember" the progress values but don't emit anything yet
         download.on("progress", state => Object.assign(_state, state))
 
+        const streams: (NodeJS.ReadableStream | NodeJS.WritableStream | NodeJS.ReadWriteStream)[] = [];
+
         // Start the download (the stream will be paused though)
-        let processPipeline: Readable = await download.run({
+        let downloadStream: Readable = await download.run({
             accessToken,
             signal: this.abortController.signal,
             requestOptions: this.options.requests
-        }).catch(e => {
+        })
+        .catch(e => {
             if (e instanceof FileDownloadError) {
                 this.emit("downloadError", {
                     body: null, // Buffer
@@ -691,6 +694,8 @@ class BulkDataClient extends EventEmitter
             throw e
         });
 
+        streams.push(downloadStream)
+
         // ---------------------------------------------------------------------
         // Create an NDJSON parser to verify that every single line is valid
         // ---------------------------------------------------------------------
@@ -709,9 +714,8 @@ class BulkDataClient extends EventEmitter
             expectedCount: exportType == "output" ? file.count || -1 : -1,
             expectedResourceType
         })
-
-        processPipeline = processPipeline.pipe(parser);
-
+            
+        streams.push(parser)
 
         // ---------------------------------------------------------------------
         // Download attachments
@@ -745,7 +749,7 @@ class BulkDataClient extends EventEmitter
 
             docRefProcessor.on("attachment", () => _state.attachments! += 1)
 
-            processPipeline = processPipeline.pipe(docRefProcessor)
+            streams.push(docRefProcessor)
         }
 
 
@@ -757,19 +761,33 @@ class BulkDataClient extends EventEmitter
             _state.resources! += 1
             onProgress(_state)
         });
-        processPipeline = processPipeline.pipe(stringify);
+        streams.push(stringify)
 
 
         // ---------------------------------------------------------------------
         // Write the file to the configured destination
         // ---------------------------------------------------------------------
-        await this.writeToDestination(fileName, processPipeline, subFolder)
+        try {
+            await pipeline(streams)
+            
+        } catch (e: any) {
+            this.emit("downloadError", {
+                body   : null,
+                code   : e.code || null,
+                fileUrl: e.fileUrl || file.url,
+                message: String(e.message || "Downloading failed")
+            })
 
+            throw e
+        }
+        
         this.emit("downloadComplete", {
             fileUrl      : file.url,
             fileSize     : _state.uncompressedBytes,
             resourceCount: _state.resources!
         })
+        
+        await this.writeToDestination(fileName, stringify, subFolder)
     }
 
     /**