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 acquireKernelInfo(
  channels: Channels,
  kernelRef: KernelRef,
  contentRef: ContentRef
) {
  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,
        codemirrorMode: l.codemirror_mode,
        nbconvertExporter: l.nbconvert_exporter
      };

      return of(
        // The original action we were using
        actions.setLanguageInfo({
          langInfo: msg.content.language_info,
          kernelRef,
          contentRef
        }),
        actions.setKernelInfo({
          kernelRef,
          info
        })
      );
    })
  );

  return Observable.create((observer: Observer<any>) => {
    const subscription = obs.subscribe(observer);
    channels.next(message);
    return subscription;
  });
}

 switchMap((action: NewKernelAction) =>
   action.payload.kernel.channels.pipe(
     ofMessageType("update_display_data"),
     map((msg: JupyterMessage) =>
       actions.updateDisplay({
         content: msg.content,
         contentRef: action.payload.contentRef
       })
     ),
     catchError(error =>
       of(
         actions.updateDisplayFailed({
           error,
           contentRef: action.payload.contentRef
         })
       )
     )
   )

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<any>) => {
    const subscription = tip$.subscribe(observer);
    channels.next(message);
    return subscription;
  });
}

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

export function acquireKernelInfo(
  channels: Channels,
  kernelRef: KernelRef,
  contentRef: ContentRef
) {
  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,
        codemirrorMode: l.codemirror_mode,
        nbconvertExporter: l.nbconvert_exporter
      };

      let result;
      if (!c.protocol_version.startsWith("5")) {
        result = [
          actions.launchKernelFailed({
            kernelRef,
            contentRef,
            error: new Error(
              "The kernel that you are attempting to launch does not support the latest version (v5) of the messaging protocol."
            )
          })
        ];
      } else {
        result = [
          // The original action we were using
          actions.setLanguageInfo({
            langInfo: msg.content.language_info,
            kernelRef,
            contentRef
          }),
          actions.setKernelInfo({
            kernelRef,
            info
          })
        ];
      }

      return of(...result);
    })
  );

  return Observable.create((observer: Observer<any>) => {
    const subscription = obs.subscribe(observer);
    channels.next(message);
    return subscription;
  });
}

export function executeCellStream(
  channels: Channels,
  id: string,
  message: ExecuteRequest,
  contentRef: ContentRef
) {
  if (!channels || !channels.pipe) {
    return throwError(new Error("kernel not connected"));
  }

  const executeRequest = message;

  // All the streams intended for all frontends
  const cellMessages = channels.pipe(
    childOf(executeRequest),
    share()
  ) as Observable<JupyterMessage>;

  // All the payload streams, intended for one user
  const payloadStream = cellMessages.pipe(payloads());

  const cellAction$ = merge(
    payloadStream.pipe(
      map((payload: PayloadMessage) =>
        actions.acceptPayloadMessage({ id, payload, contentRef })
      )
    ),

    // All actions for updating cell status
    cellMessages.pipe(
      kernelStatuses() as any,
      map((status: string) =>
        actions.updateCellStatus({ id, status, contentRef })
      )
    ),

    // Update the input numbering: `[ ]`
    cellMessages.pipe(
      executionCounts() as any,
      map((ct: number) =>
        actions.updateCellExecutionCount({ id, value: ct, contentRef })
      )
    ),

    // All actions for new outputs
    cellMessages.pipe(
      outputs() as any,
      map((output: Output) => actions.appendOutput({ id, output, contentRef }))
    ),

    // clear_output display message
    cellMessages.pipe(
      ofMessageType("clear_output") as any,
      mapTo(actions.clearOutputs({ id, contentRef }))
    )
  );

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

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

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