function getWatchHandlers(
  buildOutput$: Rx.Observable<string>,
  {
    handlerDelay = defaultHandlerDelay,
    handlerReadinessTimeout = defaultHandlerReadinessTimeout,
  }: IWatchOptions
) {
  const typescriptHandler = buildOutput$.pipe(
    first(data => data.includes('$ tsc')),
    map(() =>
      buildOutput$.pipe(first(data => data.includes('Compilation complete.')), mapTo('tsc'))
    )
  );

  const webpackHandler = buildOutput$.pipe(
    first(data => data.includes('$ webpack')),
    map(() => buildOutput$.pipe(first(data => data.includes('Chunk Names')), mapTo('webpack')))
  );

  const defaultHandler = Rx.of(undefined).pipe(
    delay(handlerReadinessTimeout),
    map(() => buildOutput$.pipe(timeout(handlerDelay), catchError(() => Rx.of('timeout'))))
  );

  return [typescriptHandler, webpackHandler, defaultHandler];
}

export async function integrationTestContext(): Promise<
    TestContext & {
        getEnvironment: () => Environment
        ready: Promise<void>
    }
> {
    const [clientTransports, serverTransports] = createMessageTransports()

    const clientController = new Controller({
        clientOptions: () => ({ createMessageTransports: () => clientTransports }),
    })
    clientController.setEnvironment(FIXTURE_ENVIRONMENT)

    // Ack all configuration updates.
    clientController.configurationUpdates.subscribe(({ resolve }) => resolve(Promise.resolve()))

    const extensionHost = createExtensionHost(
        {
            bundleURL: '',
            sourcegraphURL: 'https://example.com',
            clientApplication: 'sourcegraph',
            settingsCascade: {
                final: { a: 1 },
            },
        },
        serverTransports
    )

    // Wait for client to be ready.
    await clientController.clientEntries
        .pipe(
            filter(entries => entries.length > 0),
            first()
        )
        .toPromise()

    return {
        clientController,
        extensionHost,
        getEnvironment(): Environment {
            // This runs synchronously because the Observable's root source is a BehaviorSubject (which has an initial value).
            // Confirm it is synchronous just in case, because a bug here would be hard to diagnose.
            let value!: Environment
            let sync = false
            clientController.environment
                .pipe(first())
                .subscribe(environment => {
                    value = environment
                    sync = true
                })
                .unsubscribe()
            if (!sync) {
                throw new Error('environment is not synchronously available')
            }
            return value
        },
        ready: ready({ clientController, extensionHost }),
    }
}

  it('should allow registered filters to filter out aggTypes', async () => {
    const aggTypes = [{ name: 'count' }, { name: 'sum' }, { name: 'avg' }];
    const observable = registry.filter$(aggTypes, indexPattern, aggConfig);
    let filtered = await observable.pipe(first()).toPromise();
    expect(filtered).toEqual(aggTypes);

    registry.addFilter(() => true);
    registry.addFilter(aggType => aggType.name !== 'count');
    filtered = await observable.pipe(first()).toPromise();
    expect(filtered).toEqual([aggTypes[1], aggTypes[2]]);

    registry.addFilter(aggType => aggType.name !== 'avg');
    filtered = await observable.pipe(first()).toPromise();
    expect(filtered).toEqual([aggTypes[1]]);
  });

 private horizontalScrollGridToIndex(grid, visibleColumnIndex, callBackFunc) {
     const unpinnedIndex = this.getColumnUnpinnedIndex(visibleColumnIndex);
     grid.parentVirtDir.onChunkLoad
         .pipe(first())
         .subscribe(callBackFunc);
     grid.dataRowList.toArray()[0].virtDirRow.scrollTo(unpinnedIndex);
 }

 first3() {
   const source = from([1, 2, 3, 4, 5]);
   // no value will pass, emit default
   const example = source.pipe(first(val => val > 5, -1));
   // output: 'Nothing'
   const subscribe = example.subscribe(val => console.log(val));
 }

  it("moving the caret scrolls the pane", (done) => {
    const initial = editor.dataRoot;
    caretManager.setCaret(initial.firstChild, 0);

    // tslint:disable-next-line:no-any
    const scroller = (editor as any).scroller;

    const initialScroll = scroller.scrollTop;

    scroller.events.pipe(first()).subscribe(() => {
      // We need to wait until the scroller has fired the scroll event.
      assert.isTrue(initialScroll < scroller.scrollTop);
      const caretRect = editor.caretManager.mark.getBoundingClientRect();
      const scrollerRect = scroller.getBoundingClientRect();
      assert.equal(distFromRect(caretRect.left, caretRect.top,
                                scrollerRect.left, scrollerRect.top,
                                scrollerRect.right,
                                scrollerRect.bottom),
                   0, "caret should be in visible space");
      done();
    });

    caretManager.setCaret(initial.firstChild,
                          initial.firstChild!.childNodes.length);
  });

