Repository: mostjs/core
Pull request #148 (2017-10-08 22:14:18)

feat(snapshot): Rename sample -> snapshot, add new sample
Alternative to #142 

Uses another set of names we've discussed: snapshot, sample, zipItems, withItems.

### Todo

- [x] Update API docs

Close #142


Patch:
diff --git a/docs/api.rst b/docs/api.rst
index d4aad7c..cc6cae4 100644
--- a/docs/api.rst
+++ b/docs/api.rst
@@ -36,11 +36,11 @@ Stream
     run :: Sink a -> Scheduler -> Disposable
   }
 
-A ``Stream`` represents a view of events over time. Its ``run`` method arranges events to be propagated to the provided :ref:`Sink` in the future. Each ``Stream`` has a local clock, defined by the provided :ref:`Scheduler`, which has methods for knowing the current time and scheduling future :ref:`Task`s.
+A ``Stream`` represents a view of events over time. Its ``run`` method arranges events to be propagated to the provided :ref:`Sink` in the future. Each ``Stream`` has a local clock, defined by the provided :ref:`Scheduler`, which has methods for knowing the current time and scheduling future :ref:`Tasks <Task>`.
 
-Some ``Stream``s, like :ref:`now`, are simple, while others, like :ref:`combine`, do sophisticated things such as combining multiple ``Stream``s or dealing with higher-order ``Stream``s.
+A ``Stream`` may be simple, like :ref:`now`, or may do sophisticated things such as :ref:`combining <combine>` multiple ``Stream`` s or deal with higher-order ``Stream`` s.
 
-Some ``Stream``s act as event producers, such as from DOM events. A producer ``Stream`` must never produce an event in the same call stack as its ``run`` method is called. It must begin producing items asynchronously. In some cases, this comes for free, such as DOM events. In other cases, it must be done explicitly using the provided ref:`Scheduler` to schedule asynchronous ref:`Task`s.
+A ``Stream`` may act as an event producer, such as a ``Stream`` that produces DOM events. A producer ``Stream`` must never produce an event in the same call stack as its ``run`` method is called. It must begin producing items asynchronously. In some cases, this comes for free, such as DOM events. In other cases, it must be done explicitly using the provided ref:`Scheduler` to schedule asynchronous :ref:`Tasks <Task>`.
 
 .. _Sink:
 
@@ -87,9 +87,9 @@ Scheduler
     cancelAll :: (ScheduledTask -> boolean) -> void
   }
 
-A ``Scheduler`` provides the central notion of time for the :ref:`Stream`s in an application.
+A ``Scheduler`` provides the central notion of time for the :ref:`Streams <Stream>` in an application.
 
-An application will typically create a single "root" ``Scheduler`` so that all :ref:`Stream`s share the same underlying time.
+An application will typically create a single "root" ``Scheduler`` so that all :ref:`Streams <Stream>` share the same underlying time.
 
 .. _Clock:
 
@@ -139,7 +139,7 @@ Timeline
     runTasks :: Time -> TaskRunner -> void
   }
 
-A ``Timeline`` represents a set of :ref:`ScheduledTask`s to be executed at particular times.
+A ``Timeline`` represents a set of :ref:`ScheduledTasks <ScheduledTask>` to be executed at particular times.
 
 .. _Task:
 
@@ -155,6 +155,8 @@ Task
 
 A ``Task`` is any unit of work that can be scheduled for execution with a :ref:`Scheduler`.
 
+.. _ScheduledTask:
+
 ScheduledTask
 ^^^^^^^^^^^^^
 
@@ -189,7 +191,7 @@ runEffects
 
 Activate an event :ref:`Stream` and consume all its events.
 
