Repository: mostjs/core
Pull request #79 (2017-06-19 11:35:59)

feat(at): Add at and now. Move empty, never to own modules
Add `at(time, value)` and `now(value)`, per recent discussion around ensuring combinators always have meaningful time semantics and names.  With the recent move to event stream-local clocks, `at(t, x)` has semantics of "x occurs at time t on the newly created event stream's local clock", and `now(x)` has the same semantics, with t fixed to zero.

I didn't remove `just` (or the `src/source/core.js` module) yet, since it's used *a lot* in unit tests, and I didn't want to deal with the collateral damage until we know for sure this is the way we want to go.

Definitely looking for feedback and discussion!

### Other changes

Other changes we may want to keep regardless of now/at.

1. Moved `empty` and `never` to their own modules.
2. Updated some deps

### Todo

- [x] Remove `just`, and `src/source/core.js`
- [x] Add type defs for `at` and `now`

briancavalier: 🔪 `just` 🔪 

Removing it made the PR very noisy, but such is the price of progress 😃 
briancavalier: I think this is ready for another review.  Sorry for the noise around removing `just`.

Patch:
diff --git a/packages/core/package.json b/packages/core/package.json
index 9838e97..119c4c0 100644
--- a/packages/core/package.json
+++ b/packages/core/package.json
@@ -46,9 +46,9 @@
     "buba": "^4.0.2",
     "cpy-cli": "^1.0.1",
     "eslint": "^3.15.0",
-    "flow-bin": "^0.43.1",
-    "mocha": "^3.2.0",
-    "rollup": "^0.41.4",
+    "flow-bin": "^0.47.0",
+    "mocha": "^3.4.2",
+    "rollup": "^0.43.0",
     "rollup-plugin-buble": "^0.15.0",
     "rollup-plugin-node-resolve": "^2.0.0",
     "sinon": "^1.17.7",
diff --git a/packages/core/src/combinator/build.js b/packages/core/src/combinator/build.js
index f980673..d42699a 100644
--- a/packages/core/src/combinator/build.js
+++ b/packages/core/src/combinator/build.js
@@ -2,7 +2,7 @@
 /** @author Brian Cavalier */
 /** @author John Hann */
 
-import { just } from '../source/core'
+import { now } from '../source/now'
 import { continueWith } from './continueWith'
 
 /**
@@ -10,7 +10,7 @@ import { continueWith } from './continueWith'
  * @param {Stream} stream
  * @returns {Stream} new stream with x prepended
  */
-export const startWith = (x, stream) => concat(just(x), stream)
+export const startWith = (x, stream) => concat(now(x), stream)
 
 /**
 * @param {Stream} left
diff --git a/packages/core/src/combinator/combine.js b/packages/core/src/combinator/combine.js
index 80f2b78..aee954a 100644
--- a/packages/core/src/combinator/combine.js
+++ b/packages/core/src/combinator/combine.js
@@ -3,7 +3,7 @@
 /** @author John Hann */
 
 import { map } from './transform'
-import { empty } from '../source/core'
+import { empty } from '../source/empty'
 import Pipe from '../sink/Pipe'
 import IndexSink from '../sink/IndexSink'
 import { disposeAll, tryDispose } from '@most/disposable'
diff --git a/packages/core/src/combinator/merge.js b/packages/core/src/combinator/merge.js
index 3630e80..5dd46d1 100644
--- a/packages/core/src/combinator/merge.js
+++ b/packages/core/src/combinator/merge.js
@@ -4,7 +4,7 @@
 
 import Pipe from '../sink/Pipe'
 import IndexSink from '../sink/IndexSink'
-import { empty } from '../source/core'
+import { empty } from '../source/empty'
 import { disposeAll, tryDispose } from '@most/disposable'
 import { reduce } from '@most/prelude'
 
diff --git a/packages/core/src/combinator/promises.js b/packages/core/src/combinator/promises.js
index 1c73e07..7108d3b 100644
--- a/packages/core/src/combinator/promises.js
+++ b/packages/core/src/combinator/promises.js
@@ -3,7 +3,7 @@
 /** @author John Hann */
 
 import fatal from '../fatalError'
-import { just } from '../source/core'
+import { now } from '../source/now'
 import { compose } from '@most/prelude'
 
 /**
@@ -19,7 +19,7 @@ export const awaitPromises = stream => new Await(stream)
  * @return {Stream<T>} stream containing promise's fulfillment value.
  *  If the promise rejects, the stream will error
  */
