public get(
   query: Ha4usObjectQuery,
   aTimeout: number = 500
 ): Promise<IStatefulObject[]> {
   return this.$objects
     .observe(query)
     .pipe(
       mergeMap((aObject: Ha4usObject) => {
         return this.$states.observe(aObject.topic).pipe(
           timeoutWith(100, of({})),
           take(1),
           map((state: Ha4usMessage) => {
             delete state.topic
             delete state.match
             return {
               object: aObject,
               state: state,
             }
           })
         )
       }),
       bufferTime(aTimeout),
       timeout(aTimeout * 2)
     )
     .toPromise()
 }

  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();
  });

 querySnapshots(parentObjectId: Uuid): Promise<Snapshot[]> {
     return this.querySnapshotsByParentId(parentObjectId)
         // Unsubscribe automatically after first response event arrives.
         .pipe(
             take(1),
             // Issue an Rx.TimeoutError if queryTimeoutMillis elapses without any emitted event.
             timeout(this.options["queryTimeoutMillis"]),
         )
         .toPromise();
 }

 intercept(req: HttpRequest<any>, next: HttpHandler): Observable<HttpEvent<any>> {
     let obs = of(true).pipe(
         tap(() => this.sharedService.requestStarted(req)),
         switchMap(() => next.handle(req)),
     );
     if (this.sharedService.httpTimeout) {
         obs = obs.pipe(timeout(this.sharedService.httpTimeout));
     }
     return obs.pipe(finalize(() => this.sharedService.requestFinished(req)));
 }

 const getLicense$ = observableDefer(() => {
   const getLicense = keySystem.getLicense(message, messageType);
   return (castToObservable(getLicense) as Observable<TypedArray|ArrayBuffer|null>)
     .pipe(
       timeout(10 * 1000),
       catchError((error : unknown) : never => {
         if (error instanceof TimeoutError) {
           throw new EncryptedMediaError("KEY_LOAD_TIMEOUT",
             "The license server took more than 10 seconds to respond.", false);
         }
         if (error instanceof Error) {
           throw error;
         }
         throw new Error("An error occured when calling `getLicense`.");
       })
   );
 });

    switchMap(action => {
      const filepath = action.payload.filepath;

      return forkJoin(
        readFileObservable(filepath),
        statObservable(filepath),
        // Project onto the Contents API response
        (content: Buffer, stat: fs.Stats): contents.IContent =>
          createContentsResponse(filepath, stat, content)
      ).pipe(
        // Timeout after one minute
        timeout(60 * 1000),
        map((model: contents.IContent<"notebook">) => {
          if (model.type !== "notebook") {
            throw new Error(
              "Attempted to load a non-notebook type from desktop"
            );
          }
          if (model.content === null) {
            throw new Error("No content loaded for notebook");
          }

          return actions.fetchContentFulfilled({
            filepath: model.path,
            model,
            kernelRef: action.payload.kernelRef,
            contentRef: action.payload.contentRef
          });
        }),
        catchError((err: Error) =>
          of(
            actions.fetchContentFailed({
              filepath,
              error: err,
              kernelRef: action.payload.kernelRef,
              contentRef: action.payload.contentRef
            })
          )
        )
      );
    })