-.. _runStream:
+.. _run:
 
 run
 ```
@@ -255,6 +257,21 @@ Create a :ref:`Stream` containing a single event at a specific time. ::
 
   at(3, x): --x|
 
+.. _periodic:
+
+periodic
+````````
+
+.. code-block:: haskell
+
+  periodic :: Period -> Stream void
+
+Create an infinite :ref:`Stream` containing events that occur at a specified period. ::
+
+  periodic(3): x--x--x--x-->
+
+The first event occurs at time 0, and the event values are ``undefined``.
+
 .. _throwError:
 
 throwError
@@ -264,9 +281,7 @@ throwError
 
   throwError :: Error -> Stream void
 
-Create a :ref:`Stream` that fails at time 0 with the provided ``Error``.
-
-This can be useful for functions that need to return a :ref:`Stream` and also need to propagate an error. ::
+Create a :ref:`Stream` that fails with the provided ``Error`` at time 0.  This can be useful for functions that need to return a :ref:`Stream` and also need to propagate an error. ::
 
   throwError(X): X
 
@@ -430,37 +445,37 @@ It allows you to maintain and update a "state" (a.k.a. feedback, a.k.a. seed for
   	return { seed: values, value: avg }
   }, [], stream)
 
-.. _zipArrayValues:
+.. _zipItems:
 
-zipArrayValues
-``````````````
+zipItems
+````````
 
 .. code-block:: haskell
 
-  zipArrayValues :: ((a, b) -> c) -> [a] -> Stream b -> Stream c
+  zipItems :: ((a, b) -> c) -> [a] -> Stream b -> Stream c
 
 Apply a function to the latest event and the array value at the respective index. ::
 
-  stream:                             --10---10---10---10---10--->
   array:                              [ 1, 2, 3 ]
-  zipArrayValues(add, array, stream): --11---12---13|
+  stream:                             --10---10---10---10---10--->
+  zipItems(add, array, stream): --11---12---13|
 
 The resulting :ref:`Stream` will contain the same number of events as the input :ref:`Stream`, or ``array.length`` events, whichever is less.
 
-.. _withArrayValues:
+.. _withItems:
 
-withArrayValues
-```````````````
+withItems
+`````````
 
 .. code-block:: haskell
 
-  withArrayValues :: [a] -> Stream b -> Stream a
+  withItems :: [a] -> Stream b -> Stream a
 
-Replace each event value with the array value at the respective index. ::
+Replace each event value with the array item at the respective index. ::
 
-  array:                          [ 1, 2, 3 ]
-  stream:                         --x--x--x--x--x-->
-  withArrayValues(array, stream): --1--2--3|
+  array:                    [ 1, 2, 3 ]
+  stream:                   --x--x--x--x--x-->
+  withItems(array, stream): --1--2--3|
 
 The resulting :ref:`Stream` will contain the same number of events as the input :ref:`Stream`, or ``array.length`` events, whichever is less.
 
@@ -492,7 +507,7 @@ join
 
   join :: Stream (Stream a) -> Stream a
 
-Given a higher-order :ref:`Stream`, return a new :ref:`Stream` that merges all the inner :ref:`Stream`s as they arrive. ::
+Given a higher-order :ref:`Stream`, return a new :ref:`Stream` that merges all the inner :ref:`Streams <Stream>` as they arrive. ::
 
   s:             ---a---b---c---d-->
   t:             -1--2--3--4--5--6->
@@ -547,7 +562,7 @@ mergeConcurrently
 
   mergeConcurrently :: int -> Stream (Stream a) -> Stream a
 
-Given a higher-order :ref:`Stream`, return a new :ref:`Stream` that merges inner :ref:`Stream`s as they arrive up to the specified concurrency. Once concurrency number of :ref:`Stream`s are being merged, newly arriving :ref:`Stream`s will be merged after an existing one ends. ::
+Given a higher-order :ref:`Stream`, return a new :ref:`Stream` that merges inner :ref:`Streams <Stream>` as they arrive up to the specified concurrency. Once concurrency number of :ref:`Streams <Stream>` are being merged, newly arriving :ref:`Streams <Stream>` will be merged after an existing one ends. ::
 
   s:                            --a--b--c--d--e-->
   t:                            --x------y|
@@ -559,7 +574,7 @@ Note that ``u`` is only merged after ``t`` ends because of the concurrency level
 
 Note also that ``mergeConcurrently(Infinity, stream)`` is equivalent to ``join(stream)``.
 
-To control concurrency, ``mergeConcurrently`` must maintain an internal queue of newly arrived :ref:`Stream`s. If new :ref:`Stream`s arrive faster than the concurrency level allows them to be merged, the internal queue will grow infinitely.
+To control concurrency, ``mergeConcurrently`` must maintain an internal queue of newly arrived :ref:`Streams <Stream>`. If new :ref:`Streams <Stream>` arrive faster than the concurrency level allows them to be merged, the internal queue will grow infinitely.
 
 .. _mergeMapConcurrently:
 
@@ -570,7 +585,7 @@ mergeMapConcurrently
 
   mergeMapConcurrently :: (a -> Stream b) -> int -> Stream a -> Stream b
 
-Lazily apply a function ``f`` to each event in a :ref:`Stream`, merging them into the resulting :ref:`Stream` at the specified concurrency. Once concurrency number of :ref:`Stream`s are being merged, newly arriving :ref:`Stream`s will be merged after an existing one ends. ::
+Lazily apply a function ``f`` to each event in a :ref:`Stream`, merging them into the resulting :ref:`Stream` at the specified concurrency. Once concurrency number of :ref:`Streams <Stream>` are being merged, newly arriving :ref:`Streams <Stream>` will be merged after an existing one ends. ::
 
   stream:                             --ab--c----d----->
   f(a):                               -1-2-3|
@@ -583,7 +598,7 @@ Note that ``f(c)`` is only merged after ``f(a)`` ends.
 
 Also note that ``f`` will not get called with ``d`` until either ``f(b)`` or ``f(c)`` ends.
 
-To control concurrency, ``mergeMapConcurrently`` must maintain an internal queue of newly arrived :ref:`Stream`s. If new :ref:`Stream`s arrive faster than the concurrency level allows them to be merged, the internal queue will grow infinitely.
+To control concurrency, ``mergeMapConcurrently`` must maintain an internal queue of newly arrived :ref:`Streams <Stream>`. If new :ref:`Streams <Stream>` arrive faster than the concurrency level allows them to be merged, the internal queue will grow infinitely.
 
 Merging
 ^^^^^^^
@@ -597,13 +612,13 @@ merge
 
   merge :: Stream a -> Stream a -> Stream a
 
-Create a new :ref:`Stream` containing events from two :ref:`Stream`s. ::
+Create a new :ref:`Stream` containing events from two :ref:`Streams <Stream>`. ::
 
   s1:            -a--b----c--->
   s2:            --w---x-y--z->
   merge(s1, s2): -aw-b-x-yc-z->
 
-Merging creates a new :ref:`Stream` containing all events from the two original :ref:`Stream`s without affecting the time of the events. You can think of the events from the input :ref:`Stream`s simply being interleaved into the new, merged :ref:`Stream`. A merged :ref:`Stream` ends when all of its input :ref:`Stream`s have ended.
+Merging creates a new :ref:`Stream` containing all events from the two original :ref:`Streams <Stream>` without affecting the time of the events. You can think of the events from the input :ref:`Streams <Stream>` simply being interleaved into the new, merged :ref:`Stream`. A merged :ref:`Stream` ends when all of its input :ref:`Streams <Stream>` have ended.
 
 .. _mergeArray:
 
@@ -614,7 +629,7 @@ mergeArray
 
   mergeArray :: [ (Stream a) ] -> Stream a
 
-Array form of :ref:`merge`. Create a new :ref:`Stream` containing all events from all :ref:`Stream`s in the array. ::
+Array form of :ref:`merge`. Create a new :ref:`Stream` containing all events from all :ref:`Streams <Stream>` in the array. ::
 
   s1:                       -a--b----c---->
   s2:                       --w---x-y--z-->
@@ -636,7 +651,7 @@ Apply a function to the most recent event from each :ref:`Stream` when a new eve
   s2:                   --3---4-5--6->
   combine(add, s1, s2): --3-4-5-67-8->
 
-Note that ``combine`` waits for at least one event to arrive on all input :ref:`Stream`s before it produces any events.
+Note that ``combine`` waits for at least one event to arrive on all input :ref:`Streams <Stream>` before it produces any events.
 
 .. _combineArray:
 
@@ -647,7 +662,7 @@ combineArray
 
   combineArray :: ((a, b, ...) -> z) -> [ Stream a, Stream b, ... ] -> Stream z
 
-Array form of :ref:`combine`. Apply a function to the most recent event from all :ref:`Stream`s when a new event arrives on any :ref:`Stream`. ::
+Array form of :ref:`combine`. Apply a function to the most recent event from all :ref:`Streams <Stream>` when a new event arrives on any :ref:`Stream`. ::
 
   s1:                               -0--1----2->
   s2:                               --3---4-5-->
@@ -663,7 +678,7 @@ zip
 
   zip :: (a -> b -> c) -> Stream a -> Stream b -> Stream c
 
-Apply a function to corresponding pairs of events from the inputs :ref:`Stream`s. ::
+Apply a function to corresponding pairs of events from the inputs :ref:`Streams <Stream>`. ::
 
   s1:               -1--2--3--4->
   s2:               -1---2---3---4->
@@ -671,7 +686,7 @@ Apply a function to corresponding pairs of events from the inputs :ref:`Stream`s
 
 Zipping correlates by *index*-corresponding events from two input streams. Note that zipping a "fast" :ref:`Stream` and a "slow" :ref:`Stream` will cause buffering. Events from the fast :ref:`Stream` must be buffered in memory until an event at the corresponding index arrives on the slow :ref:`Stream`.
 
-A zipped :ref:`Stream` ends when any one of its input :ref:`Stream`s ends.
+A zipped :ref:`Stream` ends when any one of its input :ref:`Streams <Stream>` ends.
 
 .. _zipArray:
 
@@ -682,31 +697,50 @@ zipArray
 
   zipArray :: ((a, b, ...) -> z) -> [ Stream a, Stream b, ... ] -> Stream z
 
-Array form of :ref:`zip`. Apply a function to corresponding events from all the inputs :ref:`Stream`s. ::
+Array form of :ref:`zip`. Apply a function to corresponding events from all the inputs :ref:`Streams <Stream>`. ::
 
   s1:                           -1-2-3---->
   s2:                           -1--2--3-->
   s3:                           --1--2--3->
   zipArray(add3, [s1, s2, s3]): --3--6--9->
 
+_sample
+
 sample
 ``````
 
 .. code-block:: haskell
 
-  sample :: ((a, b) -> c) -> Stream a -> Stream b -> Stream c
+  sample :: Stream a -> Stream b -> Stream a
+
+For each event in a sampler :ref:`Stream`, replace the event value with the latest value in another :ref:`Stream`.  The resulting :ref:`Stream` will contain the same number of events as the sampler :ref:`Stream`. ::
+
+  values:                  -1--2--3--4--5->
+  sampler:                 -1-----2-----3->
+  sample(values, sampler): -1-----3-----5->
+
+  values:                  -1-----2-----3->
+  sampler:                 -1--2--3--4--5->
+  sample(values, sampler): -1--1--2--2--3->
+
+snapshot
+````````
+
+.. code-block:: haskell
+
+  snapshot :: ((a, b) -> c) -> Stream a -> Stream b -> Stream c
 
-For each event in a sampler :ref:`Stream`, apply a function to combine it with the most recent event in another :ref:`Stream`. The resulting :ref:`Stream` will contain the same number of events as the sampler :ref:`Stream`. ::
+For each event in a sampler :ref:`Stream`, apply a function to combine its value with the most recent event value in another :ref:`Stream`. The resulting :ref:`Stream` will contain the same number of events as the sampler :ref:`Stream`. ::
 
-  s1:                       -1--2--3--4--5->
-  sampler:                  -1-----2-----3->
-  sample(sum, sampler, s1): -2-----5-----8->
+  values:                         -1--2--3--4--5->
+  sampler:                        -1-----2-----3->
+  snapshot(sum, values, sampler): -2-----5-----8->
 
-  s1:                       -1-----2-----3->
-  sampler:                  -1--2--3--4--5->
-  sample(sum, sampler, s1): -2--3--5--6--8->
+  values:                         -1-----2-----3->
+  sampler:                        -1--2--3--4--5->
+  snapshot(sum, values, sampler): -2--3--5--6--8->
 
-Note that ``sample`` produces a value only when an event arrives on the sampler.
+In contrast to :ref:`combine`, ``snapshot`` produces a value only when an event arrives on the sampler.
 
 Filtering
 ^^^^^^^^^
@@ -946,7 +980,7 @@ delay
 
 .. code-block:: haskell
 
-  delay :: int -> Stream a -> Stream a
+  delay :: Delay -> Stream a -> Stream a
 
 Timeshift a :ref:`Stream` by *n* milliseconds. ::
 
@@ -1116,7 +1150,7 @@ Multicast allows you to build up a stream of maps, filters, and other transforma
 Tasks
 ^^^^^
 
-Helper functions for creating :ref:`Task`s to propagate events.
+Helper functions for creating :ref:`Tasks <Task>` to propagate events.
 
 .. _propagateTask:
 
@@ -1246,7 +1280,7 @@ cancelAllTasks
 
   cancelAllTasks :: (ScheduledTask -> boolean) -> Scheduler -> void
 
-Cancel all future scheduled executions of all :ref:`ScheduledTask`s for which the provided predicate is ``true``.
+Cancel all future scheduled executions of all :ref:`ScheduledTasks <ScheduledTask>` for which the provided predicate is ``true``.
 
 Creating a Scheduler
 ^^^^^^^^^^^^^^^^^^^^
@@ -1260,7 +1294,7 @@ newScheduler
 
   newScheduler :: Timer -> Timeline -> Scheduler
 
-Create a new :ref:`Scheduler` that uses the provided :ref:`Timer` and :ref:`Timeline` for scheduling :ref:`Task`s.
+Create a new :ref:`Scheduler` that uses the provided :ref:`Timer` and :ref:`Timeline` for scheduling :ref:`Tasks <Task>`.
 
 .. _newDefaultScheduler:
 
@@ -1429,7 +1463,7 @@ disposeBoth
 
   disposeBoth :: Disposable -> Disposable -> Disposable
 
-Combine two :ref:`Disposable`s into a single :ref:`Disposable` which will dispose of both.
+Combine two :ref:`Disposables <Disposable>` into a single :ref:`Disposable` which will dispose of both.
 
 .. _disposeAll:
 
@@ -1440,7 +1474,7 @@ disposeAll
 
   disposeAll :: [Disposable] -> Disposable
 
-Combine an array of :ref:`Disposable`s into a single :ref:`Disposable` which will dispose of all the :ref:`Disposable`s in the array.
+Combine an array of :ref:`Disposables <Disposable>` into a single :ref:`Disposable` which will dispose of all the :ref:`Disposables <Disposable>` in the array.
 
 .. _Disposing Disposables:
 
diff --git a/docs/concepts.rst b/docs/concepts.rst
index 4006ecb..bbe66b7 100644
--- a/docs/concepts.rst
+++ b/docs/concepts.rst
@@ -15,12 +15,52 @@ An *event stream* is a time-ordered series of events.  For example, all the mous
 
 Conceptually, you can think of an event stream as an array of (time, value) pairs.  However, whereas an array is ordered by index, an event stream is ordered by its events' occurrence times.  You can't observe the second mouse click until after you've observed the first one.
 
+Event streams may be *infinite*, *finite*, or may *fail*.
+
+Infinite Event Streams
+^^^^^^^^^^^^^^^^^^^^^^
+
+An event stream may be infinite.  For example, :ref:`periodic` creates an infinite stream of events that occur at a regular period.  Limiting operations, such as :ref:`take` or :ref:`until`, are helpful in turning an infinite event stream into a finite one.
+
+Finite Event Streams
+^^^^^^^^^^^^^^^^^^^^
+
+An event stream may be finite.  A finite event stream will produce an *end signal* to indicate that it will never produce another event.  When an event stream ends, it will free underlying resources.
+
+.. _Failed Event Streams:
+
+Failed Event Streams
+^^^^^^^^^^^^^^^^^^^^
+
+An event stream may fail.  For example, when an event stream represents a resource, such as a WebSocket, and the resource fails or closes unexpectedly, the event stream *cannot* produce more events.  When an event stream fails, it will produce an *error signal* to indicate that it cannot produce more events.  The error signal includes a *reason* (an ``Error`` object) describing the failure.
+
+A failed event stream will attempt to free any underlying resources.
+
+The :ref:`recoverWith` operation allows you to handle an event stream failure.
+
+.. _Application Errors:
+
+Stream Failure vs. Application Errors
+`````````````````````````````````````
+
+Stream failures are different from *application errors*.  A stream failure indicates that an event stream *cannot* produce more events.  Application errors may or may not indicate an event stream failure.  For example, an event stream representing messages arriving on a WebSocket *fails* if the WebSocket is disconnected unexpectedly due to a wifi drop.  In contrast, an application error may occur in application logic that receives a WebSocket message and can't process because the message application state has changed.
+
+Application error handling is outside the scope of these docs, as it is application-specific.  However, there are some general strategies for dealing with application errors with event streams:
+
+* Use try/catch, `Promise catch() <https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Promise/catch>`_ to handle the application error and transform it into:
+
+  * a useful event value
+  * a sentinel event value that can be :ref:`filtered <filter>`
+  * an `Either <https://github.com/sanctuary-js/sanctuary#either-type>`_ value, or other equivalent structure for representing a value or an error as an event value
+
+* Use :ref:`throwError` to transform the application error into a stream failure
+
 Source and sink chains
 ----------------------
 
 Applying combinators to a stream composes a *source chain* that defines the behavior of the stream.  When an observer begins observing a stream, a :ref:`run <Stream>` message is sent "backwards" through the chain, to the ultimate producer--the one that will produce events in the first place.
 
