patch fs methods as macrotask, add test cases of fs watcher

lib/common/utils.ts

@@ -481,3 +481,37 @@ export function patchMethod(
   }
   return delegate;
 }
+
+export interface MacroTaskMeta extends TaskData {
+  name: string;
+  target: any;
+  callbackIndex: number;
+  args: any[];
+}
+
+// TODO: support cancel task later if necessary
+export function patchMacroTask(
+    obj: any, funcName: string, metaCreator: (self: any, args: any[]) => MacroTaskMeta) {
+  let setNative = null;
+
+  function scheduleTask(task: Task) {
+    const data = <MacroTaskMeta>task.data;
+    data.args[data.callbackIndex] = function() {
+      task.invoke.apply(this, arguments);
+    };
+    setNative.apply(data.target, data.args);
+    return task;
+  }
+
+  setNative = patchMethod(obj, funcName, (delegate: Function) => function(self: any, args: any[]) {
+    const meta = metaCreator(self, args);
+    if (meta.callbackIndex >= 0 && typeof args[meta.callbackIndex] === 'function') {
+      const task = Zone.current.scheduleMacroTask(
+          meta.name, args[meta.callbackIndex], meta, scheduleTask, null);
+      return task;
+    } else {
+      // cause an error by calling it directly.
+      return delegate.apply(self, args);
+    }
+  });
+}

lib/node/fs.ts

@@ -6,16 +6,17 @@
  * found in the LICENSE file at https://angular.io/license
  */
 
-import {bindArguments} from '../common/utils';
+import {patchMacroTask} from '../common/utils';
 
 let fs;
 try {
   fs = require('fs');
 } catch (err) {
 }
 
-// TODO(alxhub): Patch `watch` and `unwatchFile`.
-const TO_PATCH = [
+// watch, watchFile, unwatchFile has been patched
+// because EventEmitter has been patched
+const TO_PATCH_MACROTASK_METHODS = [
   'access',  'appendFile', 'chmod',    'chown',    'close',     'exists',    'fchmod',
   'fchown',  'fdatasync',  'fstat',    'fsync',    'ftruncate', 'futimes',   'lchmod',
   'lchown',  'link',       'lstat',    'mkdir',    'mkdtemp',   'open',      'read',
@@ -24,11 +25,15 @@ const TO_PATCH = [
 ];
 
 if (fs) {
-  TO_PATCH.filter(name => !!fs[name] && typeof fs[name] === 'function').forEach(name => {
-    fs[name] = ((delegate: Function) => {
-      return function() {
-        return delegate.apply(this, bindArguments(<any>arguments, 'fs.' + name));
-      };
-    })(fs[name]);
-  });
+  TO_PATCH_MACROTASK_METHODS.filter(name => !!fs[name] && typeof fs[name] === 'function')
+      .forEach(name => {
+        patchMacroTask(fs, name, (self: any, args: any[]) => {
+          return {
+            name: 'fs.' + name,
+            args: args,
+            callbackIndex: args.length > 0 ? args.length - 1 : -1,
+            target: self
+          };
+        });
+      });
 }

test/node/fs.spec.ts

@@ -6,15 +6,83 @@
  * found in the LICENSE file at https://angular.io/license
  */
 
-import {exists} from 'fs';
+import {exists, unlink, unwatchFile, watch, watchFile, writeFile} from 'fs';
 
 describe('nodejs file system', () => {
-  it('has patched exists()', (done) => {
-    const zoneA = Zone.current.fork({name: 'A'});
-    zoneA.run(() => {
-      exists('testfile', (_) => {
-        expect(Zone.current.name).toBe(zoneA.name);
-        done();
+  describe('async method patch test', () => {
+    it('has patched exists()', (done) => {
+      const zoneA = Zone.current.fork({name: 'A'});
+      zoneA.run(() => {
+        exists('testfile', (_) => {
+          expect(Zone.current.name).toBe(zoneA.name);
+          done();
+        });
+      });
+    });
+
+    it('has patched exists as macroTask', (done) => {
+      const zoneASpec = {
+        name: 'A',
+        onScheduleTask: (delegate, currentZone, targetZone, task): Task => {
+          return delegate.scheduleTask(targetZone, task);
+        }
+      };
+      const zoneA = Zone.current.fork(zoneASpec);
+      spyOn(zoneASpec, 'onScheduleTask').and.callThrough();
+      zoneA.run(() => {
+        exists('testfile', (_) => {
+          expect(zoneASpec.onScheduleTask).toHaveBeenCalled();
+          done();
+        });
+      });
+    });
+  });
+
+  describe('watcher related methods test', () => {
+    const zoneASpec = {
+      name: 'A',
+      onScheduleTask: (delegate, currentZone, targetZone, task): Task => {
+        return delegate.scheduleTask(targetZone, task);
+      }
+    };
+
+    it('fs.watch has been patched as eventTask', (done) => {
+      spyOn(zoneASpec, 'onScheduleTask').and.callThrough();
+      const zoneA = Zone.current.fork(zoneASpec);
+      zoneA.run(() => {
+        writeFile('testfile', 'test content', () => {
+          const watcher = watch('testfile', (eventType, filename) => {
+            expect(filename).toEqual('testfile');
+            expect(eventType).toEqual('change');
+            expect(zoneASpec.onScheduleTask).toHaveBeenCalled();
+            expect(Zone.current.name).toBe('A');
+            watcher.close();
+            unlink('testfile', () => {
+              done();
+            });
+          });
+          writeFile('testfile', 'test new content');
+        });
+      });
+    });
+
+    it('fs.watchFile has been patched as eventTask', (done) => {
+      spyOn(zoneASpec, 'onScheduleTask').and.callThrough();
+      const zoneA = Zone.current.fork(zoneASpec);
+      zoneA.run(() => {
+        writeFile('testfile', 'test content', () => {
+          watchFile('testfile', {persistent: false, interval: 1000}, (curr, prev) => {
+            expect(curr.size).toBe(16);
+            expect(prev.size).toBe(12);
+            expect(zoneASpec.onScheduleTask).toHaveBeenCalled();
+            expect(Zone.current.name).toBe('A');
+            unwatchFile('testfile');
+            unlink('testfile', () => {
+              done();
+            });
+          });
+          writeFile('testfile', 'test new content');
+        });
       });
     });
   });