export function codeCompleteObservable(
  channels: Channels,
  editor: CMI,
  message: JupyterMessage
) {
  const completion$ = channels.pipe(
    childOf(message),
    ofMessageType("complete_reply"),
    map(entry => entry.content),
    first(),
    map(expand_completions(editor)),
    timeout(15000) // Large timeout for slower languages; this is just here to make sure we eventually clean up resources
  );

  // On subscription, send the message
  return Observable.create((observer: Observer<any>) => {
    const subscription = completion$.subscribe(observer);
    channels.next(message);
    return subscription;
  });
}

  it('should show error when lazy route is invalid on watch mode AOT', async () => {
    host.writeMultipleFiles(lazyModuleFiles);
    host.writeMultipleFiles(lazyModuleImport);
    host.replaceInFile(
      'src/app/app.module.ts',
      'lazy.module#LazyModule',
      'invalid.module#LazyModule',
    );

    const logger = new TestLogger('rebuild-lazy-errors');
    const overrides = { watch: true, aot: true };
    const run = await architect.scheduleTarget(target, overrides, { logger });
    await run.output.pipe(
      timeout(15000),
      tap((buildEvent) => expect(buildEvent.success).toBe(false)),
      tap(() => {
        expect(logger.includes('Could not resolve module')).toBe(true);
        logger.clear();
        host.appendToFile('src/main.ts', ' ');
      }),
      take(2),
    ).toPromise();
    await run.stop();
  });

    concatMap((action: actions.RestartKernel | actions.NewKernelAction) => {
      const state = state$.value;
      const oldKernelRef = action.payload.kernelRef;
      const notificationSystem = selectors.notificationSystem(state);

      if (!oldKernelRef) {
        notificationSystem.addNotification({
          title: "Failure to Restart",
          message: "Unable to restart kernel, please select a new kernel.",
          dismissible: true,
          position: "tr",
          level: "error"
        });
        return empty();
      }

      const oldKernel = selectors.kernel(state, { kernelRef: oldKernelRef });

      if (!oldKernelRef || !oldKernel) {
        notificationSystem.addNotification({
          title: "Failure to Restart",
          message: "Unable to restart kernel, please select a new kernel.",
          dismissible: true,
          position: "tr",
          level: "error"
        });

        // TODO: Wow do we need to send notifications through our store for
        // consistency
        return empty();
      }

      const newKernelRef = kernelRefGenerator();
      const initiatingContentRef = action.payload.contentRef;

      // TODO: Incorporate this into each of the launchKernelByName
      //       actions...
      //       This only mirrors the old behavior of restart kernel (for now)
      notificationSystem.addNotification({
        title: "Kernel Restarting...",
        message: `Kernel ${oldKernel.kernelSpecName ||
          "unknown"} is restarting.`,
        dismissible: true,
        position: "tr",
        level: "success"
      });

      const kill = actions.killKernel({
        restarting: true,
        kernelRef: oldKernelRef
      });

      const relaunch = actions.launchKernelByName({
        kernelSpecName: oldKernel.kernelSpecName,
        cwd: oldKernel.cwd,
        kernelRef: newKernelRef,
        selectNextKernel: true,
        contentRef: initiatingContentRef
      });

      const awaitKernelReady = action$.pipe(
        ofType(actions.LAUNCH_KERNEL_SUCCESSFUL),
        filter(
          (action: actions.NewKernelAction | actions.RestartKernel) =>
            action.payload.kernelRef === newKernelRef
        ),
        take(1),
        timeout(60000), // If kernel doesn't come up within this interval we will abort follow-on actions.
        concatMap(() => {
          const restartSuccess = actions.restartKernelSuccessful({
            kernelRef: newKernelRef,
            contentRef: initiatingContentRef
          });

          if (
            (action as actions.RestartKernel).payload.outputHandling ===
            "Run All"
          ) {
            return of(
              restartSuccess,
              actions.executeAllCells({ contentRef: initiatingContentRef })
            );
          } else {
            return of(restartSuccess);
          }
        }),
        catchError(error => {
          return of(
            actions.restartKernelFailed({
              error,
              kernelRef: newKernelRef,
              contentRef: initiatingContentRef
            })
          );
        })
      );

      return merge(of(kill, relaunch), awaitKernelReady);
    })

