Adding initial fix for bug issue Azure#1967

src/blob/generated/ExpressMiddlewareFactory.ts

@@ -92,13 +92,15 @@ export default class ExpressMiddlewareFactory extends MiddlewareFactory {
       handlers,
       this.logger
     );
-    return (req: Request, res: Response, next: NextFunction) => {
+      return (req: Request, res: Response, next: NextFunction) => {
       const request = new ExpressRequestAdapter(req);
       const response = new ExpressResponseAdapter(res);
+      let newContext = new Context(res.locals, this.contextPath, request, response);
       handlerMiddlewareFactory.createHandlerMiddleware()(
-        new Context(res.locals, this.contextPath, request, response),
+        newContext,
         next
       );
+      res.on("close", () => handlers.blobHandler.cleanUpBlob(newContext));
     };
   }
 

src/blob/generated/handlers/IBlobHandler.ts

@@ -14,6 +14,7 @@ import Context from "../Context";
 
 export default interface IBlobHandler {
   download(options: Models.BlobDownloadOptionalParams, context: Context): Promise<Models.BlobDownloadResponse>;
+  cleanUpBlob(context: Context): Promise<void>;
   getProperties(options: Models.BlobGetPropertiesOptionalParams, context: Context): Promise<Models.BlobGetPropertiesResponse>;
   delete(options: Models.BlobDeleteMethodOptionalParams, context: Context): Promise<Models.BlobDeleteResponse>;
   undelete(options: Models.BlobUndeleteOptionalParams, context: Context): Promise<Models.BlobUndeleteResponse>;

src/blob/handlers/BlobHandler.ts

@@ -91,6 +91,15 @@ export default class BlobHandler extends BaseHandler implements IBlobHandler {
     }
   }
 
+    /**
+   * Clean up file handles for blob to prevent leak.
+   *
+   * @memberof BlobHandler
+   */
+    public async cleanUpBlob(context: Context): Promise<void> {
+        this.extentStore.cleanStreams(context.contextId);
+    }
+
   /**
    * Get blob properties.
    *

src/common/persistence/FSExtentStore.ts

@@ -5,6 +5,7 @@ import {
   fdatasync,
   mkdir,
   open,
+  ReadStream,
   stat,
   unlink
 } from "fs";
@@ -304,6 +305,8 @@ export default class FSExtentStore implements IExtentStore {
     return this.appendQueue.operate(op, contextId);
   }
 
+    private streams: Map<string, ReadStream[]> = new Map<string, ReadStream[]>();
+
   /**
    * Read data from persistency layer according to the given IExtentChunk.
    *
@@ -353,12 +356,42 @@ export default class FSExtentStore implements IExtentStore {
             contextId
           );
         });
+
+        if (contextId != null) {
+            let existingStreams = this.streams.get(contextId);
+            if (existingStreams == null) {
+                let newStreamsArray: ReadStream[] = [stream];
+                this.streams.set(contextId, newStreamsArray);
+            }
+            else {
+                existingStreams.push(stream);
+                this.streams.set(contextId, existingStreams);
+            }
+        }       
+
         resolve(stream);
       });
 
     return this.readQueue.operate(op, contextId);
   }
 
+    public async cleanStreams(contextId?: string): Promise<void> {
+        this.logger.verbose(
+            "FSExtentStore:cleanStreams() Response object closed unexpectedly, cleaning up after streams",
+            contextId
+        );
+
+        if (contextId != null) {
+            let streamsToCleanup = this.streams.get(contextId);
+
+            if (streamsToCleanup != null) {
+                for (const stream of streamsToCleanup) {
+                    stream.destroy();
+                }
+            }
+        }
+    }
+
   /**
    * Merge several extent chunks to a ReadableStream according to the offset and count.
    *

src/common/persistence/IExtentStore.ts

@@ -59,6 +59,11 @@ export default interface IExtentStore extends IDataStore, ICleaner {
     contextId?: string
   ): Promise<NodeJS.ReadableStream>;
 
+   /**
+   * Clean up file handles from an extent when a response object is closed
+   */
+   cleanStreams(contextId?: string): Promise<void>;
+
   /**
    * Merge several extent chunks to a ReadableStream according to the offset and count.
    *