'use strict';

var through = require('through2');

/**
 * Cleans up an array of objects based on a specific object property.
 *
 * It is mainly designed to be used as an `Array.filter` callback.
 *
 * ```js
 * cleanEmptyElement('y', { y: 0 });
 * // true
 *
 * cleanEmptyElement('y', { y: null });
 * // false
 * ```
 *
 * @param valueProperty {String}
 * @param item {Object}
 * @returns {Boolean}
 */
function cleanEmptyElement(valueProperty, item){
  return item[valueProperty] !== null;
}

/**
 * Remaps an initial item to a common working structure.
 *
 * It is intended to be bound to a slayer instance.
 *
 * @this {Slayer}
 * @param originalData {Array} The original array of data
 * @param y {Number} The value number detected as being a spike
 * @param i {Number} The index location of `y` in `originalData`
 * @returns {{x: *, y: Number}}
 */
function objectMapper(originalData, y, i){
  return this.getItem({
    x: this.getValueX(originalData[i], i),
    y: y
  }, originalData[i], i);
}

/**
 * Processes an array of data and calls an `onComplete` callback with `error` and `spikes` parameters.
 *
 * @example
 ```js
 slayer()
 .fromArray(arrayData, function(err, spikes){
    if (err){
      console.error(err);
      return;
    }

    console.log(spikes);   // { x: 4, y: 12, id: '21232f297a57a5a743894a0e4a801fc3' }
  });
 ```
 *
 * @api
 * @name Slayer.prototype.fromArray
 * @this {Slayer}
 * @param data {Array.<Object|Number>}
 * @param onComplete {Function}
 */
function createReadStream(bufferSize){
  var buffer = [];
  var offset = 0;
  var self = this;
  var bufferSize = bufferSize > 0 ? bufferSize : 50;

  return through.obj(function(chunk, enc, callback){
    buffer.push(chunk);

    if (buffer.length >= bufferSize) {
      var spikes = buffer
        .map(self.getValueY.bind(self))
        .map(self.filterDataItem.bind(self))
        .map(self.algorithm.bind(self, self.config.minPeakDistance))
        .map(objectMapper.bind(self, buffer))
        .filter(cleanEmptyElement.bind(null, self.config.transformedValueProperty))
        .map(function(d) {
          d.x = d.x + offset;
          return d;
        })
        .map(function(d){
          this.push(d);
        }, this)

      buffer = buffer.slice(1);
      offset++;
    }

    callback();
  });
}

module.exports = createReadStream;

@@ -991,24 +991,362 @@
 %5Bi%5D, i);%0A%7D%0A%0A
+function incrementOffset(xProperty, offset, d) %7B%0A  d%5BxProperty%5D = d%5BxProperty%5D + offset;%0A%0A  return d;%0A%7D%0A%0Afunction pushToStream(d)%7B%0A  this.push(d);%0A%0A  return d;%0A%7D%0A%0Afunction cleanEmitted(xProperty, cache, d) %7B%0A  if (cache.indexOf(d%5BxProperty%5D) === -1) %7B%0A    cache.push(d%5BxProperty%5D);%0A%0A    return true;%0A  %7D%0A  else %7B%0A    return false;%0A  %7D%0A%7D%0A%0A
 /**%0A * Proce
@@ -1842,26 +1842,16 @@
 dStream(
-bufferSize
 )%7B%0A  var
@@ -1864,16 +1864,41 @@
 r = %5B%5D;%0A
+  var emittedCache = %5B%5D;%0A
   var of
@@ -1936,50 +1936,21 @@
 var 
-bufferSize = bufferSize %3E 0 ? bufferSize :
+BUFFER_SIZE =
  50;
@@ -2058,18 +2058,19 @@
  %3E= 
-bufferSize
+BUFFER_SIZE
 ) %7B%0A
@@ -2396,51 +2396,48 @@
 map(
-function(d) %7B%0A          d.x = d.x +
+incrementOffset.bind(self, 'x',
  offset
-;
+))
 %0A   
@@ -2445,98 +2445,149 @@
     
-  return d;%0A        %7D)%0A        .map(function(d)%7B%0A          this.push(d
+.filter(cleanEmitted.bind(self, 'x', emittedCache))%0A        .map(pushToStream, this
 );%0A
+%0A
       
-  %7D, this)%0A
+emittedCache = emittedCache.slice(-BUFFER_SIZE);
 %0A   

Implement known emitted spikes in streaming mode