pollForChanges(seriesImport: SeriesImportModel): Observable<SeriesImportModel> {

    let lastReceived = null;

    let seriesImportPoller = interval(1000)
      .pipe(
        flatMap(() => {
          return seriesImport.doc.reload();
        }),
        map(doc => {
          let parentAccountDoc = seriesImport.parent;
          seriesImport.init(parentAccountDoc, doc, false);
          return new SeriesImportModel(parentAccountDoc, doc);
        }),
        takeWhile((si) => {
          // HACK
          // https://github.com/ReactiveX/rxjs/issues/2420
          // TODO fixed in rxjs 6
          if (lastReceived !== null) {
            return false;
          }
          if (si.isFinished()) {
            lastReceived = si;
          }
          return true;
        })
      );

    return concat(
      observableOf(seriesImport),
      seriesImportPoller
    );
  }

 takeWhile1() {
   // emit 1,2,3,4,5
   const source = of(1, 2, 3, 4, 5);
   // allow values until value from source is greater than 4, then complete
   const example = source.pipe(takeWhile(val => val <= 4));
   // output: 1,2,3,4
   const subscribe = example.subscribe(val => console.log(val));
 }

  it('file replacements work with watch mode', async () => {
    const overrides = {
      fileReplacements: [
        {
          replace: normalize('/src/meaning.ts'),
          with: normalize('/src/meaning-too.ts'),
        },
      ],
      watch: true,
    };

    let buildCount = 0;
    let phase = 1;

    const run = await architect.scheduleTarget(target, overrides);
    await run.output.pipe(
      timeout(30000),
      tap((result) => {
        expect(result.success).toBe(true, 'build should succeed');

        const fileName = normalize('dist/main.js');
        const content = virtualFs.fileBufferToString(host.scopedSync().read(fileName));
        const has42 = /meaning\s*=\s*42/.test(content);
        buildCount++;
        switch (phase) {
          case 1:
            const has10 = /meaning\s*=\s*10/.test(content);

            if (has42 && !has10) {
              phase = 2;
              host.writeMultipleFiles({
                'src/meaning-too.ts': 'export var meaning = 84;',
              });
            }
            break;

          case 2:
            const has84 = /meaning\s*=\s*84/.test(content);

            if (has84 && !has42) {
              phase = 3;
            } else {
              // try triggering a rebuild again
              host.writeMultipleFiles({
                'src/meaning-too.ts': 'export var meaning = 84;',
              });
            }
            break;
        }
      }),
      takeWhile(() => phase < 3),
    ).toPromise().catch(() => {
      throw new Error(`stuck at phase ${phase} [builds: ${buildCount}]`);
    });

    await run.stop();
  });

 switchMap(() => merge( // then emit every 500ms, and when the navigation ends or cancels or errors. This delay must
                        // match the transition duration in the .scss file
   timer(0, 500).pipe(
     map(i => 100 - (100 / Math.pow(2, i))), // emit 0 then 50, 75, 87.5, etc.
     takeWhile(i => i < 99.95), // stop because that just triggers change detection with no visual change anymore
     takeUntil(end$), // but stop emitting every 500ms when the navigation ends or cancels or errors
     map(i => ({ value: i }))
   ),
   end$.pipe(first(), map(() => null)) // set back to null when the navigation ends or cancels or errors to hide
                                       // the progress bar
 ))

  it('file replacements work with watch mode', (done) => {
    const overrides = {
      fileReplacements: [
        {
          replace: normalize('/src/meaning.ts'),
          with: normalize('/src/meaning-too.ts'),
        },
      ],
      watch: true,
    };

    let buildCount = 0;
    let phase = 1;
    runTargetSpec(host, browserTargetSpec, overrides, 30000).pipe(
      tap((buildEvent) => expect(buildEvent.success).toBe(true, 'build should succeed')),
      tap(() => {
        const fileName = join(outputPath, 'main.js');
        const content = virtualFs.fileBufferToString(host.scopedSync().read(fileName));
        const has42 = /meaning\s*=\s*42/.test(content);
        buildCount++;
        switch (phase) {
          case 1:
            const has10 = /meaning\s*=\s*10/.test(content);

            if (has42 && !has10) {
              phase = 2;
              host.writeMultipleFiles({
                'src/meaning-too.ts': 'export var meaning = 84;',
              });
            }
            break;

          case 2:
            const has84 = /meaning\s*=\s*84/.test(content);

            if (has84 && !has42) {
              phase = 3;
            } else {
              // try triggering a rebuild again
              host.writeMultipleFiles({
                'src/meaning-too.ts': 'export var meaning = 84;',
              });
            }
            break;
        }
      }),
      takeWhile(() => phase < 3),
    ).toPromise().then(
      () => done(),
      () => done.fail(`stuck at phase ${phase} [builds: ${buildCount}]`),
    );
  });

  it('rebuilds TS worker', async () => {
    host.writeMultipleFiles(workerFiles);
    const overrides = {
      webWorkerTsConfig: 'src/tsconfig.worker.json',
      watch: true,
    };

    let buildCount = 0;
    let phase = 1;
    const workerPath = join(outputPath, '0.worker.js');
    let workerContent = '';

    const run = await architect.scheduleTarget(target, overrides);
    await run.output.pipe(
      // Wait for files to be written to disk.
      debounceTime(1000),
      tap((buildEvent) => expect(buildEvent.success).toBe(true, 'build should succeed')),
      tap(() => {
        buildCount++;
        switch (phase) {
          case 1:
            // Original worker content should be there.
            workerContent = virtualFs.fileBufferToString(host.scopedSync().read(workerPath));
            expect(workerContent).toContain('bar');
            // Change content of worker dependency.
            host.writeMultipleFiles({ 'src/app/dep.ts': `export const foo = 'baz';` });
            phase = 2;
            break;

          case 2:
            workerContent = virtualFs.fileBufferToString(host.scopedSync().read(workerPath));
            // TODO(filipesilva): Should this change? Check with Jason Miller.
            // The worker changes name with each rebuild. But sometimes it also changes from 0 to
            // 1 and then back to 0. It's hard to know where the updated content is, but it should
            // be in one of these two.
            const anotherWorkerPath = join(outputPath, '1.worker.js');
            const anotherWorkerContent = virtualFs.fileBufferToString(
              host.scopedSync().read(anotherWorkerPath));
            // Worker content should have changed.
            expect(
              workerContent.includes('baz') || anotherWorkerContent.includes('baz'),
            ).toBeTruthy('Worker bundle did not contain updated content.');
            phase = 3;
            break;
        }
      }),
      takeWhile(() => phase < 3),
    ).toPromise();
    await run.stop();
  });

 private pollRendition(nodeId: string, encoding: string, intervalSize: number = 1000, retries: number = 5) {
     let attempts = 0;
     return interval(intervalSize)
         .pipe(
             switchMap(() => this.getRendition(nodeId, encoding)),
             takeWhile((renditionEntry: RenditionEntry) => {
                 attempts += 1;
                 if (attempts > retries) {
                     return false;
                 }
                 return (renditionEntry.entry.status.toString() !== 'CREATED');
             })
         );
 }

  takeWhile2() {
    //  emit 3, 3, 3, 9, 1, 4, 5, 8, 96, 3, 66, 3, 3, 3
    const source = of(3, 3, 3, 9, 1, 4, 5, 8, 96, 3, 66, 3, 3, 3);

    //  allow values until value from source equals 3, then complete
    //  output: [3, 3, 3]
    source
      .pipe(takeWhile(it => it === 3))
      .subscribe(val => console.log('takeWhile', val));

    //  output: [3, 3, 3, 3, 3, 3, 3]
    source
      .pipe(filter(it => it === 3))
      .subscribe(val => console.log('filter', val));
  }

 return function (errors: Observable<any>): Observable<any> { //該函數主要功能就是要每隔[delayT]秒試一次，不超[attempts]次
   return errors.pipe(
     scan((acc, value) => {
       console.log(acc, value);
       acc += 1;
       if (acc < attempts) {
         return acc;
       } else {
         throw new Error(''+value);
       }
     }, 0),
     takeWhile((acc => acc < attempts)),
     delay(delayT)
   );
 };

  it('recovers from compilation failures in watch mode', async () => {
    let buildCount = 0;
    let phase = 1;

    // The current linux-based CI environments may not fully settled in regards to filesystem
    // changes from previous tests which reuse the same directory and fileset.
    // The initial delay helps mitigate false positive rebuild triggers in such scenarios.
    const { run } = await timer(1000).pipe(
      switchMap(() => architect.scheduleTarget(karmaTargetSpec, { watch: true })),
      switchMap(run => run.output.pipe(map(output => ({ run, output })))),
      debounceTime(500),
      tap(({ output }) => {
        buildCount += 1;
        switch (phase) {
          case 1:
            // Karma run should succeed.
            // Add a compilation error.
            expect(output.success).toBe(true);
            // Add an syntax error to a non-main file.
            host.appendToFile('src/app/app.component.spec.ts', `]]]`);
            phase = 2;
            break;

          case 2:
            // Karma run should fail due to compilation error. Fix it.
            expect(output.success).toBe(false);
            host.replaceInFile('src/app/app.component.spec.ts', `]]]`, '');
            phase = 3;
            break;

          case 3:
            // Karma run should succeed again.
            expect(output.success).toBe(true);
            phase = 4;
            break;
        }
      }),
      takeWhile(() => phase < 4),
      catchError((_, caught) => {
        fail(`stuck at phase ${phase} [builds: ${buildCount}]`);

        return caught;
      }),
    ).toPromise();

    await run.stop();
  });

  it('should NOT handle hot inner empty', (done: MochaDone) => {
    const source = of(1, 2, 3, 4, 5, 6, 7, 8, 9);
    const closing = EMPTY;
    const TOO_MANY_INVOCATIONS = 30;

    source.pipe(
      bufferWhen(() => closing),
      takeWhile((val: any, index: number) => index < TOO_MANY_INVOCATIONS)
    ).subscribe((val: any) => {
      expect(Array.isArray(val)).to.be.true;
      expect(val.length).to.equal(0);
    }, (err: any) => {
      done(new Error('should not be called'));
    }, () => {
      done();
    });
  });

  switchMap4() {
    const countdownSeconds = 10;
    const setHTML = id => val => (document.getElementById(id).innerHTML = val);
    const pauseButton = document.getElementById('pause');
    const resumeButton = document.getElementById('resume');
    const interval$ = interval(1000).pipe(mapTo(-1));

    const pause$ = fromEvent(pauseButton, 'click').pipe(mapTo(false));
    const resume$ = fromEvent(resumeButton, 'click').pipe(mapTo(true));

    const timer$ = merge(pause$, resume$)
      .pipe(
        startWith(true),
        switchMap(val => (val ? interval$ : EMPTY)),
        scan((acc, curr) => (curr ? curr + acc : acc), countdownSeconds),
        takeWhile(v => v >= 0)
      )
      .subscribe(setHTML('remaining'));
  }

  it('recovers from compilation failures in watch mode', (done) => {
    const overrides = { watch: true };
    let buildCount = 0;
    let phase = 1;

    runTargetSpec(host, karmaTargetSpec, overrides, DefaultTimeout * 3).pipe(
      debounceTime(500),
      tap((buildEvent) => {
        buildCount += 1;
        switch (phase) {
          case 1:
            // Karma run should succeed.
            // Add a compilation error.
            expect(buildEvent.success).toBe(true);
            // Add an syntax error to a non-main file.
            host.appendToFile('src/app/app.component.spec.ts', `]]]`);
            phase = 2;
            break;

          case 2:
            // Karma run should fail due to compilation error. Fix it.
            expect(buildEvent.success).toBe(false);
            host.replaceInFile('src/app/app.component.spec.ts', `]]]`, '');
            phase = 3;
            break;

          case 3:
            // Karma run should succeed again.
            expect(buildEvent.success).toBe(true);
            phase = 4;
            break;
        }
      }),
      takeWhile(() => phase < 4),
    ).toPromise().then(
      () => done(),
      () => done.fail(`stuck at phase ${phase} [builds: ${buildCount}]`),
    );
  });

  it('should stop listening to a synchronous observable when unsubscribed', () => {
    const sideEffects: number[] = [];
    const synchronousObservable = concat(
      defer(() => {
        sideEffects.push(1);
        return of(1);
      }),
      defer(() => {
        sideEffects.push(2);
        return of(2);
      }),
      defer(() => {
        sideEffects.push(3);
        return of(3);
      })
    );

    of(null).pipe(
      exhaustMap(() => synchronousObservable),
      takeWhile((x) => x != 2) // unsubscribe at the second side-effect
    ).subscribe(() => { /* noop */ });

    expect(sideEffects).to.deep.equal([1, 2]);
  });

  it('should stop listening to a synchronous observable when unsubscribed', () => {
    const sideEffects: number[] = [];
    const synchronousObservable = concat(
      defer(() => {
        sideEffects.push(1);
        return of(1);
      }),
      defer(() => {
        sideEffects.push(2);
        return of(2);
      }),
      defer(() => {
        sideEffects.push(3);
        return of(3);
      })
    );

    throwError(new Error('Some error')).pipe(
      onErrorResumeNext(synchronousObservable),
      takeWhile((x) => x != 2) // unsubscribe at the second side-effect
    ).subscribe(() => { /* noop */ });

    expect(sideEffects).to.deep.equal([1, 2]);
  });