-export const fromPromise = compose(awaitPromises, just)
+export const fromPromise = compose(awaitPromises, now)
 
 class Await {
   constructor (source) {
diff --git a/packages/core/src/combinator/slice.js b/packages/core/src/combinator/slice.js
index 2a34b56..fe88237 100644
--- a/packages/core/src/combinator/slice.js
+++ b/packages/core/src/combinator/slice.js
@@ -3,7 +3,7 @@
 /** @author John Hann */
 
 import Pipe from '../sink/Pipe'
-import { empty } from '../source/core'
+import { empty } from '../source/empty'
 import Map from '../fusion/Map'
 
 /**
diff --git a/packages/core/src/combinator/withArrayValues.js b/packages/core/src/combinator/withArrayValues.js
index 7ff7ce3..2f3c5ae 100644
--- a/packages/core/src/combinator/withArrayValues.js
+++ b/packages/core/src/combinator/withArrayValues.js
@@ -1,5 +1,6 @@
 /** @license MIT License (c) copyright 2010-2017 original author or authors */
-import { empty } from '../source/core'
+
+import { empty } from '../source/empty'
 import Pipe from '../sink/Pipe'
 
 export const withArrayValues = (array, stream) =>
diff --git a/packages/core/src/combinator/zip.js b/packages/core/src/combinator/zip.js
index bf25b42..972adbd 100644
--- a/packages/core/src/combinator/zip.js
+++ b/packages/core/src/combinator/zip.js
@@ -3,7 +3,7 @@
 /** @author John Hann */
 
 import { map } from './transform'
-import { empty } from '../source/core'
+import { empty } from '../source/empty'
 import Pipe from '../sink/Pipe'
 import IndexSink from '../sink/IndexSink'
 import { disposeAll } from '@most/disposable'
diff --git a/packages/core/src/index.js b/packages/core/src/index.js
index 7cbe063..3d4e501 100644
--- a/packages/core/src/index.js
+++ b/packages/core/src/index.js
@@ -1,7 +1,9 @@
 /** @license MIT License (c) copyright 2016 original author or authors */
 import { curry2, curry3 } from '@most/prelude'
 
-export { just, empty, never } from './source/core'
+export { now } from './source/now'
+export { empty } from './source/empty'
+export { never } from './source/never'
 
 export { periodic } from './source/periodic'
 
diff --git a/packages/core/src/index.js.flow b/packages/core/src/index.js.flow
index 1b231a1..0f4d5f3 100644
--- a/packages/core/src/index.js.flow
+++ b/packages/core/src/index.js.flow
@@ -10,7 +10,11 @@ declare export function runEffects <A> (s: Stream<A>, scheduler: Scheduler): Pro
 
 declare export function empty (): Stream<any>
 declare export function never (): Stream<any>
-declare export function just <A> (a: A): Stream<A>
+
+declare export function now <A> (a: A): Stream<A>
+
+declare export function at <A> (t: Time, a: A): Stream<A>
+declare export function at <A> (t: Time): (a: A) => Stream<A>
 
 export type RunStream<A> = (sink: Sink<A>, scheduler: Scheduler) => Disposable
 declare export function newStream <A> (run: RunStream<A>): Stream<A>
diff --git a/packages/core/src/source/at.js b/packages/core/src/source/at.js
new file mode 100644
index 0000000..13f1d2b
--- /dev/null
+++ b/packages/core/src/source/at.js
@@ -0,0 +1,21 @@
+/** @license MIT License (c) copyright 2010-2017 original author or authors */
+
+import { propagateTask } from '../scheduler/PropagateTask'
+
+export const at = (t, x) => new At(t, x)
+
+class At {
+  constructor (t, x) {
+    this.time = t
+    this.value = x
+  }
+
+  run (sink, scheduler) {
+    return scheduler.delay(this.time, propagateTask(runAt, this.value, sink))
+  }
+}
+
+function runAt (t, x, sink) {
+  sink.event(t, x)
+  sink.end(t)
+}
diff --git a/packages/core/src/source/core.js b/packages/core/src/source/core.js
deleted file mode 100644
index f9418c0..0000000
--- a/packages/core/src/source/core.js
+++ /dev/null
@@ -1,56 +0,0 @@
-/** @license MIT License (c) copyright 2010-2016 original author or authors */
-/** @author Brian Cavalier */
-/** @author John Hann */
-
-import { disposeNone } from '@most/disposable'
-import { propagateTask, propagateEndTask } from '../scheduler/PropagateTask'
-
-/**
- * Stream containing only x
- * @param {*} x
- * @returns {Stream}
- */
-export const just = x => new Just(x)
-
-class Just {
-  constructor (x) {
-    this.value = x
-  }
-
-  run (sink, scheduler) {
-    return scheduler.asap(propagateTask(runJust, this.value, sink))
-  }
-}
-
-function runJust (t, x, sink) {
-  sink.event(t, x)
-  sink.end(t, undefined)
-}
-
-/**
- * Stream containing no events and ends immediately
- * @returns {Stream}
- */
-export const empty = () => EMPTY
-
-class Empty {
-  run (sink, scheduler) {
-    return scheduler.asap(propagateEndTask(sink))
-  }
-}
-
-const EMPTY = new Empty()
-
-/**
- * Stream containing no events and never ends
- * @returns {Stream}
- */
-export const never = () => NEVER
-
-class Never {
-  run () {
-    return disposeNone()
-  }
-}
-
-const NEVER = new Never()
diff --git a/packages/core/src/source/empty.js b/packages/core/src/source/empty.js
new file mode 100644
index 0000000..bd06d03
--- /dev/null
+++ b/packages/core/src/source/empty.js
@@ -0,0 +1,13 @@
+/** @license MIT License (c) copyright 2010-2017 original author or authors */
+
+import { propagateEndTask } from '../scheduler/PropagateTask'
+
+export const empty = () => EMPTY
+
+class Empty {
+  run (sink, scheduler) {
+    return scheduler.asap(propagateEndTask(sink))
+  }
+}
+
+const EMPTY = new Empty()
diff --git a/packages/core/src/source/never.js b/packages/core/src/source/never.js
new file mode 100644
index 0000000..dd211e1
--- /dev/null
+++ b/packages/core/src/source/never.js
@@ -0,0 +1,13 @@
+/** @license MIT License (c) copyright 2010-2017 original author or authors */
+
+import { disposeNone } from '@most/disposable'
+
+export const never = () => NEVER
+
+class Never {
+  run () {
+    return disposeNone()
+  }
+}
+
+const NEVER = new Never()
diff --git a/packages/core/src/source/now.js b/packages/core/src/source/now.js
new file mode 100644
index 0000000..b3ac367
--- /dev/null
+++ b/packages/core/src/source/now.js
@@ -0,0 +1,5 @@
+/** @license MIT License (c) copyright 2010-2017 original author or authors */
+
+import { at } from './at'
+
+export const now = x => at(0, x)
diff --git a/packages/core/type-definitions/most.d.ts b/packages/core/type-definitions/most.d.ts
index 338485c..68fbc45 100644
--- a/packages/core/type-definitions/most.d.ts
+++ b/packages/core/type-definitions/most.d.ts
@@ -24,6 +24,9 @@ export * from './combinator/transform';
 export * from './combinator/zip';
 export * from './combinator/withArrayValues';
 
-export * from './source/core';
+export * from './source/never';
+export * from './source/empty';
+export * from './source/now';
+export * from './source/at';
 export * from './source/newStream';
 export * from './source/periodic';
diff --git a/packages/core/type-definitions/source/at.d.ts b/packages/core/type-definitions/source/at.d.ts
new file mode 100644
index 0000000..8150363
--- /dev/null
+++ b/packages/core/type-definitions/source/at.d.ts
@@ -0,0 +1,4 @@
+import { Time, Stream } from '@most/types';
+
+export function at<A>(t: Time, a: A): Stream<A>;
+export function at<A>(t: Time): (a: A) => Stream<A>;
diff --git a/packages/core/type-definitions/source/core.d.ts b/packages/core/type-definitions/source/core.d.ts
deleted file mode 100644
index a49ab14..0000000
--- a/packages/core/type-definitions/source/core.d.ts
+++ /dev/null
@@ -1,5 +0,0 @@
-import { Stream } from '@most/types';
-
-export function just<A>(a: A): Stream<A>;
-export function empty(): Stream<any>;
-export function never(): Stream<any>;
diff --git a/packages/core/type-definitions/source/empty.d.ts b/packages/core/type-definitions/source/empty.d.ts
new file mode 100644
index 0000000..28d1436
--- /dev/null
+++ b/packages/core/type-definitions/source/empty.d.ts
@@ -0,0 +1,3 @@
+import { Stream } from '@most/types';
+
+export function empty(): Stream<any>;
diff --git a/packages/core/type-definitions/source/never.d.ts b/packages/core/type-definitions/source/never.d.ts
new file mode 100644
index 0000000..912fb1c
--- /dev/null
+++ b/packages/core/type-definitions/source/never.d.ts
@@ -0,0 +1,3 @@
+import { Stream } from '@most/types';
+
+export function never(): Stream<any>;
diff --git a/packages/core/type-definitions/source/now.d.ts b/packages/core/type-definitions/source/now.d.ts
new file mode 100644
index 0000000..24d5340
--- /dev/null
+++ b/packages/core/type-definitions/source/now.d.ts
@@ -0,0 +1,3 @@
+import { Stream } from '@most/types';
+
+export function now<A>(a: A): Stream<A>;
diff --git a/packages/disposable/package.json b/packages/disposable/package.json
index 8f1c0a5..c2f46a1 100644
--- a/packages/disposable/package.json
+++ b/packages/disposable/package.json
@@ -45,9 +45,9 @@
     "buba": "^4.0.2",
     "cpy-cli": "^1.0.1",
     "eslint": "^3.15.0",
-    "flow-bin": "^0.43.1",
-    "mocha": "^3.2.0",
-    "rollup": "^0.41.4",
+    "flow-bin": "^0.47.0",
+    "mocha": "^3.4.2",
+    "rollup": "^0.43.0",
     "rollup-plugin-buble": "^0.15.0",
     "rollup-plugin-node-resolve": "^2.0.0",
     "sinon": "^1.17.7",
diff --git a/packages/prelude/package.json b/packages/prelude/package.json
index 86fb45c..125716d 100644
--- a/packages/prelude/package.json
+++ b/packages/prelude/package.json
@@ -38,7 +38,7 @@
     "flow-bin": "^0.47.0",
     "mocha": "^3.4.2",
     "nyc": "^10.3.2",
-    "rollup": "^0.41.6",
+    "rollup": "^0.43.0",
     "rollup-plugin-buble": "^0.15.0"
   }
 }
