How to use the @nteract/messaging.ofMessageType function in @nteract/messaging

getKernelInfo() {
    // Set up a receiver for kernel info
    let kernelInfo = null;
    if (!this.props.kernel) {
      return;
    }

    this.props.kernel.channels
      .pipe(
        messaging.ofMessageType("kernel_info_reply"),
        first(),
        map(msg => msg.content)
      )
      .subscribe(content => {
        kernelInfo = content;
      });

    var intervalId;
    intervalId = setInterval(() => {
      if (kernelInfo) {
        clearInterval(intervalId);
        return;
      }
      this.props.kernel.channels.next(messaging.kernelInfoRequest());
    }, 300);
  }

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

export function tooltipObservable(
  channels: Channels,
  editor: CMI,
  message: JupyterMessage
) {
  const tip$ = channels.pipe(
    childOf(message),
    ofMessageType("inspect_reply"),
    map((entry: JupyterMessage) => entry.content),
    first(),
    map(results => ({
      dict: results.data
    }))
  );
  // On subscription, send the message
  return Observable.create((observer: Observer) => {
    const subscription = tip$.subscribe(observer);
    channels.next(message);
    return subscription;
  });
}

export function codeCompleteObservable(
  channels: Channels,
  editor: Doc,
  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) => {
    const subscription = completion$.subscribe(observer);
    channels.next(message);
    return subscription;
  });
}

getKernelInfo() {
    // Set up a receiver for kernel info
    let kernelInfo = null;
    this.props.kernel.channels
      .pipe(
        messaging.ofMessageType("kernel_info_reply"),
        first(),
        map(msg => msg.content)
      )
      .subscribe(content => {
        kernelInfo = content;
      });

    var intervalId;
    intervalId = setInterval(() => {
      if (kernelInfo) {
        clearInterval(intervalId);
        return;
      }
      this.props.kernel.channels.next(messaging.kernelInfoRequest());
    }, 300);
  }

const commOpenAction$ = kernel.channels.pipe(
        ofMessageType("comm_open"),
        map(commOpenAction),
        takeUntil(
          action$.pipe(
            ofType(KILL_KERNEL_SUCCESSFUL),
            filter(
              (action: KillKernelSuccessful | NewKernelAction) =>
                action.payload.kernelRef === kernelRef
            )
          )
        )
      );

      const commMessageAction$ = kernel.channels.pipe(
        ofMessageType("comm_msg"),
        map(commMessageAction),
        takeUntil(
          action$.pipe(
            ofType(KILL_KERNEL_SUCCESSFUL),
            filter(
              (action: KillKernelSuccessful | NewKernelAction) =>
                action.payload.kernelRef === kernelRef
            )
          )
        )
      );

      return merge(
        ipywidgetsModel$(kernel, model, contentRef),
        commOpenAction$,
        commMessageAction$

export function acquireKernelInfo(
  channels: Channels,
  kernelRef: KernelRef,
  contentRef: ContentRef,
  state: AppState
) {
  const message = createMessage("kernel_info_request");

  const obs = channels.pipe(
    childOf(message),
    ofMessageType("kernel_info_reply"),
    first(),
    mergeMap(msg => {
      const c = msg.content;
      const l = c.language_info;

      const info: KernelInfo = {
        protocolVersion: c.protocol_version,
        implementation: c.implementation,
        implementationVersion: c.implementation_version,
        banner: c.banner,
        helpLinks: c.help_links,
        languageName: l.name,
        languageVersion: l.version,
        mimetype: l.mimetype,
        fileExtension: l.file_extension,
        pygmentsLexer: l.pygments_lexer,

export function commActionObservable(action: NewKernelAction) {
  const {
    payload: { kernel }
  } = action;
  const commOpenAction$ = kernel.channels.pipe(
    ofMessageType("comm_open"),
    map(commOpenAction)
  );

  const commMessageAction$ = kernel.channels.pipe(
    ofMessageType("comm_msg"),
    map(commMessageAction)
  );

  return merge(commOpenAction$, commMessageAction$).pipe(retry());
}

export function commActionObservable(action: NewKernelAction) {
  const {
    payload: { kernel }
  } = action;
  const commOpenAction$ = kernel.channels.pipe(
    ofMessageType("comm_open"),
    map(commOpenAction)
  );

  const commMessageAction$ = kernel.channels.pipe(
    ofMessageType("comm_msg"),
    map(commMessageAction)
  );

  return merge(commOpenAction$, commMessageAction$).pipe(retry());
}