<commit_before>function produceStream(StreamClass, PropertyClass) {
  return function() {
    return new StreamClass(this, arguments);
  };
}
function produceProperty(StreamClass, PropertyClass) {
  return function() {
    return new PropertyClass(this, arguments);
  };
}



// .toProperty()

withOneSource('toProperty', {
  _init: function(args) {
    if (args.length > 0) {
      this._send(VALUE, args[0]);
    }
  }
}, {propertyMethod: produceProperty, streamMethod: produceProperty});





// .changes()

withOneSource('changes', {
  _handleValue: function(x, isCurrent) {
    if (!isCurrent) {
      this._send(VALUE, x);
    }
  },
  _handleError: function(x, isCurrent) {
    if (!isCurrent) {
      this._send(ERROR, x);
    }
  }
}, {
  streamMethod: function() {
    return function() {
      return this;
    };
  },
  propertyMethod: produceStream
});




// .withHandler()

withOneSource('withHandler', {
  _init: function(args) {
    this._handler = args[0];
    this._forcedCurrent = false;
    var $ = this;
    this._emitter = {
      emit: function(x) {
        $._send(VALUE, x, $._forcedCurrent);
      },
      error: function(x) {
        $._send(ERROR, x, $._forcedCurrent);
      },
      end: function() {
        $._send(END, null, $._forcedCurrent);
      },
      emitEvent: function(e) {
        $._send(e.type, e.value, $._forcedCurrent);
      }
    };
  },
  _free: function() {
    this._handler = null;
    this._emitter = null;
  },
  _handleAny: function(event) {
    this._forcedCurrent = event.current;
    this._handler(this._emitter, event);
    this._forcedCurrent = false;
  }
});




// .flatten(fn)

withOneSource('flatten', {
  _init: function(args) {
    this._fn = args[0] ? args[0] : id;
  },
  _free: function() {
    this._fn = null;
  },
  _handleValue: function(x, isCurrent) {
    var xs = this._fn(x);
    for (var i = 0; i < xs.length; i++) {
      this._send(VALUE, xs[i], isCurrent);
    }
  }
});







// .transduce(transducer)

function xformForObs(obs) {
  return {
    step: function(res, input) {
      obs._send(VALUE, input, obs._forcedCurrent);
      return null;
    },
    result: function(res) {
      obs._send(END, null, obs._forcedCurrent);
      return null;
    }
  };
}

withOneSource('transduce', {
  _init: function(args) {
    this._xform = args[0](xformForObs(this));
  },
  _free: function() {
    this._xform = null;
  },
  _handleValue: function(x, isCurrent) {
    this._forcedCurrent = isCurrent;
    if (this._xform.step(null, x) !== null) {
      this._xform.result(null);
    }
    this._forcedCurrent = false;
  },
  _handleEnd: function(__, isCurrent) {
    this._forcedCurrent = isCurrent;
    this._xform.result(null);
    this._forcedCurrent = false;
  }
});








// .map(fn)
<commit_msg>fix .transduce method to support new protocol (see #78)<commit_after>function produceStream(StreamClass, PropertyClass) {
  return function() {
    return new StreamClass(this, arguments);
  };
}
function produceProperty(StreamClass, PropertyClass) {
  return function() {
    return new PropertyClass(this, arguments);
  };
}



// .toProperty()

withOneSource('toProperty', {
  _init: function(args) {
    if (args.length > 0) {
      this._send(VALUE, args[0]);
    }
  }
}, {propertyMethod: produceProperty, streamMethod: produceProperty});





// .changes()

withOneSource('changes', {
  _handleValue: function(x, isCurrent) {
    if (!isCurrent) {
      this._send(VALUE, x);
    }
  },
  _handleError: function(x, isCurrent) {
    if (!isCurrent) {
      this._send(ERROR, x);
    }
  }
}, {
  streamMethod: function() {
    return function() {
      return this;
    };
  },
  propertyMethod: produceStream
});




// .withHandler()

withOneSource('withHandler', {
  _init: function(args) {
    this._handler = args[0];
    this._forcedCurrent = false;
    var $ = this;
    this._emitter = {
      emit: function(x) {
        $._send(VALUE, x, $._forcedCurrent);
      },
      error: function(x) {
        $._send(ERROR, x, $._forcedCurrent);
      },
      end: function() {
        $._send(END, null, $._forcedCurrent);
      },
      emitEvent: function(e) {
        $._send(e.type, e.value, $._forcedCurrent);
      }
    };
  },
  _free: function() {
    this._handler = null;
    this._emitter = null;
  },
  _handleAny: function(event) {
    this._forcedCurrent = event.current;
    this._handler(this._emitter, event);
    this._forcedCurrent = false;
  }
});




// .flatten(fn)

withOneSource('flatten', {
  _init: function(args) {
    this._fn = args[0] ? args[0] : id;
  },
  _free: function() {
    this._fn = null;
  },
  _handleValue: function(x, isCurrent) {
    var xs = this._fn(x);
    for (var i = 0; i < xs.length; i++) {
      this._send(VALUE, xs[i], isCurrent);
    }
  }
});







// .transduce(transducer)

var TRANSFORM_METHODS_OLD = {
  step: 'step',
  result: 'result'
};

var TRANSFORM_METHODS_NEW = {
  step: '@@transducer/step',
  result: '@@transducer/result'
};


function xformForObs(obs) {
  function step(res, input) {
    obs._send(VALUE, input, obs._forcedCurrent);
    return null;
  }
  function result(res) {
    obs._send(END, null, obs._forcedCurrent);
    return null;
  }
  return {
    step: step,
    result: result,
    '@@transducer/step': step,
    '@@transducer/result': result
  };
}

withOneSource('transduce', {
  _init: function(args) {
    var xf = args[0](xformForObs(this));
    if (isFn(xf[TRANSFORM_METHODS_NEW.step]) && isFn(xf[TRANSFORM_METHODS_NEW.result])) {
      this._transformMethods = TRANSFORM_METHODS_NEW;
    } else if (isFn(xf[TRANSFORM_METHODS_OLD.step]) && isFn(xf[TRANSFORM_METHODS_OLD.result])) {
      this._transformMethods = TRANSFORM_METHODS_OLD;
    } else {
      throw new Error('Unsuported transducers protocol');
    }
    this._xform = xf;
  },
  _free: function() {
    this._xform = null;
    this._transformMethods = null;
  },
  _handleValue: function(x, isCurrent) {
    this._forcedCurrent = isCurrent;
    if (this._xform[this._transformMethods.step](null, x) !== null) {
      this._xform[this._transformMethods.result](null);
    }
    this._forcedCurrent = false;
  },
  _handleEnd: function(__, isCurrent) {
    this._forcedCurrent = isCurrent;
    this._xform[this._transformMethods.result](null);
    this._forcedCurrent = false;
  }
});








// .map(fn)

{'fraction_non_alphanumeric': 0.15714959886739027, 'fraction_numerical': 0.002516910492370615, 'mean_word_length': 2.8982219497240957, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, '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': '5728541', 'n_tokens_mistral': 2248, 'n_tokens_neox': 2166, 'n_words': 404}