-As it travels, that message composes a *:ref:`Sink` chain* analogous to the source chain.  When the messages reaches the producer, it begins producing events.  With the exception of a few combinators (such as :ref:`delay`), events propagate *synchronously* "forward" through the sink chain.
+As it travels, that message composes a :ref:`Sink` chain analogous to the source chain.  When the messages reaches the producer, it begins producing events.  With the exception of a few combinators (such as :ref:`delay`), events propagate *synchronously* "forward" through the sink chain.
 
 **Note**: a producer must not *begin* producing events synchronously.  It must schedule the *start* of its production, using the :ref:`Scheduler` passed to its :ref:`run <Stream>` method.  However, once it does begin, it may then produce events synchronously.
 
@@ -34,4 +74,6 @@ Some combinators, like :ref:`delay`, introduce asynchrony into the sink chain.
 Error propagation
 -----------------
 
+.. attention:: Uncaught exceptions in a sink chain are considered to be :ref:`failures <Failed Event Streams>`, and not *application errors*.  See :ref:`Stream Failure vs. Application Errors <Application Errors>` for more information.
+
 If an exception is thrown during event propagation, it will stop the propagation and travel "backwards" through the sink chain, by unwinding the call stack.  If that exception is not caught, it will reach the producer, and finally, the scheduler.  The scheduler will catch it and send the error "forward" again synchronously, using the `error` channel of the sink chain.