diff --git a/packages/scheduler/package.json b/packages/scheduler/package.json
index ba37d26..a4f81fa 100644
--- a/packages/scheduler/package.json
+++ b/packages/scheduler/package.json
@@ -46,9 +46,9 @@
     "buba": "^4.0.2",
     "cpy-cli": "^1.0.1",
     "eslint": "^3.15.0",
-    "flow-bin": "^0.43.1",
-    "mocha": "^3.2.0",
-    "rollup": "^0.41.4",
+    "flow-bin": "^0.47.0",
+    "mocha": "^3.4.2",
+    "rollup": "^0.43.0",
     "rollup-plugin-buble": "^0.15.0",
     "rollup-plugin-node-resolve": "^2.0.0",
     "sinon": "^1.17.7",
diff --git a/packages/types/package.json b/packages/types/package.json
index 80fa9eb..794cf5a 100644
--- a/packages/types/package.json
+++ b/packages/types/package.json
@@ -31,6 +31,6 @@
   "author": "brian@hovercraftstudios.com",
   "license": "MIT",
   "devDependencies": {
-    "flow-bin": "^0.43.1"
+    "flow-bin": "^0.47.0"
   }
 }


Test patch:
diff --git a/packages/core/test/build-test.js b/packages/core/test/build-test.js
index feacc34..bff71d4 100644
--- a/packages/core/test/build-test.js
+++ b/packages/core/test/build-test.js
@@ -2,7 +2,8 @@ import { describe, it } from 'mocha'
 import { eq } from '@briancavalier/assert'
 
 import { startWith, concat } from '../src/combinator/build'
-import { just, empty } from '../src/source/core'
+import { now } from '../src/source/now'
+import { empty } from '../src/source/empty'
 import { delay } from '../src/combinator/delay'
 
 import { collectEventsFor, makeEventsFromArray } from './helper/testEnv'