export function takeWhile<T>(this: Observable<T>, predicate: (value: T, index: number) => boolean): Observable<T> {
  return higherOrder(predicate)(this);
}

  it('rebuilds on TS file changes', (done) => {
    const goldenValueFiles: { [path: string]: string } = {
      'src/app/app.module.ts': `
        import { BrowserModule } from '@angular/platform-browser';
        import { NgModule } from '@angular/core';

        import { AppComponent } from './app.component';

        @NgModule({
          declarations: [
            AppComponent
          ],
          imports: [
            BrowserModule
          ],
          providers: [],
          bootstrap: [AppComponent]
        })
        export class AppModule { }

        console.log('$$_E2E_GOLDEN_VALUE_1');
        export let X = '$$_E2E_GOLDEN_VALUE_2';
      `,
      'src/main.ts': `
        import { enableProdMode } from '@angular/core';
        import { platformBrowserDynamic } from '@angular/platform-browser-dynamic';

        import { AppModule } from './app/app.module';
        import { environment } from './environments/environment';

        if (environment.production) {
          enableProdMode();
        }

        platformBrowserDynamic().bootstrapModule(AppModule);

        import * as m from './app/app.module';
        console.log(m.X);
        console.log('$$_E2E_GOLDEN_VALUE_3');
      `,
    };

    const overrides = { watch: true };

    let buildCount = 0;
    let phase = 1;
    runTargetSpec(host, browserTargetSpec, overrides, DefaultTimeout * 3).pipe(
      tap((buildEvent) => expect(buildEvent.success).toBe(true, 'build should succeed')),
      tap(() => {
        buildCount++;
        const hasLazyChunk = host.scopedSync().exists(join(outputPath, 'lazy-lazy-module.js'));
        switch (phase) {
          case 1:
            // No lazy chunk should exist.
            if (!hasLazyChunk) {
              phase = 2;
              host.writeMultipleFiles({ ...lazyModuleFiles, ...lazyModuleImport });
            }
            break;

          case 2:
            // A lazy chunk should have been with the filename.
            if (hasLazyChunk) {
              phase = 3;
              host.writeMultipleFiles(goldenValueFiles);
            }
            break;

          case 3:
            // The golden values should be present and in the right order.
            const re = new RegExp(
              /\$\$_E2E_GOLDEN_VALUE_1(.|\n|\r)*/.source
              + /\$\$_E2E_GOLDEN_VALUE_2(.|\n|\r)*/.source
              + /\$\$_E2E_GOLDEN_VALUE_3/.source,
            );
            const fileName = './dist/main.js';
            const content = virtualFs.fileBufferToString(
              host.scopedSync().read(normalize(fileName)),
            );

            if (re.test(content)) {
              phase = 4;
            }
            break;
        }
      }),
      takeWhile(() => phase < 4),
      ).toPromise().then(
        () => done(),
        () => done.fail(`stuck at phase ${phase} [builds: ${buildCount}]`),
      );
  });

 observable1 = takeZone1.run(() => {
   return observable1.pipe(takeWhile((val: any) => {
     expect(Zone.current.name).toEqual(takeZone1.name);
     return val < 2;
   }));
 });

