<commit_before>/**
 * <p>
 *  Constructs a ProAct.Stream. The stream is a simple {@link ProAct.Actor}, without state.
 * </p>
 * <p>
 *  The streams are ment to emit values, events, changes and can be plugged into another actor.
 *  For example you can connect many streams, to merge them and to divide them, to plug them into properties.
 * </p>
 * <p>
 *  The reactive environment consists of the properties and the objects containing them, but
 *  the outside world is not reactive. It is possible to use the ProAct.Streams as connections from the
 *  outside world to the reactive environment.
 * </p>
 * <p>
 *    The transformations can be used to change the events or values emitetted.
 * </p>
 * <p>
 *  ProAct.Stream is part of the streams module of ProAct.js.
 * </p>
 *
 * @class ProAct.Stream
 * @extends ProAct.Actor
 * @param {String} queueName
 *      The name of the queue all the updates should be pushed to.
 *      <p>
 *        If this parameter is null/undefined the default queue of
 *        {@link ProAct.flow} is used.
 *      </p>
 *      <p>
 *        If this parameter is not a string it is used as the
 *        <i>source</i>.
 *      </p>
 * @param {ProAct.Actor} source
 *      A default source of the stream, can be null.
 * @param {Array} transforms
 *      A list of transformation to be used on all incoming chages.
 */
function Stream (queueName, source, transforms) {
  if (queueName && !P.U.isString(queueName)) {
    transforms = source;
    source = queueName;
    queueName = null;
  }
  P.Actor.call(this, queueName, transforms);

  this.sourceNumber = 0;

  if (source) {
    this.into(source);
  }
}
ProAct.Stream = ProAct.S = Stream;