it('should enforce types of scheduler', () => {
  const o = of('a', 'b', 'c').pipe(timeout(5, 'foo')); // $ExpectError
});

 concatMap(duration =>
   makeRequest(duration).pipe(
     timeout(2500),
     catchError(error => of(`Request timed out after: ${duration}`))
   )

  async checkForUpdates() {
    if (this.checking) {
      return;
    } else {
      this.checking = true;
    }

    try {
      await this.platform.ready();
      await this.hasDeploy();

      const hasUpdate = await Promise.race([
        this.pro.deploy().check(),
        new Promise((_, reject) => setTimeout(reject, 8000)),
      ]);

      console.log(hasUpdate);
      if (hasUpdate !== 'true') {
        return false;
      }

      switch (this.network.type) {
        // Don't update if the network type is slow (it probably costs a little bit)
        case Connection.CELL:
        case Connection.CELL_2G:
        case Connection.CELL_3G:
        case Connection.UNKNOWN:
        case Connection.NONE:
          this.network
            .onchange()
            .pipe(take(1))
            .subscribe(() => {
              this.checkForUpdates();
            });
          return false;

        case Connection.WIFI:
        case Connection.CELL_4G:
        default:
          break;
      }

      await this.pro
        .deploy()
        .download()
        .pipe(
          tap(status => console.log('Download status:', status)),
          filter(status => status === 'true' || status === 'done'),
          switchMap(() => this.pro.deploy().extract()),
          tap(status => console.log('Extract status:', status)),
          filter(status => status === 'true' || status === 'done'),
          take(1),
          timeout(30 * 1000),
        )
        .toPromise();

      if (this.settings.isDeveloper) {
        const alert = await this.alertCtrl.show({
          title: 'New update available',
          message: 'There is a new update available',
          buttons: [
            {
              text: 'Install',
              role: 'install',
            },
            {
              text: 'Postpone',
              role: 'cancel',
            },
          ],
        });

        return new Promise(resolve => {
          alert.onDidDismiss((_, role) => {
            if (role === 'install') {
              this.pro.deploy().redirect();
              resolve(true);
            } else {
              resolve(false);
            }
          });
        });
      }
      return false;
    } finally {
      this.checking = false;
    }
  }

export function timeout<T>(this: Observable<T>,
                           due: number | Date,
                           scheduler: SchedulerLike = asyncScheduler): Observable<T> {
  return higherOrder(due, scheduler)(this) as Observable<T>;
}

 observable1 = constructorZone1.run(() => {
   return of(1).pipe(timeout(10));
 });

  shutdownEpic(timeoutMs: number = 2000) {
    const request: JupyterMessage<"shutdown_request", any> = shutdownRequest({
      restart: false
    });

    // Try to make a shutdown request
    // If we don't get a response within X time, force a shutdown
    // Either way do the same cleanup
    const shutDownHandling = this.channels.pipe(
      /* Get the first response to our message request. */
      childOf(request),
      ofMessageType("shutdown_reply"),
      first(),
      // If we got a reply, great! :)
      map((msg: { content: { restart: boolean } }) => {
        return {
          status: "shutting down",
          content: msg.content
        };
      }),
      /**
       * If we don't get a response within timeoutMs, then throw an error.
       */
      timeout(timeoutMs),
      catchError(err => of({ error: err, status: "error" })),
      /**
       * Even if we don't receive a shutdown_reply from the kernel to our
       * shutdown_request, we will go forward with cleaning up the RxJS
       * subject and killing the kernel process.
       */
      mergeMap(
        async (
          event:
            | { status: string; error: Error }
            | {
                status: string;
                content: { restart: boolean };
              }
        ) => {
          // End all communication on the channels
          this.channels.complete();
          await this.shutdownProcess();

          const finalResponse: { error?: Error; status: string } = {
            status: "shutdown"
          };
          if (event.status === "error") {
            finalResponse.error = (event as {
              error: Error;
              status: string;
            }).error;
            finalResponse.status = "error";
          }

          return of(finalResponse);
        }
      ),
      catchError(err =>
        // Catch all, in case there were other errors here
        of({ error: err, status: "error" })
      )
    );

    // On subscription, send the message
    return Observable.create((observer: Observer<any>) => {
      const subscription = shutDownHandling.subscribe(observer);
      this.channels.next(request);
      return subscription;
    });
  }

 map(() =>
   buildOutput$.pipe(
     timeout(handlerDelay),
     catchError(() => Rx.of('timeout'))
   )

    concatMap((action: actions.KillKernelAction) => {
      const kernelRef = action.payload.kernelRef;
      if (!kernelRef) {
        console.warn("tried to kill a kernel without a kernelRef");
        return empty();
      }

      const kernel = selectors.kernel(state$.value, { kernelRef });

      if (!kernel) {
        // tslint:disable-next-line:no-console
        console.warn("tried to kill a kernel that doesn't exist");
        return empty();
      }

      // Ignore the action if the specified kernel is not ZMQ.
      if (kernel.type !== "zeromq") {
        return empty();
      }

      const request = shutdownRequest({ restart: false });

      // Try to make a shutdown request
      // If we don't get a response within X time, force a shutdown
      // Either way do the same cleanup
      const shutDownHandling = kernel.channels.pipe(
        childOf(request),
        ofMessageType("shutdown_reply"),
        first(),
        // If we got a reply, great! :)
        map((msg: { content: { restart: boolean } }) =>
          actions.shutdownReplySucceeded({ content: msg.content, kernelRef })
        ),
        // If we don't get a response within 2s, assume failure :(
        timeout(1000 * 2),
        catchError(err =>
          of(actions.shutdownReplyTimedOut({ error: err, kernelRef }))
        ),
        mergeMap(resultingAction => {
          // End all communication on the channels
          kernel.channels.complete();

          if (kernel.spawn) {
            killSpawn(kernel.spawn);
          }

          return merge(
            // Pass on our intermediate action (whether or not kernel ACK'd shutdown request promptly)
            of(resultingAction),
            // Indicate overall success (channels cleaned up)
            of(
              actions.killKernelSuccessful({
                kernelRef
              })
            ),
            // Inform about the state
            of(
              actions.setExecutionState({
                kernelStatus: "shutting down",
                kernelRef
              })
            )
          );
        }),
        catchError(err =>
          // Catch all, in case there were other errors here
          of(actions.killKernelFailed({ error: err, kernelRef }))
        )
      );

      // On subscription, send the message
      return new Observable(observer => {
        const subscription = shutDownHandling.subscribe(observer);
        kernel.channels.next(request);
        return subscription;
      });
    })

 intercept(
   context: ExecutionContext,
   call$: Observable<any>,
 ): Observable<any> {
   return call$.pipe(timeout(5000));
 }