diff --git a/packages/core/src/combinator/sample.js b/packages/core/src/combinator/sample.js
deleted file mode 100644
index 2746aa5..0000000
--- a/packages/core/src/combinator/sample.js
+++ /dev/null
@@ -1,53 +0,0 @@
-/** @license MIT License (c) copyright 2010-2016 original author or authors */
-
-import Pipe from '../sink/Pipe'
-import { disposeBoth } from '@most/disposable'
-
-export const sample = (f, sampler, stream) =>
-  new Sample(f, sampler, stream)
-
-export class Sample {
-  constructor (f, sampler, stream) {
-    this.source = stream
-    this.sampler = sampler
-    this.f = f
-  }
-
-  run (sink, scheduler) {
-    const sampleSink = new SampleSink(this.f, this.source, sink)
-    const sourceDisposable = this.source.run(sampleSink.hold, scheduler)
-    const samplerDisposable = this.sampler.run(sampleSink, scheduler)
-
-    return disposeBoth(samplerDisposable, sourceDisposable)
-  }
-}
-
-export class SampleSink extends Pipe {
-  constructor (f, source, sink) {
-    super(sink)
-    this.source = source
-    this.f = f
-    this.hold = new SampleHold(this)
-  }
-
-  event (t, x) {
-    if (this.hold.hasValue) {
-      const f = this.f
-      this.sink.event(t, f(x, this.hold.value))
-    }
-  }
-}
-
-export class SampleHold extends Pipe {
-  constructor (sink) {
-    super(sink)
-    this.hasValue = false
-  }
-
-  event (t, x) {
-    this.value = x
-    this.hasValue = true
-  }
-
-  end () {}
-}
diff --git a/packages/core/src/combinator/snapshot.js b/packages/core/src/combinator/snapshot.js
new file mode 100644
index 0000000..1b133b0
--- /dev/null
+++ b/packages/core/src/combinator/snapshot.js
@@ -0,0 +1,55 @@
+/** @license MIT License (c) copyright 2010 original author or authors */
+
+import Pipe from '../sink/Pipe'
+import { disposeBoth } from '@most/disposable'
+
+export const sample = (values, sampler) =>
+  snapshot((_, x) => x, values, sampler)
+
+export const snapshot = (f, values, sampler) =>
+  new Snapshot(f, values, sampler)
+
+export class Snapshot {
+  constructor (f, values, sampler) {
+    this.f = f
+    this.values = values
+    this.sampler = sampler
+  }
+
+  run (sink, scheduler) {
+    const sampleSink = new SnapshotSink(this.f, sink)
+    const valuesDisposable = this.values.run(sampleSink.latest, scheduler)
+    const samplerDisposable = this.sampler.run(sampleSink, scheduler)
+
+    return disposeBoth(samplerDisposable, valuesDisposable)
+  }
+}
+
+export class SnapshotSink extends Pipe {
+  constructor (f, sink) {
+    super(sink)
+    this.f = f
+    this.latest = new LatestValueSink(this)
+  }
+
+  event (t, x) {
+    if (this.latest.hasValue) {
+      const f = this.f
+      this.sink.event(t, f(this.latest.value, x))
+    }
+  }
+}
+
+export class LatestValueSink extends Pipe {
+  constructor (sink) {
+    super(sink)
+    this.hasValue = false
+  }
+
+  event (t, x) {
+    this.value = x
+    this.hasValue = true
+  }
+
+  end () {}
+}
diff --git a/packages/core/src/combinator/withArrayValues.js b/packages/core/src/combinator/withArrayValues.js
deleted file mode 100644
index 2f3c5ae..0000000
--- a/packages/core/src/combinator/withArrayValues.js
+++ /dev/null
@@ -1,45 +0,0 @@
-/** @license MIT License (c) copyright 2010-2017 original author or authors */
-
-import { empty } from '../source/empty'
-import Pipe from '../sink/Pipe'
-
-export const withArrayValues = (array, stream) =>
-  zipArrayValues(keepLeft, array, stream)
-
-export const zipArrayValues = (f, array, stream) =>
-  array.length === 0 || stream === empty()
-    ? empty()
-    : new ZipArrayValues(f, array, stream)
-
-const keepLeft = (a, _) => a
-
-class ZipArrayValues {
-  constructor (f, values, source) {
-    this.f = f
-    this.values = values
-    this.source = source
-  }
-
-  run (sink, scheduler) {
-    return this.source.run(new ZipArrayValuesSink(this.f, this.values, sink), scheduler)
-  }
-}
-
-class ZipArrayValuesSink extends Pipe {
-  constructor (f, values, sink) {
-    super(sink)
-    this.f = f
-    this.values = values
-    this.index = 0
-  }
-
-  event (t, b) {
-    const f = this.f
-    this.sink.event(t, f(this.values[this.index], b))
-
-    this.index += 1
-    if (this.index >= this.values.length) {
-      this.sink.end(t)
-    }
-  }
-}
diff --git a/packages/core/src/combinator/zipItems.js b/packages/core/src/combinator/zipItems.js
new file mode 100644
index 0000000..7241765
--- /dev/null
+++ b/packages/core/src/combinator/zipItems.js
@@ -0,0 +1,45 @@
+/** @license MIT License (c) copyright 2017 original author or authors */
+
+import { empty } from '../source/empty'
+import Pipe from '../sink/Pipe'
+
+export const withItems = (items, stream) =>
+  zipItems(keepLeft, items, stream)
+
+export const zipItems = (f, items, stream) =>
+  items.length === 0 || stream === empty()
+    ? empty()
+    : new ZipItems(f, items, stream)
+
+const keepLeft = (a, _) => a
+
+class ZipItems {
+  constructor (f, items, source) {
+    this.f = f
+    this.items = items
+    this.source = source
+  }
+
+  run (sink, scheduler) {
+    return this.source.run(new ZipItemsSink(this.f, this.items, sink), scheduler)
+  }
+}
+
+class ZipItemsSink extends Pipe {
+  constructor (f, items, sink) {
+    super(sink)
+    this.f = f
+    this.items = items
+    this.index = 0
+  }
+
+  event (t, b) {
+    const f = this.f
+    this.sink.event(t, f(this.items[this.index], b))
+
+    this.index += 1
+    if (this.index >= this.items.length) {
+      this.sink.end(t)
+    }
+  }
+}
diff --git a/packages/core/src/index.js b/packages/core/src/index.js
index cc762c8..896e872 100644
--- a/packages/core/src/index.js
+++ b/packages/core/src/index.js
@@ -11,9 +11,9 @@ export { periodic } from './source/periodic'
 
 export { newStream } from './source/newStream'
 