ProAct.Stream.prototype = P.U.ex(Object.create(P.Actor.prototype), {

  /**
   * Reference to the constructor of this object.
   *
   * @memberof ProAct.Stream
   * @instance
   * @constant
   * @type {Object}
   * @default ProAct.Stream
   */
  constructor: ProAct.Stream,

  /**
   * Creates the <i>event</i> to be send to the listeners on update.
   * <p>
   *  Streams don't create new events by default, the event is the source.
   * </p>
   *
   * @memberof ProAct.Stream
   * @instance
   * @method makeEvent
   * @param {ProAct.Event} source
   *      The source event of the event. It can be null
   * @return {ProAct.Event}
   *      The event.
   */
  makeEvent: function (source) {
    return source;
  },

  /**
   * Creates the <i>listener</i> of this stream.
   * <p>
   *  The listener of the stream just calls the method {@link ProAct.Stream#trigger} with the incoming event/value.
   * </p>
   *
   * @memberof ProAct.Stream
   * @instance
   * @method makeListener
   * @return {Object}
   *      The <i>listener of this stream</i>.
   */
  makeListener: function () {
    if (!this.listener) {
      var stream = this;
      this.listener = function (event) {
        stream.trigger(event, true);
      };
    }

    return this.listener;
  },

  /**
   * Creates the <i>error listener</i> of this stream.
   * <p>
   *  The listener just calls {@link ProAct.Stream#triggerErr} of <i>this</i> with the incoming error.
   * </p>
   *
   * @memberof ProAct.Stream
   * @instance
   * @method makeErrListener
   * @return {Object}
   *      The <i>error listener of this stream</i>.
   */
  makeErrListener: function () {
    if (!this.errListener) {
      var stream = this;
      this.errListener = function (error) {
        stream.triggerErr(error);
      };
    }

    return this.errListener;
  },

  /**
   * Creates the <i>closing listener</i> of this stream.
   * <p>
   *  The listener just calls {@link ProAct.Stream#triggerClose} of <i>this</i> with the incoming closing data.
   * </p>
   *
   * @memberof ProAct.Stream
   * @instance
   * @method makeCloseListener
   * @return {Object}
   *      The <i>closing listener of this stream</i>.
   */
  makeCloseListener: function () {
    if (!this.closeListener) {
      var stream = this;
      this.closeListener = function (error) {
        stream.triggerClose(error);
      };
    }

    return this.closeListener;
  },

  /**
   * Defers a ProAct.Actor listener.
   * <p>
   *  For streams this means pushing it to active flow using {@link ProAct.Flow#push}.
   *  If the listener is object with 'property' field, it is done using {@link ProAct.Actor#defer}.
   *  That way the reactive environment is updated only once, but the streams are not part of it.
   * </p>
   *
   * @memberof ProAct.Stream
   * @instance
   * @method defer
   * @param {Object} event
   *      The event/value to pass to the listener.
   * @param {Object} listener
   *      The listener to defer. It should be a function or object defining the <i>call</i> method.
   * @return {ProAct.Actor}
   *      <i>this</i>
   * @see {@link ProAct.Actor#willUpdate}
   * @see {@link ProAct.Actor#makeListener}
   * @see {@link ProAct.flow}
   */
  defer: function (event, listener) {
    if (listener.property) {
      P.Actor.prototype.defer.call(this, event, listener);
      return;
    }
    var queueName = (listener.queueName) ? listener.queueName : this.queueName;

    if (P.U.isFunction(listener)) {
      P.flow.push(queueName, listener, [event]);
    } else {
      P.flow.push(queueName, listener, listener.call, [event]);
    }
  },

  /**
   * <p><commit_msg>Bugfix for null listeners for streams.<commit_after>/**
 * <p>
 *  Constructs a ProAct.Stream. The stream is a simple {@link ProAct.Actor}, without state.
 * </p>
 * <p>
 *  The streams are ment to emit values, events, changes and can be plugged into another actor.
 *  For example you can connect many streams, to merge them and to divide them, to plug them into properties.
 * </p>
 * <p>
 *  The reactive environment consists of the properties and the objects containing them, but
 *  the outside world is not reactive. It is possible to use the ProAct.Streams as connections from the
 *  outside world to the reactive environment.
 * </p>
 * <p>
 *    The transformations can be used to change the events or values emitetted.
 * </p>
 * <p>
 *  ProAct.Stream is part of the streams module of ProAct.js.
 * </p>
 *
 * @class ProAct.Stream
 * @extends ProAct.Actor
 * @param {String} queueName
 *      The name of the queue all the updates should be pushed to.
 *      <p>
 *        If this parameter is null/undefined the default queue of
 *        {@link ProAct.flow} is used.
 *      </p>
 *      <p>
 *        If this parameter is not a string it is used as the
 *        <i>source</i>.
 *      </p>
 * @param {ProAct.Actor} source
 *      A default source of the stream, can be null.
 * @param {Array} transforms
 *      A list of transformation to be used on all incoming chages.
 */
function Stream (queueName, source, transforms) {
  if (queueName && !P.U.isString(queueName)) {
    transforms = source;
    source = queueName;
    queueName = null;
  }
  P.Actor.call(this, queueName, transforms);

  this.sourceNumber = 0;

  if (source) {
    this.into(source);
  }
}
ProAct.Stream = ProAct.S = Stream;

ProAct.Stream.prototype = P.U.ex(Object.create(P.Actor.prototype), {

  /**
   * Reference to the constructor of this object.
   *
   * @memberof ProAct.Stream
   * @instance
   * @constant
   * @type {Object}
   * @default ProAct.Stream
   */
  constructor: ProAct.Stream,

  /**
   * Creates the <i>event</i> to be send to the listeners on update.
   * <p>
   *  Streams don't create new events by default, the event is the source.
   * </p>
   *
   * @memberof ProAct.Stream
   * @instance
   * @method makeEvent
   * @param {ProAct.Event} source
   *      The source event of the event. It can be null
   * @return {ProAct.Event}
   *      The event.
   */
  makeEvent: function (source) {
    return source;
  },

  /**
   * Creates the <i>listener</i> of this stream.
   * <p>
   *  The listener of the stream just calls the method {@link ProAct.Stream#trigger} with the incoming event/value.
   * </p>
   *
   * @memberof ProAct.Stream
   * @instance
   * @method makeListener
   * @return {Object}
   *      The <i>listener of this stream</i>.
   */
  makeListener: function () {
    if (!this.listener) {
      var stream = this;
      this.listener = function (event) {
        stream.trigger(event, true);
      };
    }

    return this.listener;
  },

  /**
   * Creates the <i>error listener</i> of this stream.
   * <p>
   *  The listener just calls {@link ProAct.Stream#triggerErr} of <i>this</i> with the incoming error.
   * </p>
   *
   * @memberof ProAct.Stream
   * @instance
   * @method makeErrListener
   * @return {Object}
   *      The <i>error listener of this stream</i>.
   */
  makeErrListener: function () {
    if (!this.errListener) {
      var stream = this;
      this.errListener = function (error) {
        stream.triggerErr(error);
      };
    }

    return this.errListener;
  },

  /**
   * Creates the <i>closing listener</i> of this stream.
   * <p>
   *  The listener just calls {@link ProAct.Stream#triggerClose} of <i>this</i> with the incoming closing data.
   * </p>
   *
   * @memberof ProAct.Stream
   * @instance
   * @method makeCloseListener
   * @return {Object}
   *      The <i>closing listener of this stream</i>.
   */
  makeCloseListener: function () {
    if (!this.closeListener) {
      var stream = this;
      this.closeListener = function (error) {
        stream.triggerClose(error);
      };
    }

    return this.closeListener;
  },

  /**
   * Defers a ProAct.Actor listener.
   * <p>
   *  For streams this means pushing it to active flow using {@link ProAct.Flow#push}.
   *  If the listener is object with 'property' field, it is done using {@link ProAct.Actor#defer}.
   *  That way the reactive environment is updated only once, but the streams are not part of it.
   * </p>
   *
   * @memberof ProAct.Stream
   * @instance
   * @method defer
   * @param {Object} event
   *      The event/value to pass to the listener.
   * @param {Object} listener
   *      The listener to defer. It should be a function or object defining the <i>call</i> method.
   * @return {ProAct.Actor}
   *      <i>this</i>
   * @see {@link ProAct.Actor#willUpdate}
   * @see {@link ProAct.Actor#makeListener}
   * @see {@link ProAct.flow}
   */
  defer: function (event, listener) {
    if (!listener) {
      return;
    }

    if (listener.property) {
      P.Actor.prototype.defer.call(this, event, listener);
      return;
    }
    var queueName = (listener.queueName) ? listener.queueName : this.queueName;

    if (P.U.isFunction(listener)) {
      P.flow.push(queueName, listener, [event]);
    } else {
      P.flow.push(queueName, listener, listener.call, [event]);
    }
  },

  /**
   * <p>
{'fraction_non_alphanumeric': 0.12313887068077535, 'fraction_numerical': 0.00018728345350688266, 'mean_word_length': 2.54228855721393, 'pattern_counts': {'":': 0, '<': 101, '<?xml version=': 0, '>': 101, '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': '2326594', 'n_tokens_mistral': 3410, 'n_tokens_neox': 3349, 'n_words': 1107}