it('should reset stubs used by .select', () => {
    const instance = TestBed.createComponent(TestComponent).debugElement
      .componentInstance;

    const stub1 = MockNgRedux.getSelectorStub('baz');
    stub1.next(1);
    stub1.next(2);
    stub1.complete();

    instance.baz$
      .pipe(toArray())
      .subscribe((values: number[]) => expect(values).toEqual([1, 2]));

    MockNgRedux.reset();

    // Reset should result in a new stub getting created.
    const stub2 = MockNgRedux.getSelectorStub('baz');
    expect(stub1 === stub2).toBe(false);

    stub2.next(3);
    stub2.complete();

    instance.obs$
      .pipe(toArray())
      .subscribe((values: number[]) => expect(values).toEqual([3]));
  });

  it('emits 0 initially, the right count when sources emit their own count, and ends with zero', async () => {
    const { httpService, http } = setup();

    const countA$ = new Rx.Subject<number>();
    const countB$ = new Rx.Subject<number>();
    const countC$ = new Rx.Subject<number>();
    const promise = http
      .getLoadingCount$()
      .pipe(toArray())
      .toPromise();

    http.addLoadingCount(countA$);
    http.addLoadingCount(countB$);
    http.addLoadingCount(countC$);

    countA$.next(100);
    countB$.next(10);
    countC$.next(1);
    countA$.complete();
    countB$.next(20);
    countC$.complete();
    countB$.next(0);

    httpService.stop();
    expect(await promise).toMatchSnapshot();
  });

test('observe latest', t => {
  const service = new MqttService(t.context.mqtt);
  const obs = service.observeLatest('/test1/hallo1', '/test/+').pipe(
    take(2),
    Ha4usOperators.pickEach('ts'),
    toArray()
  );
  obs.subscribe((msgs: any[]) => {
    t.deepEqual(msgs, [[1, 2], [1, 3]]);
  });

  t.context.mqtt.publish('test1/hallo1', '{"val":"helloworld","ts":1}', {
    retain: false,
    qos: 0,
  });

  t.context.mqtt.publish('test/hallo2', '{"val":"helloworld","ts":2}', {
    retain: false,
    qos: 0,
  });
  t.context.mqtt.publish('test/hallo3', '{"val":"helloworld","ts":3}', {
    retain: false,
    qos: 0,
  });
});

 test("returns an Observable with an initial state of idle", done => {
   const action$ = ActionsObservable.of({
     type: actionsModule.LAUNCH_KERNEL_SUCCESSFUL,
     payload: {
       kernel: {
         channels: of({
           header: { msg_type: "status" },
           content: { execution_state: "idle" }
         }) as Subject<any>,
         cwd: "/home/tester",
         type: "websocket"
       },
       kernelRef: "fakeKernelRef",
       contentRef: "fakeContentRef",
       selectNextKernel: false
     }
   });
   const obs = watchExecutionStateEpic(action$);
   obs.pipe(toArray()).subscribe(
     // Every action that goes through should get stuck on an array
     actions => {
       const types = actions.map(({ type }) => type);
       expect(types).toEqual([actionsModule.SET_EXECUTION_STATE]);
     },
     err => done.fail(err), // It should not error in the stream
     () => done()
   );
 });

 test("Informs about disconnected kernels, allows reconnection", async () => {
   const state$ = {
     value: {
       core: stateModule.makeStateRecord({
         kernelRef: "fake",
         entities: stateModule.makeEntitiesRecord({
           contents: stateModule.makeContentsRecord({
             byRef: Immutable.Map({
               fakeContent: stateModule.makeNotebookContentRecord()
             })
           }),
           kernels: stateModule.makeKernelsRecord({
             byRef: Immutable.Map({
               fake: stateModule.makeRemoteKernelRecord({
                 channels: null,
                 status: "not connected"
               })
             })
           })
         })
       }),
       app: {
         notificationSystem: { addNotification: jest.fn() }
       }
     }
   };
   const action$ = ActionsObservable.of(
     actions.executeCell({ id: "first", contentRef: "fakeContentRef" })
   );
   const responses = await executeCellEpic(action$, state$)
     .pipe(toArray())
     .toPromise();
   expect(responses).toEqual([]);
 });

 it("extracts all execution counts from a session", () => {
   return of(
     status("starting"),
     status("idle"),
     status("busy"),
     executeInput({
       code: "display('woo')\ndisplay('hoo')",
       execution_count: 0
     }),
     displayData({ data: { "text/plain": "woo" } }),
     displayData({ data: { "text/plain": "hoo" } }),
     executeInput({
       code: "",
       execution_count: 1
     }),
     status("idle")
   )
     .pipe(
       executionCounts(),
       toArray()
     )
     .toPromise()
     .then(arr => {
       expect(arr).toEqual([0, 1]);
     });
 });

  test("handles multiple socket routing underneath", () => {
    const shellSocket = new EventEmitter();
    const iopubSocket = new EventEmitter();
    const sockets = ({
      shell: shellSocket,
      iopub: iopubSocket
    } as any) as Sockets;

    const channels = createMainChannelFromSockets(sockets);

    const p = channels
      .pipe(
        take(2),
        toArray()
      )
      .toPromise();

    shellSocket.emit("message", { yolo: false });
    iopubSocket.emit("message", { yolo: true });

    return p.then((modifiedMessages: any) => {
      expect(modifiedMessages).toEqual([
        { channel: "shell", yolo: false },
        { channel: "iopub", yolo: true }
      ]);
    });
  });