@@ -40,14 +41,14 @@ describe('build', function () {
     })
 
     it('should satisfy left identity', function () {
-      const s = concat(just(sentinel), empty())
+      const s = concat(now(sentinel), empty())
 
       return collectEventsFor(1, s)
         .then(assertSingleEvent(sentinel))
     })
 
     it('should satisfy right identity', function () {
-      const s = concat(empty(), just(sentinel))
+      const s = concat(empty(), now(sentinel))
 
       return collectEventsFor(1, s)
         .then(assertSingleEvent(sentinel))
diff --git a/packages/core/test/combinator/applicative-test.js b/packages/core/test/combinator/applicative-test.js
index 4fe0cfe..991734c 100644
--- a/packages/core/test/combinator/applicative-test.js
+++ b/packages/core/test/combinator/applicative-test.js
@@ -2,25 +2,25 @@ import { describe, it } from 'mocha'
 
 import { assertSame } from '../helper/stream-helper'
 import { ap } from '../../src/combinator/applicative'
-import { just } from '../../src/source/core'
+import { now } from '../../src/source/now'
 
 const sentinel = { value: 'sentinel' }
 
 describe('ap', function () {
   it('should satisfy identity', function () {
     // P.of(function(a) { return a; }).ap(v) ~= v
-    const v = just(sentinel)
-    return assertSame(ap(just(x => x), v), v)
+    const v = now(sentinel)
+    return assertSame(ap(now(x => x), v), v)
   })
 
   it('should satisfy composition', function () {
     // P.of(function(f) { return function(g) { return function(x) { return f(g(x))}; }; }).ap(u).ap(v
-    const u = just(x => 'u' + x)
-    const v = just(x => 'v' + x)
-    const w = just('w')
+    const u = now(x => 'u' + x)
+    const v = now(x => 'v' + x)
+    const w = now('w')
 
     return assertSame(
-      ap(ap(ap(just(f => g => x => f(g(x))), u), v), w),
+      ap(ap(ap(now(f => g => x => f(g(x))), u), v), w),
       ap(u, ap(v, w))
     )
   })
@@ -29,19 +29,19 @@ describe('ap', function () {
     // P.of(f).ap(P.of(x)) ~= P.of(f(x)) (homomorphism)
     const f = x => x + 'f'
     const x = 'x'
-    return assertSame(ap(just(f), just(x)), just(f(x)))
+    return assertSame(ap(now(f), now(x)), now(f(x)))
   })
 
   it('should satisfy interchange', function () {
     // u.ap(a.of(y)) ~= a.of(function(f) { return f(y); }).ap(u)
     const f = x => x + 'f'
 
-    const u = just(f)
+    const u = now(f)
     const y = 'y'
 
     return assertSame(
-      ap(u, just(y)),
-      ap(just(f => f(y)), u)
+      ap(u, now(y)),
+      ap(now(f => f(y)), u)
     )
   })
 })
diff --git a/packages/core/test/combinator/chain-test.js b/packages/core/test/combinator/chain-test.js
index 138b091..ab55df2 100644
--- a/packages/core/test/combinator/chain-test.js
+++ b/packages/core/test/combinator/chain-test.js
@@ -7,7 +7,8 @@ import { delay } from '../../src/combinator/delay'
 import { concat } from '../../src/combinator/build'
 import { take } from '../../src/combinator/slice'
 import { drain } from '../helper/observe'
-import { just, never } from '../../src/source/core'
+import { now } from '../../src/source/now'
+import { never } from '../../src/source/never'
 
 import { assertSame } from '../helper/stream-helper'
 import { collectEventsFor, makeEventsFromArray } from '../helper/testEnv'
@@ -18,10 +19,10 @@ const sentinel = { value: 'sentinel' }
 describe('chain', function () {
   it('should satisfy associativity', function () {
     // m.chain(f).chain(g) ~= m.chain(function(x) { return f(x).chain(g); })
-    const f = x => just(x + 'f')
-    const g = x => just(x + 'g')
+    const f = x => now(x + 'f')
+    const g = x => now(x + 'g')
 
-    const m = just('m')
+    const m = now('m')
 
     return assertSame(
       chain(x => chain(g, f(x)), m),
@@ -30,7 +31,7 @@ describe('chain', function () {
   })
 
   it('should preserve time order', function () {
-    const s = chain(x => delay(x, just(x)), makeEventsFromArray(0, [2, 1]))
+    const s = chain(x => delay(x, now(x)), makeEventsFromArray(0, [2, 1]))
     const expected = [ { time: 1, value: 1 }, { time: 2, value: 2 } ]
 
     return collectEventsFor(3, s)
@@ -59,8 +60,8 @@ describe('join', function () {
 
   it('should dispose outer stream', function () {
     const dispose = spy()
-    const inner = just(sentinel)
-    const outer = just(inner)
+    const inner = now(sentinel)
+    const outer = now(inner)
 
     const s = join(new FakeDisposeStream(dispose, outer))
 
@@ -69,15 +70,15 @@ describe('join', function () {
 
   it('should dispose inner stream', function () {
     const dispose = spy()
-    const inner = new FakeDisposeStream(dispose, just(sentinel))
+    const inner = new FakeDisposeStream(dispose, now(sentinel))
 
-    const s = join(just(inner))
+    const s = join(now(inner))
 
     return drain(s).then(() => assert(dispose.calledOnce))
   })
 
   it('should dispose inner stream immediately', function () {
-    const s = just(concat(just(1), never()))
+    const s = now(concat(now(1), never()))
     return drain(take(1, join(s))).then(() => assert(true))
   })
 
@@ -85,7 +86,7 @@ describe('join', function () {
     const values = [1, 2, 3]
     const spies = values.map(() => spy())
 
-    const inners = values.map((x, i) => new FakeDisposeStream(spies[i], just(x)))
+    const inners = values.map((x, i) => new FakeDisposeStream(spies[i], now(x)))
 
     const s = join(makeEventsFromArray(1, inners))
 
diff --git a/packages/core/test/combinator/concatMap-test.js b/packages/core/test/combinator/concatMap-test.js
index 269cf83..8fec505 100644
--- a/packages/core/test/combinator/concatMap-test.js
+++ b/packages/core/test/combinator/concatMap-test.js
@@ -8,7 +8,8 @@ import * as concatMap from '../../src/combinator/concatMap'
 import { concat } from '../../src/combinator/build'
 import { take } from '../../src/combinator/slice'
 import { drain } from '../helper/observe'
-import { just, never } from '../../src/source/core'
+import { now } from '../../src/source/now'
+import { never } from '../../src/source/never'
 
 import { ticks, atTimes, makeEventsFromArray, collectEventsFor, collectEvents } from '../helper/testEnv'
 import FakeDisposeSource from '../helper/FakeDisposeStream'
@@ -20,10 +21,10 @@ const identity = x => x
 describe('concatMap', function () {
   it('should satisfy associativity', function () {
     // m.concatMap(f).concatMap(g) ~= m.concatMap(function(x) { return f(x).concatMap(g); })
-    const f = x => just(x + 'f')
-    const g = x => just(x + 'g')
+    const f = x => now(x + 'f')
+    const g = x => now(x + 'g')
 
-    const m = just('m')
+    const m = now('m')
 
     return assertSame(
       concatMap.concatMap(x => concatMap.concatMap(g, f(x)), m),
@@ -65,8 +66,8 @@ describe('concatMap', function () {
 
   it('should dispose outer stream', function () {
     const dispose = spy()
-    const inner = just(sentinel)
-    const outer = just(inner)
+    const inner = now(sentinel)
+    const outer = now(inner)
 
     const s = concatMap.concatMap(identity, new FakeDisposeSource(dispose, outer))
 
@@ -75,15 +76,15 @@ describe('concatMap', function () {
 
   it('should dispose inner stream', function () {
     const dispose = spy()
-    const inner = new FakeDisposeSource(dispose, just(sentinel))
+    const inner = new FakeDisposeSource(dispose, now(sentinel))
 
-    const s = concatMap.concatMap(identity, just(inner))
+    const s = concatMap.concatMap(identity, now(inner))
 
     return drain(s).then(() => assert(dispose.called))
   })
 
   it('should dispose inner stream immediately', function () {
-    const s = just(concat(just(1), never()))
+    const s = now(concat(now(1), never()))
 
     return drain(take(1, concatMap.concatMap(identity, s)))
   })
@@ -92,7 +93,7 @@ describe('concatMap', function () {
     const values = [1, 2, 3]
     const spies = values.map(() => spy())
 
-    const inners = values.map((x, i) => new FakeDisposeSource(spies[i], just(x)))
+    const inners = values.map((x, i) => new FakeDisposeSource(spies[i], now(x)))
 
     const s = concatMap.concatMap(identity, makeEventsFromArray(1, inners))
 
diff --git a/packages/core/test/combinator/continueWith-test.js b/packages/core/test/combinator/continueWith-test.js
index 10da3e4..07b1ab0 100644
--- a/packages/core/test/combinator/continueWith-test.js
+++ b/packages/core/test/combinator/continueWith-test.js
@@ -3,12 +3,12 @@ import { is, fail } from '@briancavalier/assert'
 
 import { continueWith } from '../../src/combinator/continueWith'
 import { drain } from '../helper/observe'
-import { just } from '../../src/source/core'
+import { now } from '../../src/source/now'
 
 describe('continueWith', () => {
   it('when f throws, should propagate error', () => {
     const error = new Error()
-    const s = continueWith(() => { throw error }, just(0))
+    const s = continueWith(() => { throw error }, now(0))
     return drain(s).then(fail, is(error))
   })
 })
diff --git a/packages/core/test/combinator/errors-test.js b/packages/core/test/combinator/errors-test.js
index c988aea..209eabd 100644
--- a/packages/core/test/combinator/errors-test.js
+++ b/packages/core/test/combinator/errors-test.js
@@ -4,7 +4,7 @@ import { fail, is, eq } from '@briancavalier/assert'
 import { throwError, recoverWith } from '../../src/combinator/errors'
 import { map } from '../../src/combinator/transform'
 import { observe, drain } from '../helper/observe'
-import { just } from '../../src/source/core'
+import { now } from '../../src/source/now'
 
 const sentinel = { value: 'sentinel' }
 const other = { value: 'other' }
@@ -20,23 +20,23 @@ describe('throwError', () => {
 
 describe('recoverWith', () => {
   it('when an error is thrown should continue with returned stream', () => {
-    const s = recoverWith(() => just(sentinel), throwError(other))
+    const s = recoverWith(() => now(sentinel), throwError(other))
     return observe(eq(sentinel), s)
   })
 
   it('should recover from errors before recoverWith', () => {
     const s = map(() => {
       throw new Error()
-    }, just(other))
+    }, now(other))
 
     return observe(eq(sentinel),
-      recoverWith(() => just(sentinel), s))
+      recoverWith(() => now(sentinel), s))
   })
 
   it('should not recover from errors after recoverWith', () => {
     const s = recoverWith(function (e) {
       throw other
-    }, just(123))
+    }, now(123))
 
     return observe(() => {
       throw sentinel
diff --git a/packages/core/test/combinator/loop-test.js b/packages/core/test/combinator/loop-test.js
index 295f1a4..1652144 100644
--- a/packages/core/test/combinator/loop-test.js
+++ b/packages/core/test/combinator/loop-test.js
@@ -4,7 +4,7 @@ import { spy } from 'sinon'
 
 import { loop } from '../../src/combinator/loop'
 import { throwError } from '../../src/combinator/errors'
-import { just } from '../../src/source/core'
+import { now } from '../../src/source/now'
 
 import FakeDisposeSource from '../helper/FakeDisposeStream'
 
@@ -43,7 +43,7 @@ describe('loop', function () {
   it('should dispose', function () {
     const dispose = spy()
 
-    const stream = new FakeDisposeSource(dispose, just(sentinel))
+    const stream = new FakeDisposeSource(dispose, now(sentinel))
     const s = loop(toPair, 0, stream)
 
     return collectEventsFor(1, s).then(() => assert(dispose.calledOnce))
diff --git a/packages/core/test/combinator/merge-test.js b/packages/core/test/combinator/merge-test.js
index 1bb05c6..8faa0ed 100644
--- a/packages/core/test/combinator/merge-test.js
+++ b/packages/core/test/combinator/merge-test.js
@@ -1,7 +1,7 @@
 import { describe, it } from 'mocha'
 import { eq } from '@briancavalier/assert'
 
-import { just } from '../../src/source/core'
+import { now } from '../../src/source/now'
 import { delay } from '../../src/combinator/delay'
 import { merge, mergeArray } from '../../src/combinator/merge'
 
@@ -13,9 +13,9 @@ describe('merge', function () {
   })
 
   it('should be associative', function () {
-    const s1 = just(1)
-    const s2 = just(2)
-    const s3 = just(3)
+    const s1 = now(1)
+    const s2 = now(2)
+    const s3 = now(3)
 
     const m1 = merge(merge(s1, s2), s3)
     const m2 = merge(s1, merge(s2, s3))
diff --git a/packages/core/test/combinator/mergeConcurrently-test.js b/packages/core/test/combinator/mergeConcurrently-test.js
index 93c4c99..ca53b8f 100644
--- a/packages/core/test/combinator/mergeConcurrently-test.js
+++ b/packages/core/test/combinator/mergeConcurrently-test.js
@@ -5,7 +5,7 @@ import { mergeMapConcurrently, mergeConcurrently } from '../../src/combinator/me
 import { periodic } from '../../src/source/periodic'
 import { take } from '../../src/combinator/slice'
 import { constant } from '../../src/combinator/transform'
-import { just } from '../../src/source/core'
+import { now } from '../../src/source/now'
 import { collectEventsFor, makeEventsFromArray } from '../helper/testEnv'
 
 const sentinel = { value: 'sentinel' }
@@ -14,7 +14,7 @@ const periodicConstant = (ms, x) => constant(x, periodic(ms))
 
 describe('mergeConcurrently', () => {
   it('should be identity for 1 stream', () => {
-    const s = mergeConcurrently(1, just(periodicConstant(1, sentinel)))
+    const s = mergeConcurrently(1, now(periodicConstant(1, sentinel)))
     const n = 3
 
     const expected = [
@@ -75,7 +75,7 @@ describe('mergeConcurrently', () => {
 describe('mergeMapConcurrently', () => {
   it('when mapping function throws, it should catch and propagate error', () => {
     const error = new Error()
-    const s = mergeMapConcurrently(x => { throw error }, 1, just(0))
+    const s = mergeMapConcurrently(x => { throw error }, 1, now(0))
     return collectEventsFor(1, s).then(fail, is(error))
   })
 })
diff --git a/packages/core/test/combinator/sample-test.js b/packages/core/test/combinator/sample-test.js
index 51962d3..2ed7c9f 100644
--- a/packages/core/test/combinator/sample-test.js
+++ b/packages/core/test/combinator/sample-test.js
@@ -4,7 +4,8 @@ import { describe, it } from 'mocha'
 import { is, eq } from '@briancavalier/assert'
 
 import { throwError } from '../../src/combinator/errors'
-import { just, never } from '../../src/source/core'
+import { now } from '../../src/source/now'
+import { never } from '../../src/source/never'
 import { sample } from '../../src/combinator/sample'
 
 import { makeEvents, ticks, collectEvents } from '../helper/testEnv'
@@ -17,7 +18,7 @@ describe('sample', () => {
     const justValue = Math.random()
 
     const f = (a, b) => [a, b]
-    const s = sample(f, just(samplerValue), just(justValue))
+    const s = sample(f, now(samplerValue), now(justValue))
 
     return collectEvents(s, ticks(1)).then(events => {
       eq(1, events.length)
@@ -44,7 +45,7 @@ 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), just(x))
+    const s = sample(Array, makeEvents(1, n), now(x))
 
     return collectEvents(s, ticks(n)).then(events => {
       eq(n, events.length)
diff --git a/packages/core/test/combinator/withArrayValues-test.js b/packages/core/test/combinator/withArrayValues-test.js
index f954af8..e5a0703 100644
--- a/packages/core/test/combinator/withArrayValues-test.js
+++ b/packages/core/test/combinator/withArrayValues-test.js
@@ -3,7 +3,7 @@ import { describe, it } from 'mocha'
 import { eq, fail } from '@briancavalier/assert'
 
 import { zipArrayValues, withArrayValues } from '../../src/combinator/withArrayValues'
-import { empty } from '../../src/source/core'
+import { empty } from '../../src/source/empty'
 
 import { collectEventsFor, makeEvents } from '../helper/testEnv'
 
diff --git a/packages/core/test/combine-test.js b/packages/core/test/combine-test.js
index 680de4c..3868f54 100644
--- a/packages/core/test/combine-test.js
+++ b/packages/core/test/combine-test.js
@@ -3,7 +3,7 @@ import { eq } from '@briancavalier/assert'
 
 import { combine, combineArray } from '../src/combinator/combine'
 import { delay } from '../src/combinator/delay'
-import { just } from '../src/source/core'
+import { now } from '../src/source/now'
 
 import { collectEventsFor, makeEvents, makeEventsFromArray } from './helper/testEnv'
 
@@ -11,8 +11,8 @@ const sentinel = { value: 'sentinel' }
 
 describe('combine', function () {
   it('should yield initial only after all inputs yield', function () {
-    const s1 = just(1)
-    const s2 = just(sentinel)
+    const s1 = now(1)
+    const s2 = now(sentinel)
 
     const sc = combine(Array, s1, delay(1, s2))
 
@@ -41,8 +41,8 @@ describe('combine', function () {
 
 describe('combineArray', function () {
   it('should yield initial only after all inputs yield', function () {
-    const s1 = just(1)
-    const s2 = just(sentinel)
+    const s1 = now(1)
+    const s2 = now(sentinel)
 
     const sc = combineArray(Array, [s1, delay(1, s2)])
 
diff --git a/packages/core/test/delay-test.js b/packages/core/test/delay-test.js
index 1c224d2..3e130d3 100644
--- a/packages/core/test/delay-test.js
+++ b/packages/core/test/delay-test.js
@@ -2,7 +2,7 @@ import { describe, it } from 'mocha'
 import { eq } from '@briancavalier/assert'
 
 import { delay } from '../src/combinator/delay'
-import { just } from '../src/source/core'
+import { now } from '../src/source/now'
 
 import { collectEventsFor } from './helper/testEnv'
 
@@ -10,7 +10,7 @@ describe('delay', function () {
   it('should delay events by delayTime', function () {
     const time = 1
     const value = Math.random()
-    const s = delay(time, just(value))
+    const s = delay(time, now(value))
 
     return collectEventsFor(time + 1, s)
       .then(eq([{ time, value }]))
diff --git a/packages/core/test/flow/index.js b/packages/core/test/flow/index.js
index 6d13d1f..49ff89e 100644
--- a/packages/core/test/flow/index.js
+++ b/packages/core/test/flow/index.js
@@ -2,13 +2,13 @@
 
 // Flow type tests
 
-import { just, switchLatest, map, constant, runEffects } from '../../src/index'
+import { now, switchLatest, map, constant, runEffects } from '../../src/index'
 import { newDefaultScheduler } from '@most/scheduler'
 
 const f: (number) => number = a => a
 
 const scheduler = newDefaultScheduler()
-const s = constant('hi')(map(f)(switchLatest(map(just, just(123)))))
+const s = constant('hi')(map(f)(switchLatest(map(now, now(123)))))
 
 runEffects(s, scheduler)
   .then(console.log)
diff --git a/packages/core/test/limit-test.js b/packages/core/test/limit-test.js
index 4c04a51..98b9772 100644
--- a/packages/core/test/limit-test.js
+++ b/packages/core/test/limit-test.js
@@ -5,7 +5,8 @@ import { debounce, throttle } from '../src/combinator/limit'
 import { zip } from '../src/combinator/zip'
 import { map } from '../src/combinator/transform'
 import { take } from '../src/combinator/slice'
-import { empty, just } from '../src/source/core'
+import { empty } from '../src/source/empty'
+import { now } from '../src/source/now'
 import { default as Map } from '../src/fusion/Map'
 
 import { atTimes, collectEventsFor, makeEventsFromArray, makeEvents } from './helper/testEnv'
@@ -41,7 +42,7 @@ describe('debounce', function () {
     })
 
     it('should be identity when source is singleton', function () {
-      const s = debounce(1, just(sentinel))
+      const s = debounce(1, now(sentinel))
       return collectEventsFor(2, s)
         .then(eq([{ time: 0, value: sentinel }]))
     })
@@ -89,7 +90,7 @@ describe('throttle', function () {
 
     it('should commute map', function () {
       const id = x => x
-      const s = throttle(1, map(id, just()))
+      const s = throttle(1, map(id, now()))
 
       assert(s instanceof Map)
       is(id, s.f)
diff --git a/packages/core/test/slice-test.js b/packages/core/test/slice-test.js
index 46e64dc..9fdf1bc 100644
--- a/packages/core/test/slice-test.js
+++ b/packages/core/test/slice-test.js
@@ -3,7 +3,7 @@ import { eq, is, assert } from '@briancavalier/assert'
 
 import { slice, take, skip, takeWhile, skipWhile, skipAfter } from '../src/combinator/slice'
 import { map } from '../src/combinator/transform'
-import { empty } from '../src/source/core'
+import { empty } from '../src/source/empty'
 import { default as Map } from '../src/fusion/Map'
 
 import { makeEventsFromArray, collectEventsFor, makeEvents } from './helper/testEnv'
diff --git a/packages/core/test/source/at-test.js b/packages/core/test/source/at-test.js
new file mode 100644
index 0000000..2041b08
--- /dev/null
+++ b/packages/core/test/source/at-test.js
@@ -0,0 +1,16 @@
+// @flow
+import { describe, it } from 'mocha'
+import { eq } from '@briancavalier/assert'
+
+import { at } from '../../src/source/at'
+
+import { collectEventsFor } from '../helper/testEnv'
+
+describe('at', () => {
+  it('should contain event at time t', function () {
+    const t = Math.floor(Math.random() * 100)
+    const expected = Math.random()
+    return collectEventsFor(t, at(t, expected))
+      .then(eq([{ time: t, value: expected }]))
+  })
+})
diff --git a/packages/core/test/source/core-test.js b/packages/core/test/source/core-test.js
deleted file mode 100644
index 97a880c..0000000
--- a/packages/core/test/source/core-test.js
+++ /dev/null
@@ -1,20 +0,0 @@
-import { describe, it } from 'mocha'
-import { eq } from '@briancavalier/assert'
-
-import { just, empty } from '../../src/source/core'
-
-import { collectEventsFor } from '../helper/testEnv'
-
-describe('just', function () {
-  it('should contain one item', function () {
-    const expected = Math.random()
-    return collectEventsFor(1, just(expected))
-      .then(eq([{ time: 0, value: expected }]))
-  })
-})
-
-describe('empty', function () {
-  it('should yield no items before end', function () {
-    return collectEventsFor(1, empty()).then(eq([]))
-  })
-})
diff --git a/packages/core/test/source/empty-test.js b/packages/core/test/source/empty-test.js
new file mode 100644
index 0000000..5a84042
--- /dev/null
+++ b/packages/core/test/source/empty-test.js
@@ -0,0 +1,13 @@
+// @flow
+import { describe, it } from 'mocha'
+import { eq } from '@briancavalier/assert'
+
+import { empty } from '../../src/source/empty'
+
+import { collectEventsFor } from '../helper/testEnv'
+
+describe('empty', () => {
+  it('should yield no items before end', () => {
+    return collectEventsFor(1, empty()).then(eq([]))
+  })
+})
diff --git a/packages/core/test/source/now-test.js b/packages/core/test/source/now-test.js
new file mode 100644
index 0000000..ce99dbe
--- /dev/null
+++ b/packages/core/test/source/now-test.js
@@ -0,0 +1,15 @@
+// @flow
+import { describe, it } from 'mocha'
+import { eq } from '@briancavalier/assert'
+
+import { now } from '../../src/source/now'
+
+import { collectEventsFor } from '../helper/testEnv'
+
+describe('now', () => {
+  it('should contain event at time 0', function () {
+    const expected = Math.random()
+    return collectEventsFor(1, now(expected))
+      .then(eq([{ time: 0, value: expected }]))
+  })
+})
diff --git a/packages/core/test/switch-test.js b/packages/core/test/switch-test.js
index 0b6619f..264e654 100644
--- a/packages/core/test/switch-test.js
+++ b/packages/core/test/switch-test.js
@@ -5,7 +5,8 @@ import { switchLatest } from '../src/combinator/switch'
 import { take } from '../src/combinator/slice'
 import { constant, map, tap } from '../src/combinator/transform'
 import { periodic } from '../src/source/periodic'
-import { empty, just } from '../src/source/core'
+import { empty } from '../src/source/empty'
+import { now } from '../src/source/now'
 import { ticks, collectEventsFor, makeEvents, makeEventsFromArray } from './helper/testEnv'
 import { runEffects } from '../src/runEffects'
 
@@ -23,7 +24,7 @@ describe('switch', () => {
     const events = []
     const push = x => events.push(x)
     const toInner = x => x === 0
-      ? switchLatest(map(just, tap(push, take(1, periodic(1, 1)))))
+      ? switchLatest(map(now, tap(push, take(1, periodic(1, 1)))))
       : empty()
 
     const s = switchLatest(map(toInner, makeEventsFromArray(0, [0, 1])))
@@ -33,7 +34,7 @@ describe('switch', () => {
 
   describe('when input contains a single stream', () => {
     it('should return an equivalent stream', () => {
-      const s = just(makeEvents(1, 3))
+      const s = now(makeEvents(1, 3))
 
       return collectEventsFor(3, switchLatest(s))
         .then(eq([
diff --git a/packages/core/test/timeslice-test.js b/packages/core/test/timeslice-test.js
index 1e87016..ae87382 100644
--- a/packages/core/test/timeslice-test.js
+++ b/packages/core/test/timeslice-test.js
@@ -4,7 +4,8 @@ import { eq, assert } from '@briancavalier/assert'
 import { until, since, during } from '../src/combinator/timeslice'
 import { take } from '../src/combinator/slice'
 import { periodic } from '../src/source/periodic'
-import { just, never } from '../src/source/core'
+import { now } from '../src/source/now'
+import { never } from '../src/source/never'
 import { delay } from '../src/combinator/delay'
 
 import { ticks, collectEvents, collectEventsFor, makeEvents } from './helper/testEnv'
@@ -15,7 +16,7 @@ import { spy } from 'sinon'
 describe('during', function () {
   it('should contain events at or later than min and earlier than max', function () {
     const stream = take(10, periodic(1))
-    const timespan = delay(1, just(delay(5, just())))
+    const timespan = delay(1, now(delay(5, now())))
 
     const s = during(timespan, stream)
     return collectEventsFor(10, s)
@@ -31,7 +32,7 @@ describe('during', function () {
   it('should dispose source stream', function () {
     const dispose = spy()
     const stream = FakeDisposeStream.from(dispose, periodic(1))
-    const timespan = delay(1, just(delay(5, just())))
+    const timespan = delay(1, now(delay(5, now())))
 
     const s = during(timespan, stream)
     return collectEvents(s, ticks(6))
@@ -42,7 +43,7 @@ describe('during', function () {
     const dispose = spy()
 
     const stream = periodic(1)
-    const timespan = delay(1, just(delay(5, just())))
+    const timespan = delay(1, now(delay(5, now())))
     const dt = FakeDisposeStream.from(dispose, timespan)
 
     const s = during(dt, stream)
@@ -57,7 +58,7 @@ describe('during', function () {
 describe('until', function () {
   it('should only contain events earlier than signal', function () {
     const stream = makeEvents(1, 10)
-    const signal = delay(3, just())
+    const signal = delay(3, now())
 
     const s = until(signal, stream)
     return collectEventsFor(10, s)
@@ -72,7 +73,7 @@ describe('until', function () {
   it('should dispose source stream', function () {
     const dispose = spy()
     const stream = FakeDisposeStream.from(dispose, periodic(1))
-    const signal = delay(3, just())
+    const signal = delay(3, now())
 
     const s = until(signal, stream)
     return collectEventsFor(5, s)
@@ -82,7 +83,7 @@ describe('until', function () {
   it('should end immediately on signal', function () {
     const dispose = spy()
     const stream = FakeDisposeStream.from(dispose, never())
-    const signal = just()
+    const signal = now()
 
     const s = until(signal, stream)
     return collectEventsFor(1, s)
@@ -95,7 +96,7 @@ describe('until', function () {
   it('should dispose signal', function () {
     const dispose = spy()
     const stream = periodic(1)
-    const signal = FakeDisposeStream.from(dispose, delay(3, just()))
+    const signal = FakeDisposeStream.from(dispose, delay(3, now()))
 
     const s = until(signal, stream)
     return collectEventsFor(5, s)
@@ -114,7 +115,7 @@ describe('since', function () {
   it('should only contain events at or later than signal', function () {
     const n = 5
     const stream = take(n, periodic(1))
-    const signal = delay(3, just())
+    const signal = delay(3, now())
 
     const s = since(signal, stream)
     return collectEventsFor(n, s)
@@ -127,7 +128,7 @@ describe('since', function () {
   it('should dispose signal', function () {
     const dispose = spy()
     const stream = take(5, periodic(1))
-    const signal = FakeDisposeStream.from(dispose, delay(3, just()))
+    const signal = FakeDisposeStream.from(dispose, delay(3, now()))
 
     const s = since(signal, stream)
     return collectEventsFor(5, s)
diff --git a/packages/core/test/transform-test.js b/packages/core/test/transform-test.js
index 544b816..de637c5 100644
--- a/packages/core/test/transform-test.js
+++ b/packages/core/test/transform-test.js
@@ -3,14 +3,14 @@ import { eq } from '@briancavalier/assert'
 
 import { assertSame } from './helper/stream-helper'
 import { map, tap, constant } from '../src/combinator/transform'
-import { just } from '../src/source/core'
+import { now } from '../src/source/now'
 
 import { collectEventsFor } from './helper/testEnv'
 
 describe('map', function () {
   it('should satisfy identity', function () {
     // u.map(function(a) { return a; })) ~= u
-    const u = just(Math.random())
+    const u = now(Math.random())
     return assertSame(map(x => x, u), u)
   })
 
@@ -19,7 +19,7 @@ describe('map', function () {
     const f = x => x + 'f'
     const g = x => x + 'g'
 
-    const u = just('e')
+    const u = now('e')
 
     return assertSame(
       map(x => f(g(x)), u),
@@ -31,7 +31,7 @@ describe('map', function () {
 describe('constant', function () {
   it('should satisfy identity', function () {
     // u.constant(x) ~= u.map(function(){return x;})
-    const u = just('e')
+    const u = now('e')
     const x = 1
     const f = () => x
     return assertSame(
@@ -44,7 +44,7 @@ describe('constant', function () {
 describe('tap', function () {
   it('should not transform stream items', function () {
     const expected = Math.random()
-    const s = tap(() => -1, just(expected))
+    const s = tap(() => -1, now(expected))
 
     return collectEventsFor(1, s)
       .then(eq([{ time: 0, value: expected }]))