-import { zipArrayValues as _zipArrayValues, withArrayValues as _withArrayValues } from './combinator/withArrayValues'
-export const zipArrayValues = curry3(_zipArrayValues)
-export const withArrayValues = curry2(_withArrayValues)
+import { zipItems as _zipItems, withItems as _withItems } from './combinator/zipItems'
+export const zipItems = curry3(_zipItems)
+export const withItems = curry2(_withItems)
 
 // -----------------------------------------------------------------------
 // Observing
@@ -100,9 +100,10 @@ export const combineArray = curry2(_combineArray)
 // -----------------------------------------------------------------------
 // Sampling
 
-import { sample as _sample } from './combinator/sample'
+import { sample as _sample, snapshot as _snapshot } from './combinator/snapshot'
 
-export const sample = curry3(_sample)
+export const sample = curry2(_sample)
+export const snapshot = curry3(_snapshot)
 
 // -----------------------------------------------------------------------
 // Zipping
diff --git a/packages/core/src/index.js.flow b/packages/core/src/index.js.flow
index d4e23e0..363f4e3 100644
--- a/packages/core/src/index.js.flow
+++ b/packages/core/src/index.js.flow
@@ -77,10 +77,13 @@ declare export function merge <A> (s1: Stream<A>, s2: Stream<A>): Stream<A>
 declare export function merge <A> (s1: Stream<A>): (s2: Stream<A>) => Stream<A>
 declare export function mergeArray <A> (ss: Array<Stream<A>>): Stream<A>
 
