export async function scanDelete(options: Options) {
  const { directory, regularExpressions, concurrency = 20 } = options;

  assertAbsolute(directory);

  // get an observable of absolute paths within a directory
  const getChildPath$ = (path: string) =>
    getReadDir$(path).pipe(
      mergeAll(),
      map((name: string) => join(path, name))
    );

  // get an observable of all paths to be deleted, by starting with the arg
  // and recursively iterating through all children, unless a child matches
  // one of the supplied regular expressions
  const getPathsToDelete$ = (path: string): Rx.Observable<string> => {
    if (regularExpressions.some(re => re.test(path))) {
      return Rx.of(path);
    }

    return getStat$(path).pipe(
      mergeMap(stat => (stat.isDirectory() ? getChildPath$(path) : Rx.EMPTY)),
      mergeMap(getPathsToDelete$)
    );
  };

  return await Rx.of(directory)
    .pipe(
      mergeMap(getPathsToDelete$),
      mergeMap(async path => await del(path), concurrency),
      count()
    )
    .toPromise();
}

  asDiagram('count')('should count the values of an observable', () => {
    const source = hot('--a--b--c--|');
    const subs =       '^          !';
    const expected =   '-----------(x|)';

    expectObservable(source.pipe(count())).toBe(expected, {x: 3});
    expectSubscriptions(source.subscriptions).toBe(subs);
  });

  it("handles both the legacy and current arguments for ofMessageType", () => {
    from(([
      { header: { msg_type: "stream" } },
      { header: { msg_type: "error" } },
      { header: { msg_type: "status" } },
      { header: { msg_type: "stream" } },
      { header: { msg_type: "status" } }
    ] as any[]) as JupyterMessage[])
      .pipe(
        ofMessageType(["stream", "status"]),
        tap(val => {
          expect(
            val.header.msg_type === "stream" || val.header.msg_type === "status"
          );
        }),
        map(entry => entry.header.msg_type),
        count()
      )
      .toPromise()
      .then(val => {
        expect(val).toEqual(4);
      });

    from(([
      { header: { msg_type: "stream" } },
      { header: { msg_type: "status" } },
      { header: { msg_type: "error" } },
      { header: { msg_type: "stream" } },
      { header: { msg_type: "status" } }
    ] as any[]) as JupyterMessage[])
      .pipe(
        // Note the lack of array brackets on the arguments
        ofMessageType("stream", "status"),
        tap(val => {
          expect(
            val.header.msg_type === "stream" || val.header.msg_type === "status"
          );
        }),
        map(entry => entry.header.msg_type),
        count()
      )
      .toPromise()
      .then(val => {
        expect(val).toEqual(4);
      });
  });

  it('should be never when source doesn\'t complete', () => {
    const e1 = hot('--x--^--y--');
    const e1subs =      '^     ';
    const expected =    '------';

    expectObservable(e1.pipe(count())).toBe(expected);
    expectSubscriptions(e1.subscriptions).toBe(e1subs);
  });