feat: default to calculating speed over a range of time

BREAKING CHANGES:
- `new StreamSpeed(timeUnit)` - `new StreamSpeed({ timeUnit })`
- Speed is now calculated based on total data read over a range of
  time, defaults to 1sec. Previously there were 2 speeds, current speed
  and average speed. Current speed was calculated based on amount of
  data received in latest read over the time difference between latest
  read and last read.
- `speed` event now only emits one speed.

Author: fent

README.md

@@ -15,7 +15,7 @@ let ss = new StreamSpeed();
 ss.add(rs);
 
 // Listen for events emitted by streamspeed on the given stream.
-ss.on('speed', (speed, avgSpeed) => {
+ss.on('speed', (speed) => {
   console.log('Reading at', speed, 'bytes per second');
 });
 ```
@@ -28,16 +28,18 @@ group.add(stream1);
 group.add(stream2);
 group.add(stream3);
 
-group.on('speed', (speed, avg) => {
+group.on('speed', (speed) => {
   console.log('now reading at', speed, 'bps');
 });
 ```
 
 ![example img](http://i.imgur.com/y47Sc.png)
 
 # API
-### new StreamSpeed([timeUnit])
-A group that can be used to watch several streams. Will emit `speed` events. `timeUnit` defaults to `1000` for speed per second.
+### new StreamSpeed([options])
+A group that can be used to watch several streams. Will emit `speed` events. `options` can have the following properties,
+- `timeUnit` - Defaults to `1000` for speed per second. If you want another unit such as per hour, use `1000 * 60 * 60`.
+- `range` - The time in ms to use to calculate the speed over. Defaults to 1000. The longer this is, the longer it'll speed will ramp up and down until it stabilizes for a big readstream. The shorter it is, the more responsive it is to sudden speed changes.
 
 ### StreamSpeed#add(stream)
 Adds stream to group.
@@ -51,9 +53,6 @@ Returns a list of all streams in the group.
 ### StreamSpeed#speed
 Curent speed.
 
-### StreamSpeed#avg
-Current average speed.
-
 ### StreamSpeed.toHuman(bytes, options)
 Convenient method to convert `bytes` to a human readable string.
 
@@ -66,9 +65,8 @@ StreamSpeed.toHuman(1024 * 1024 * 20.5, { precision: 3 });   // 20.50MB
 
 ### Event: 'speed'
 * `number` - Speed at which streams in the group are being read.
-* `number` - Average speed.
 
-Will be emitted after the second time a stream is read and only if there is a change in speed.
+Will be emitted every time a stream is read and only if there is a change in speed.
 
 
 # Install

lib/index.js

@@ -9,14 +9,18 @@ module.exports = class StreamSpeed extends EventEmitter {
    *
    * @constructor
    * @extends {EventEmitter}
-   * @param {number?} per The time unit speed will be measured in.
+   * @param {Object?} options
+   * @param {number?} options.timeUnit The time unit speed will be measured in.
+   * @param {number?} options.range Time in ms to calculate speed over.
    */
-  constructor(per) {
+  constructor(options = {}) {
     super();
-    this.per = per || 1000;
+    this.options = Object.assign({
+      timeUnit: 1000,
+      range: 1000,
+    }, options);
     this._streams = [];
     this.speed = 0;
-    this.avg = 0;
   }
 
 
@@ -25,11 +29,9 @@ module.exports = class StreamSpeed extends EventEmitter {
    *
    * @param {Object} meta
    * @param {number} speed
-   * @param {number} avg
    */
-  _update(meta, speed, avg) {
+  _update(meta, speed) {
     meta.speed = speed;
-    meta.avg = avg;
 
     this._streams.forEach((m) => {
       // Skip own stream, streams that haven't started,
@@ -40,12 +42,10 @@ module.exports = class StreamSpeed extends EventEmitter {
 
       // Add other streams' speeds to total.
       speed += m.speed;
-      avg += m.avg;
     });
 
     this.speed = speed;
-    this.avg = avg;
-    this.emit('speed', speed, avg);
+    this.emit('speed', speed);
   }
 
 
@@ -87,11 +87,10 @@ module.exports = class StreamSpeed extends EventEmitter {
     const meta = {
       stream  : origstream,
       speed   : 0,
-      avg     : 0,
       cleanup,
     };
     this._streams.push(meta);
-    const reader = new Speedometer(this.per);
+    const reader = new Speedometer(this.options);
     const stream = origstream.pipe(new PassThrough());
     const onUpdate = this._update.bind(this, meta);
 

lib/speedometer.js

@@ -3,14 +3,14 @@ module.exports = class Speedometer {
    * Helps count the number of bytes per data event in streams.
    *
    * @constructor
-   * @param {number} per
+   * @param {Object?} options
+   * @param {number?} options.timeUnit
+   * @param {number?} options.range
    */
-  constructor(per) {
-    this.per = per;
-    this.starttime = null;
+  constructor(options) {
+    this.options = options;
     this.history = [];
     this.speed = 0;
-    this.avg = 0;
   }
 
 
@@ -23,49 +23,32 @@ module.exports = class Speedometer {
   update(data, callback) {
     const now = Date.now();
 
-    // Check if this is the first data event.
-    if (!this.starttime) {
-      this.starttime = now;
-      return;
-    }
-
-    let lastpoint = this.history[this.history.length - 1];
-    let currpoint;
-
-    if (now === this.starttime) {
-      // Do nothing on the rare occassion we get all the data at the start.
-      // We need two time points to measure speed.
-      return;
-
-    } else if (lastpoint && now === lastpoint.time) {
-      // If more data is read on the same ms, aggregate it.
-      let lastlastpoint = this.history[this.history.length - 2];
-      if (!lastlastpoint) { return; }
-      currpoint = lastpoint;
-      currpoint.speed += data.length / (now - lastlastpoint.time);
-
-    } else {
-      // Compare now to last time.
-      let lasttime = lastpoint ? lastpoint.time : this.starttime;
-      this.history.push(currpoint = { 
-        speed: data.length / (now - lasttime),
-        time: now,
+    if (this.history.length) {
+      // Remove old data events.
+      let index = this.history.findIndex(item => {
+        return item.time > now - this.options.range;
       });
+      this.history = index > -1 ? this.history.slice(index) : [];
     }
 
-    let speed = Math.round(currpoint.speed * this.per);
+    this.history.push({ 
+      speed: data.length,
+      time: now,
+    });
+
+    // Get total data emitted in `range` time period.
+    let totaldata = this.history.reduce((sum, point) => point.speed + sum, 0);
 
-    // Get average speed.
-    let total = this.history.reduce((sum, point) => point.speed + sum, 0);
-    const avg = Math.round((total * this.per) / this.history.length);
+    let speed = Math.round(
+      totaldata / this.options.range * this.options.timeUnit
+    );
 
-    const change = this.speed !== speed || this.avg !== avg;
+    const change = this.speed !== speed;
     this.speed = speed;
-    this.avg = avg;
 
-    // Only emit event if there is a change in speed or avg.
+    // Only emit event if there is a change in speed.
     if (change) {
-      callback(speed, avg);
+      callback(speed);
     }
   }
 };

test/group-test.js

@@ -41,11 +41,13 @@ describe('Create a group and write to it', () => {
     group.on('speed', spy);
 
     // Write at 500 B/s on 2 streams, 1000 B/s in total.
-    s1.interval(100, 6, 200, { end: true, skipTick: true });
-    s2.interval(100, 6, 200, { end: true });
+    s1.interval(500, 6, 1000, { end: true, skipTick: true });
+    s2.interval(500, 6, 1000, { end: true });
+
     s2.on('finish', () => {
-      assert.deepEqual(spy.firstCall.args, [500, 500]);
-      assert.deepEqual(spy.secondCall.args, [1000, 1000]);
+      assert.equal(spy.callCount, 2);
+      assert.deepEqual(spy.getCall(0).args, [500]);
+      assert.deepEqual(spy.getCall(1).args, [1000]);
       done();
     });
   });

test/mockstream.js

@@ -2,8 +2,8 @@ const PassThrough = require('stream').PassThrough;
 const sinon       = require('sinon');
 
 let clock;
-before(() => { clock = sinon.useFakeTimers(); });
-after(() => { clock.restore(); });
+beforeEach(() => { clock = sinon.useFakeTimers(); });
+afterEach(() => { clock.restore(); });
 
 
 /**
@@ -26,29 +26,47 @@ module.exports = class Mock extends PassThrough {
       options.amountPerInterval = options.amountPerInterval || 1;
       let i = 0;
 
-      const iid = setInterval(() => {
-        let amountSoFar = 0;
-        const write = () => {
-          if (++amountSoFar < options.amountPerInterval) {
-            this.write(Buffer.alloc(length), write);
-          } else {
-            this.write(Buffer.alloc(length));
-            if (++i === amount) {
-              clearInterval(iid);
-              if (options.end) {
-                process.nextTick(this.end.bind(this));
-              }
-              resolve();
-            } else if (!options.skipTick) {
-              process.nextTick(clock.tick.bind(clock, interval));
-            }
+      const iid = setInterval(async () => {
+        for (let soFar = 0; soFar < options.amountPerInterval; soFar++) {
+          await this.writeSize(length);
+        }
+        if (++i === amount) {
+          clearInterval(iid);
+          if (options.end) {
+            process.nextTick(this.end.bind(this));
           }
-        };
-        write();
+          resolve();
+        } else if (!options.skipTick) {
+          process.nextTick(clock.tick.bind(clock, interval));
+        }
       }, interval);
       if (!options.skipTick) {
-        process.nextTick(clock.tick.bind(clock, interval));
+        if (interval) {
+          process.nextTick(clock.tick.bind(clock, interval));
+        } else {
+          process.nextTick(clock.next);
+        }
       }
     });
   }
+
+  /**
+   * An async function that writes arbitrary data to the stream.
+   *
+   * @param {number} length
+   */
+  writeSize(length) {
+    return new Promise((resolve) => {
+      this.write(Buffer.alloc(length), resolve);
+    });
+  }
+
+  /**
+   * Async version of `setTimeout`.
+   *
+   * @param {number} ms
+   */
+  static timeout(ms) {
+    clock.tick(ms);
+  }
 };

test/remove-test.js

@@ -32,24 +32,26 @@ describe('Immediately remove a stream', () => {
 
 describe('Unwatch after several writes', () => {
   it('Emits no events after calling remove', async () => {
-    const ss = new StreamSpeed(1);
+    const ss = new StreamSpeed();
     const s = new MockStream();
     ss.add(s);
     const spy = sinon.spy();
     ss.on('speed', spy);
 
     // Write at 1 bps for 0.5 seconds.
     await s.interval(100, 2, 200, { amountPerInterval: 2 });
-    assert.equal(spy.callCount, 1);
+    assert.ok(spy.called);
+    let callCount = spy.callCount;
+
     ss.remove(s);
     await s.interval(100, 1, 200, { amountPerInterval: 2 });
-    assert.equal(spy.callCount, 1);
+    assert.equal(spy.callCount, callCount);
   });
 
   describe('Try removing stream again', () => {
     it('Throws error', () => {
       assert.throws(() => {
-        const ss = new StreamSpeed(1);
+        const ss = new StreamSpeed({ timeUnit: 1 });
         const s = new MockStream();
         ss.add(s);
         ss.remove(s);