-declare export function sample <A, B, C> (f: (A, B) => C, sampler: Stream<A>, sampled: Stream<B>): Stream<C>
-declare export function sample <A, B, C> (f: (A, B) => C): (sampler: Stream<A>, sampled: Stream<B>) => Stream<C>
-declare export function sample <A, B, C> (f: (A, B) => C, sampler: Stream<A>): (sampled: Stream<B>) => Stream<C>
-declare export function sample <A, B, C> (f: (A, B) => C): (sampler: Stream<A>) => (sampled: Stream<B>) => Stream<C>
+declare export function sample <A> (values: Stream<A>, sampler: Stream<any>): Stream<A>
+declare export function sample <A> (values: Stream<A>): (sampler: Stream<any>) => Stream<A>
+
+declare export function snapshot <A, B, C> (f: (A, B) => C, values: Stream<A>, sampler: Stream<B>): Stream<C>
+declare export function snapshot <A, B, C> (f: (A, B) => C): (values: Stream<A>, sampler: Stream<B>) => Stream<C>
+declare export function snapshot <A, B, C> (f: (A, B) => C, values: Stream<A>): (sampler: Stream<B>) => Stream<C>
+declare export function snapshot <A, B, C> (f: (A, B) => C): (values: Stream<A>) => (sampler: Stream<B>) => Stream<C>
 
 declare export function combine <A, B, Z> (f: (A, B) => Z, sa: Stream<A>, sb: Stream<B>): Stream<Z>
 declare export function combine <A, B, Z> (f: (A, B) => Z): (sa: Stream<A>, sb: Stream<B>) => Stream<Z>
