add throttleImmediate

spec/specs/throttle.coffee

@@ -7,6 +7,7 @@ expect = require("chai").expect
   expectStreamTimings,
   expectPropertyEvents,
   error,
+  semiunstable,
   lessThan,
   map,
   fromArray
@@ -28,6 +29,22 @@ describe "EventStream.throttle(delay)", ->
   it "toString", ->
     expect(Bacon.never().throttle(1).toString()).to.equal("Bacon.never().throttle(1)")
 
+describe "EventStream.throttleImmediate(delay)", ->
+  describe "outputs first event immediately, then ignores events for given amount of milliseconds", ->
+    expectStreamTimings(
+      -> series(2, [1, 2, 3, 4]).throttleImmediate(t(3))
+      [[2, 1], [5, 2], [8, 3], [11, 4]], semiunstable)
+  describe "works with synchronous source", ->
+    expectStreamEvents(
+      -> fromArray([1, 2, 3, 4]).throttleImmediate(t(3))
+      [1, 4], semiunstable)
+  describe "does not emit double events with synchronous source", ->
+    expectStreamEvents(
+      -> fromArray([1]).throttleImmediate(t(1))
+      [1], semiunstable)
+  it "toString", ->
+    expect(Bacon.never().throttleImmediate(1).toString()).to.equal("Bacon.never().throttleImmediate(1)")
+
 describe "Property.throttle", ->
   describe "throttles changes, but not initial value", ->
     expectPropertyEvents(
@@ -39,3 +56,15 @@ describe "Property.throttle", ->
       [1])
   it "toString", ->
     expect(Bacon.constant(0).throttle(1).toString()).to.equal("Bacon.constant(0).throttle(1)")
+
+describe "Property.throttleImmediate", ->
+  describe "throttles changes, but not initial value", ->
+    expectPropertyEvents(
+      -> series(1, [1,2,3]).toProperty(0).throttleImmediate(t(4))
+      [0,3], semiunstable)
+  describe "works with Bacon.once", ->
+    expectPropertyEvents(
+      -> once(1).toProperty().throttleImmediate(1)
+      [1], semiunstable)
+  it "toString", ->
+    expect(Bacon.constant(0).throttleImmediate(1).toString()).to.equal("Bacon.constant(0).throttleImmediate(1)")

src/throttle.js

@@ -3,7 +3,30 @@ import "./delaychanges";
 import "./map";
 import Observable from "./observable";
 import { Desc } from "./describe";
+import Bacon from "./core";
 
 Observable.prototype.throttle = function (delay) {
-  return this.delayChanges(new Desc(this, "throttle", [delay]), (changes) => changes.bufferWithTime(delay).map((values) => values[values.length - 1]));
+  return this.delayChanges(new Desc(this, "throttle", [delay]), (changes) => 
+    changes
+      .bufferWithTime(delay)
+      .map((values) => values[values.length - 1]));
+};
+
+Observable.prototype.throttleImmediate = function (delay) {
+  return this.delayChanges(new Desc(this, "throttleImmediate", [delay]), (changes) =>
+      changes.flatMapFirst(first => {
+        let holding;
+        const flushHolding = () => {
+          const flushing = holding ? [holding] : [];
+          holding = null;
+          return Bacon.fromArray(flushing);
+        };
+        const silence = Bacon.once()
+          .concat(this.doAction(value => holding = value))
+          .flatMapLatest(() => Bacon.later(delay, true))
+          .take(1);
+        const sampler = Bacon.interval(delay).takeUntil(silence);
+        return Bacon.once(first).concat(sampler.flatMap(flushHolding));
+      })
+  );
 };