it('should support a user-defined type guard', () => {
  const o = of('foo').pipe(takeWhile((s): s is 'foo' => true)); // $ExpectType Observable<"foo">
});

//.........这里部分代码省略.........

        return observableConcat(
          observableOf(...neededActions),
          status.isFull ? observableOf(EVENTS.fullBuffer(bufferType)) : EMPTY
        );
      }

      if (!currentSegmentRequest) {
        log.debug("Buffer: start downloading queue.", bufferType);
        downloadQueue = neededSegments;
        startQueue$.next(); // restart the queue
      } else if (currentSegmentRequest.segment.id !== mostNeededSegment.segment.id) {
        log.debug("Buffer: restart download queue.", bufferType);
        downloadQueue = neededSegments;
        startQueue$.next(); // restart the queue
      } else if (currentSegmentRequest.priority !== mostNeededSegment.priority) {
        log.debug("Buffer: update request priority.", bufferType);
        const { request$ } = currentSegmentRequest;
        segmentFetcher.updatePriority(request$, mostNeededSegment.priority);
        currentSegmentRequest.priority = mostNeededSegment.priority;
      } else {
        log.debug("Buffer: update downloading queue", bufferType);

        // Update the previous queue to be all needed segments but the first one,
        // for which a request is already pending
        downloadQueue = neededSegments.slice().splice(1, neededSegments.length);
      }

      return observableConcat(
        observableOf(...neededActions),
        observableOf(EVENTS.activeBuffer(bufferType))
      );
    }),
    takeWhile((e) : e is IBufferNeededActions|IBufferStateFull|IBufferStateActive =>
      e.type !== "terminated"
    )
  );

  // Buffer Queue:
  //   - download every segments queued sequentially
  //   - append them to the SourceBuffer
  const bufferQueue$ = startQueue$.pipe(
    switchMap(() => downloadQueue.length ? loadSegmentsFromQueue() : EMPTY),
    mergeMap(appendSegment)
  );

  return observableMerge(status$, bufferQueue$).pipe(share());

  /**
   * Request every Segment in the ``downloadQueue`` on subscription.
   * Emit the data of a segment when a request succeeded.
   *
   * Important side-effects:
   *   - Mutates `currentSegmentRequest` when doing and finishing a request.
   *   - Will emit from finishedDownloadQueue$ Subject after it's done.
   * @returns {Observable}
   */
  function loadSegmentsFromQueue() : Observable<ILoadedSegmentObject<T>> {
    const requestNextSegment$ : Observable<ILoadedSegmentObject<T>> =
      observableDefer(() => {
        const currentNeededSegment = downloadQueue.shift();
        if (currentNeededSegment == null) {
          nextTick(() => { finishedDownloadQueue$.next(); });
          return EMPTY;
        }