function getNpmConfigOption(
  option: string,
  scope?: string,
  tryWithoutScope?: boolean,
): Observable<string | undefined> {
  if (scope && tryWithoutScope) {
    return concat(
      getNpmConfigOption(option, scope),
      getNpmConfigOption(option),
    ).pipe(
      filter(result => !!result),
      defaultIfEmpty(),
      first(),
    );
  }

  const fullOption = `${scope ? scope + ':' : ''}${option}`;

  let value = npmConfigOptionCache.get(fullOption);
  if (value) {
    return value;
  }

  value = option.startsWith('_')
      ? getOptionFromNpmRc(fullOption)
      : getOptionFromNpmCli(fullOption);

  npmConfigOptionCache.set(fullOption, value);

  return value;
}

  return moduleRefPromise.then((moduleRef) => {
    const transitionId = moduleRef.injector.get(ÉľTRANSITION_ID, null);
    if (!transitionId) {
      throw new Error(
          `renderModule[Factory]() requires the use of BrowserModule.withServerTransition() to ensure
the server-rendered app can be properly bootstrapped into a client app.`);
    }
    const applicationRef: ApplicationRef = moduleRef.injector.get(ApplicationRef);
    return applicationRef.isStable.pipe((first((isStable: boolean) => isStable)))
        .toPromise()
        .then(() => {
          const platformState = platform.injector.get(PlatformState);

          // Run any BEFORE_APP_SERIALIZED callbacks just before rendering to string.
          const callbacks = moduleRef.injector.get(BEFORE_APP_SERIALIZED, null);
          if (callbacks) {
            for (const callback of callbacks) {
              try {
                callback();
              } catch (e) {
                // Ignore exceptions.
                console.warn('Ignoring BEFORE_APP_SERIALIZED Exception: ', e);
              }
            }
          }

          const output = platformState.renderToString();
          platform.destroy();
          return output;
        });
  });

  test('returns data and admin client observables as a part of the contract', async () => {
    const mockAdminClusterClientInstance = { close: jest.fn() };
    const mockDataClusterClientInstance = { close: jest.fn() };
    MockClusterClient.mockImplementationOnce(
      () => mockAdminClusterClientInstance
    ).mockImplementationOnce(() => mockDataClusterClientInstance);

    const setupContract = await elasticsearchService.setup();

    const [esConfig, adminClient, dataClient] = await combineLatest(
      setupContract.legacy.config$,
      setupContract.adminClient$,
      setupContract.dataClient$
    )
      .pipe(first())
      .toPromise();

    expect(adminClient).toBe(mockAdminClusterClientInstance);
    expect(dataClient).toBe(mockDataClusterClientInstance);

    expect(MockClusterClient).toHaveBeenCalledTimes(2);
    expect(MockClusterClient).toHaveBeenNthCalledWith(
      1,
      esConfig,
      expect.objectContaining({ context: ['elasticsearch', 'admin'] })
    );
    expect(MockClusterClient).toHaveBeenNthCalledWith(
      2,
      esConfig,
      expect.objectContaining({ context: ['elasticsearch', 'data'] })
    );

    expect(mockAdminClusterClientInstance.close).not.toHaveBeenCalled();
    expect(mockDataClusterClientInstance.close).not.toHaveBeenCalled();
  });

  return <T extends {}>(
    schematic: FileSystemSchematicDescription,
    options: T,
    context?: FileSystemSchematicContext,
  ): Observable<T> => {
    // Prevent a schematic from changing the options object by making a copy of it.
    options = deepCopy(options);

    const withPrompts = context ? context.interactive : true;

    if (schematic.schema && schematic.schemaJson) {
      // Make a deep copy of options.
      return registry
        .compile(schematic.schemaJson)
        .pipe(
          mergeMap(validator => validator(options, { withPrompts })),
          first(),
          map(result => {
            if (!result.success) {
              throw new InvalidInputOptions(options, result.errors || []);
            }

            return options;
          }),
        );
    }

    return observableOf(options);
  };