diff --git a/packages/core/type-definitions/combinator/sample.d.ts b/packages/core/type-definitions/combinator/sample.d.ts
deleted file mode 100644
index 1ddae25..0000000
--- a/packages/core/type-definitions/combinator/sample.d.ts
+++ /dev/null
@@ -1,10 +0,0 @@
-import { Stream } from '@most/types';
-
-// all 3 args
-export function sample<A, B, C>(f: (a: A, b: B) => C, sampler: Stream<A>, s: Stream<B>): Stream<C>;
-// first arg then 2
-export function sample<A, B, C>(f: (a: A, b: B) => C): (sampler: Stream<A>, s: Stream<B>) => Stream<C>;
-// first 2 then 1
-export function sample<A, B, C>(f: (a: A, b: B) => C, sampler: Stream<A>): (s: Stream<B>) => Stream<C>;
-// 1 arg at a time
-export function sample<A, B, C>(f: (a: A, b: B) => C): (sampler: Stream<A>) => (s: Stream<B>) => Stream<C>;
diff --git a/packages/core/type-definitions/combinator/snapshot.d.ts b/packages/core/type-definitions/combinator/snapshot.d.ts
new file mode 100644
index 0000000..a939607
--- /dev/null
+++ b/packages/core/type-definitions/combinator/snapshot.d.ts
@@ -0,0 +1,13 @@
+import { Stream } from '@most/types';
+
+export function sample <A, B> (values: Stream<A>, sampler: Stream<B>): Stream<A>;
+export function sample <A, B> (values: Stream<A>): (sampler: Stream<B>) => Stream<A>;
+
+// all 3 args
+export function snapshot <A, B, C>(f: (a: A, b: B) => C, values: Stream<A>, sampler: Stream<B>): Stream<C>;
+// first 1 then 2
+export function snapshot <A, B, C>(f: (a: A, b: B) => C): (values: Stream<A>, sampler: Stream<B>) => Stream<C>;
+// first 2 then 1
+export function snapshot <A, B, C>(f: (a: A, b: B) => C, values: Stream<A>): (sampler: Stream<B>) => Stream<C>;
+// 1 arg at a time
+export function snapshot <A, B, C>(f: (a: A, b: B) => C): (values: Stream<A>) => (sampler: Stream<B>) => Stream<C>;
diff --git a/packages/core/type-definitions/combinator/withArrayValues.d.ts b/packages/core/type-definitions/combinator/zipItems.d.ts
similarity index 100%
rename from packages/core/type-definitions/combinator/withArrayValues.d.ts
rename to packages/core/type-definitions/combinator/zipItems.d.ts
diff --git a/packages/core/type-definitions/most.d.ts b/packages/core/type-definitions/most.d.ts
index b89af77..771eead 100644
--- a/packages/core/type-definitions/most.d.ts
+++ b/packages/core/type-definitions/most.d.ts
@@ -16,16 +16,16 @@ export * from './combinator/merge';
 export * from './combinator/mergeConcurrently';
 export * from './combinator/multicast';
 export * from './combinator/promises';
-export * from './combinator/sample';
 export * from './combinator/scan';
 export * from './combinator/slice';
+export * from './combinator/snapshot';
 export * from './combinator/startWith';
 export * from './combinator/switch';
 export * from './combinator/timeslice';
 export * from './combinator/transform';
-export * from './combinator/zip';
-export * from './combinator/withArrayValues';
 export * from './combinator/withLocalTime';
+export * from './combinator/zip';
+export * from './combinator/zipItems';
 
 export * from './source/never';
 export * from './source/empty';


Test patch:
diff --git a/packages/core/test/combinator/sample-test.js b/packages/core/test/combinator/snapshot-test.js
similarity index 62%
rename from packages/core/test/combinator/sample-test.js
rename to packages/core/test/combinator/snapshot-test.js
index 2ed7c9f..254ae68 100644
--- a/packages/core/test/combinator/sample-test.js
+++ b/packages/core/test/combinator/snapshot-test.js
@@ -1,34 +1,33 @@
-/** @license MIT License (c) copyright 2016 original author or authors */
-
+// @flow
 import { describe, it } from 'mocha'
 import { is, eq } from '@briancavalier/assert'
 
 import { throwError } from '../../src/combinator/errors'
+import { at } from '../../src/source/at'
 import { now } from '../../src/source/now'
 import { never } from '../../src/source/never'
-import { sample } from '../../src/combinator/sample'
+import { snapshot } from '../../src/combinator/snapshot'
 
 import { makeEvents, ticks, collectEvents } from '../helper/testEnv'
 
 const rint = n => Math.ceil(n * Math.random())
 
