test("cleans up the outputs, pagers, and status", () => {
    const notebook = appendCellToNotebook(emptyNotebook, emptyCodeCell);
    const id = notebook.get("cellOrder").first();

    const initialState = makeDocumentRecord({
      filename: "test.ipynb",
      notebook,
      cellPagers: Immutable.Map({
        // Hokey data, we're just expecting it to be cleared
        id: Immutable.List(["a", "b"])
      }),
      transient: Immutable.Map({
        cellMap: Immutable.Map({
          id: Immutable.Map({
            status: "idle"
          })
        })
      })
    });

    const state = reducers(
      initialState,
      actions.sendExecuteRequest({ id, message: {} })
    );

    expect(state.getIn(["transient", "cellMap", id, "status"])).toEqual(
      "queued"
    );

    expect(state.getIn(["cellPagers", id])).toEqual(Immutable.List());
  });

 test("errors if the kernel is not connected in create", done => {
   const frontendToShell = new Subject();
   const shellToFrontend = new Subject();
   const mockShell = Subject.create(frontendToShell, shellToFrontend);
   const channels = mockShell;
   const state$ = {
     value: {
       core: stateModule.makeStateRecord({
         kernelRef: "fake",
         entities: stateModule.makeEntitiesRecord({
           kernels: stateModule.makeKernelsRecord({
             byRef: Immutable.Map({
               fake: stateModule.makeRemoteKernelRecord({
                 channels,
                 status: "not connected"
               })
             })
           })
         })
       }),
       app: {
         notificationSystem: { addNotification: jest.fn() }
       }
     }
   };
   const action$ = ActionsObservable.of(
     actions.sendExecuteRequest({
       id: "id",
       message: "test",
       contentRef: "fakeContentRef"
     })
   );
   const observable = createExecuteCellStream(
     action$,
     state$.value,
     createMessage("execute_request"),
     "id",
     "fakeContentRef"
   );
   observable.pipe(toArray()).subscribe(
     actions => {
       const errors = actions.map(({ payload: { error } }) =>
         error.toString()
       );
       expect(errors).toEqual(["Error: Kernel not connected!"]);
       done();
     },
     err => done.fail(err)
   );
 });

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