<commit_before>// @flow
import {
  createMessage,
  createExecuteRequest,
  ofMessageType,
  childOf,
  updatedOutputs,
  outputs,
  payloads,
  kernelStatuses,
  executionCounts
} from "@nteract/messaging";

import { Observable } from "rxjs/Observable";
import { of } from "rxjs/observable/of";
import { from } from "rxjs/observable/from";
import { merge } from "rxjs/observable/merge";
import { empty } from "rxjs/observable/empty";
import { _throw } from "rxjs/observable/throw";

import type { ContentRef } from "../state/refs";

import {
  groupBy,
  filter,
  scan,
  map,
  mapTo,
  switchMap,
  mergeMap,
  startWith,
  mergeAll,
  takeUntil,
  catchError,
  concatMap,
  tap
} from "rxjs/operators";

import { ofType } from "redux-observable";

import * as actions from "../actions";
import * as actionTypes from "../actionTypes";
import * as selectors from "../selectors";

import type { ActionsObservable } from "redux-observable";

import type {
  NewKernelAction,
  ExecuteCell,
  ExecuteFocusedCell,
  ExecuteAllCells,
  ExecuteAllCellsBelow,
  ExecuteCanceled,
  RemoveCell
} from "../actionTypes";

const Immutable = require("immutable");

/**
 * Observe all the reactions to running code for cell with id.
 *
 * @param {Subject} channels - The standard channels specified in the Jupyter
 * specification.
 * @param {String} id - Universally Unique Identifier of cell to be executed.
 * @param {String} code - Source code to be executed.
 * @return {Observable<Action>} updatedOutputs - It returns an observable with
 * a stream of events that need to happen after a cell has been executed.
 */
export function executeCellStream(
  channels: Channels,
  id: string,
  message: ExecuteRequest,
  contentRef: ContentRef
) {
  if (!channels || !channels.pipe) {
    return _throw(new Error("kernel not connected"));
  }

  const executeRequest = message;

  // All the streams intended for all frontends
  const cellMessages = channels.pipe(childOf(executeRequest));

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

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

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

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

    // All actions for new outputs
    cellMessages.pipe(
      outputs(),
      map(output => actions.appendOutput({ id, output, contentRef }))
    ),
<commit_msg>share source observable of cell messages with subs<commit_after>// @flow
import {
  createMessage,
  createExecuteRequest,
  ofMessageType,
  childOf,
  updatedOutputs,
  outputs,
  payloads,
  kernelStatuses,
  executionCounts
} from "@nteract/messaging";

import { Observable } from "rxjs/Observable";
import { of } from "rxjs/observable/of";
import { from } from "rxjs/observable/from";
import { merge } from "rxjs/observable/merge";
import { empty } from "rxjs/observable/empty";
import { _throw } from "rxjs/observable/throw";

import type { ContentRef } from "../state/refs";

import {
  groupBy,
  filter,
  scan,
  map,
  mapTo,
  switchMap,
  mergeMap,
  startWith,
  mergeAll,
  takeUntil,
  catchError,
  concatMap,
  tap,
  share
} from "rxjs/operators";

import { ofType } from "redux-observable";

import * as actions from "../actions";
import * as actionTypes from "../actionTypes";
import * as selectors from "../selectors";

import type { ActionsObservable } from "redux-observable";

import type {
  NewKernelAction,
  ExecuteCell,
  ExecuteFocusedCell,
  ExecuteAllCells,
  ExecuteAllCellsBelow,
  ExecuteCanceled,
  RemoveCell
} from "../actionTypes";

const Immutable = require("immutable");

/**
 * Observe all the reactions to running code for cell with id.
 *
 * @param {Subject} channels - The standard channels specified in the Jupyter
 * specification.
 * @param {String} id - Universally Unique Identifier of cell to be executed.
 * @param {String} code - Source code to be executed.
 * @return {Observable<Action>} updatedOutputs - It returns an observable with
 * a stream of events that need to happen after a cell has been executed.
 */
export function executeCellStream(
  channels: Channels,
  id: string,
  message: ExecuteRequest,
  contentRef: ContentRef
) {
  if (!channels || !channels.pipe) {
    return _throw(new Error("kernel not connected"));
  }

  const executeRequest = message;

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

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

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

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

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

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

{'fraction_non_alphanumeric': 0.1121731289449955, 'fraction_numerical': 0.0, 'mean_word_length': 3.7240204429301533, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7201666', 'n_tokens_mistral': 1663, 'n_tokens_neox': 1617, 'n_words': 507}