-describe('sample', () => {
+describe('snapshot', () => {
   it('should pass in the sampler value', () => {
-    const samplerValue = Math.random()
-    const justValue = Math.random()
+    const streamValue = Math.random()
+    const referenceValue = Math.random()
 
-    const f = (a, b) => [a, b]
-    const s = sample(f, now(samplerValue), now(justValue))
+    const s = snapshot(Array, now(referenceValue), now(streamValue))
 
     return collectEvents(s, ticks(1)).then(events => {
       eq(1, events.length)
-      eq({ time: 0, value: [samplerValue, justValue] }, events[0])
+      eq({ time: 0, value: [referenceValue, streamValue] }, events[0])
     })
   })
 
   it('should produce no events before first stream event', () => {
     const n = rint(10)
-    const s = sample(Array, makeEvents(1, n), never())
+    const s = snapshot(Array, never(), makeEvents(1, n))
 
     return collectEvents(s, ticks(n))
       .then(events => eq(0, events.length))
@@ -36,7 +35,7 @@ describe('sample', () => {
 
   it('should end when sampler ends', () => {
     const n = rint(10)
-    const s = sample(Array, makeEvents(1, n), makeEvents(5, n * 2))
+    const s = snapshot(Array, makeEvents(5, n * 2), makeEvents(1, n))
 
     return collectEvents(s, ticks(n))
       .then(events => eq(n, events.length))
@@ -45,18 +44,18 @@ describe('sample', () => {
   it('should repeat last value after source ends', () => {
     const n = 1 + rint(10)
     const x = Math.random()
-    const s = sample(Array, makeEvents(1, n), now(x))
+    const s = snapshot(Array, now(x), makeEvents(1, n))
 
     return collectEvents(s, ticks(n)).then(events => {
       eq(n, events.length)
       events.forEach((event, i) =>
-        eq({ time: i, value: [i, x] }, event))
+        eq({ time: i, value: [x, i] }, event))
     })
   })
 
-  it('should error if stream errors', () => {
+  it('should error if values stream errors', () => {
     const error = new Error('fail')
-    const s = sample(Array, makeEvents(1, 1), throwError(error))
+    const s = snapshot(Array, throwError(error), at(1, 1))
 
     return collectEvents(s, ticks(1))
       .catch(is(error))
diff --git a/packages/core/test/combinator/withArrayValues-test.js b/packages/core/test/combinator/zipItems-test.js
similarity index 54%
rename from packages/core/test/combinator/withArrayValues-test.js
rename to packages/core/test/combinator/zipItems-test.js
index e5a0703..8ead31f 100644
--- a/packages/core/test/combinator/withArrayValues-test.js
+++ b/packages/core/test/combinator/zipItems-test.js
@@ -2,27 +2,27 @@
 import { describe, it } from 'mocha'
 import { eq, fail } from '@briancavalier/assert'
 
-import { zipArrayValues, withArrayValues } from '../../src/combinator/withArrayValues'
+import { zipItems, withItems } from '../../src/combinator/zipItems'
 import { empty } from '../../src/source/empty'
 
 import { collectEventsFor, makeEvents } from '../helper/testEnv'
 
-describe('withArrayValues', () => {
-  describe('zipArrayValues', () => {
-    it('should be empty for empty array', () => {
-      const s = zipArrayValues(fail, [], makeEvents(1, 1))
+describe('zipItems', () => {
+  describe('zipItems', () => {
+    it('should be empty given no items', () => {
+      const s = zipItems(fail, [], makeEvents(1, 1))
       return collectEventsFor(1, s).then(eq([]))
     })
 
-    it('should be empty for empty stream', () => {
-      const s = zipArrayValues(fail, [1, 2, 3], empty())
+    it('should be empty given empty stream', () => {
+      const s = zipItems(fail, [1, 2, 3], empty())
       return collectEventsFor(1, s).then(eq([]))
     })
 
-    it('should contain zipped values when more events than values', () => {
+    it('should contain zipped items given more events than items', () => {
       const a = ['a', 'b', 'c']
       const n = a.length + 1
-      const s = zipArrayValues((a, b) => a + String(b), a, makeEvents(1, n))
+      const s = zipItems((a, b) => a + String(b), a, makeEvents(1, n))
 
       return collectEventsFor(n, s)
         .then(eq([
@@ -32,10 +32,10 @@ describe('withArrayValues', () => {
         ]))
     })
 
-    it('should contain zipped values when more values than events', () => {
+    it('should contain zipped items given more values than items', () => {
       const a = ['a', 'b', 'c']
       const n = a.length - 1
-      const s = zipArrayValues((a, b) => a + String(b), a, makeEvents(1, n))
+      const s = zipItems((a, b) => a + String(b), a, makeEvents(1, n))
 
       return collectEventsFor(a.length, s)
         .then(eq([
@@ -45,21 +45,21 @@ describe('withArrayValues', () => {
     })
   })
 
-  describe('withArrayValues', () => {
-    it('should be empty for empty array', () => {
-      const s = withArrayValues([], makeEvents(1, 1))
+  describe('withItems', () => {
+    it('should be empty given no items', () => {
+      const s = withItems([], makeEvents(1, 1))
       return collectEventsFor(1, s).then(eq([]))
     })
 
-    it('should be empty for empty stream', () => {
-      const s = withArrayValues([1, 2, 3], empty())
+    it('should be empty given empty stream', () => {
+      const s = withItems([1, 2, 3], empty())
       return collectEventsFor(1, s).then(eq([]))
     })
 
-    it('should contain array values when more events than values', () => {
+    it('should contain items given more events than items', () => {
       const a = ['a', 'b', 'c']
       const n = a.length + 1
-      const s = withArrayValues(a, makeEvents(1, n))
+      const s = withItems(a, makeEvents(1, n))
 
       return collectEventsFor(n, s)
         .then(eq([
@@ -69,10 +69,10 @@ describe('withArrayValues', () => {
         ]))
     })
 
-    it('should contain array values when more values than events', () => {
+    it('should contain items given more items than events', () => {
       const a = ['a', 'b', 'c']
       const n = a.length - 1
-      const s = withArrayValues(a, makeEvents(1, n))
+      const s = withItems(a, makeEvents(1, n))
 
       return collectEventsFor(a.length, s)
         .then(eq([