catchError((error, source) =>
   merge(
     of(
       actions.executeFailed({
         error,
         contentRef: action.payload.contentRef
       })
     ),
     source
   )

 catchError((error: Error, source) => {
   // Either we ensure that all errors are caught when the action.payload.contentRef
   // is in scope or we make this be a generic ERROR
   return merge(
     of(
       actions.executeFailed({
         error
       })
     ),
     source
   );
 })

export function createExecuteCellStream(
  action$: ActionsObservable<
    | ExecuteCanceled
    | DeleteCell
    | LaunchKernelAction
    | LaunchKernelByNameAction
    | KillKernelAction
    | ExecuteCell
    | ExecuteFocusedCell
  >,
  state: any,
  message: ExecuteRequest,
  id: string,
  contentRef: ContentRef
): Observable<any> {
  const kernel = selectors.currentKernel(state);

  const channels = kernel ? kernel.channels : null;

  const kernelConnected =
    kernel &&
    channels &&
    !(kernel.status === "starting" || kernel.status === "not connected");

  if (!kernelConnected || !channels) {
    return of(
      actions.executeFailed({
        error: new Error("Kernel not connected!"),
        contentRef
      })
    );
  }

  const cellStream = executeCellStream(channels, id, message, contentRef).pipe(
    takeUntil(
      merge(
        action$.pipe(
          ofType(actions.EXECUTE_CANCELED, actions.DELETE_CELL),
          filter(
            (
              action:
                | actions.ExecuteCanceled
                | actions.DeleteCell
                | actions.LaunchKernelAction
                | actions.LaunchKernelByNameAction
                | actions.KillKernelAction
                | actions.ExecuteCell
                | actions.ExecuteFocusedCell
            ) => (action as actions.ExecuteCanceled).payload.id === id
          )
        ),
        action$.pipe(
          ofType(
            actions.LAUNCH_KERNEL,
            actions.LAUNCH_KERNEL_BY_NAME,
            actions.KILL_KERNEL
          )
        )
      )
    )
  );

  return merge(
    // We make sure to propagate back to "ourselves" the actual message
    // that we sent to the kernel with the sendExecuteRequest action
    of(actions.sendExecuteRequest({ id, message, contentRef })),
    // Merging it in with the actual stream
    cellStream
  );
}