cujojs / most Goto Github PK

These are a bit less well-defined, and other projects take various approaches, and even differ in how they view stream errors. For example, in RxJS stream errors effectively end the stream, and so it provides combinators (like onErrorResumeNext) for continuing with a different stream. In contrast, Bacon.js has recoverable stream errors, where you can catch individual errors in a stream and turn them back into non-errors (for example, via the mapError combinator).

First of all, I've tinkered around with RxJS, as well as Highland, and I prefer most because of the way it nails the interaction with promises. Also, your docs with the stream diagrams are just miles above the other two.

Second, I'm using most in nodejs, and I'm wondering if you have any thoughts about whether node streams have a future in most. I've been able to use most.create to get stream data through a most stream, but there really needs to be a way to pipe a most through another node stream. As well as ways to handle back pressure. Otherwise it's not really very useful, reading from a large node stream is like a firehose, and it will fill up the internal buffer and grind node to a halt after a while.

Have you had any thoughts about adapting most to node streams? Is there any priority on additions like this? I am happy mixing and matching with highland, but like I said -- I much prefer most, and it would be great to be able to do things like this with it. Maybe if I have some time I'll try to learn the concepts you have around "sources" and "sinks" to see if I can fork this repo and add something. Any guidance on that front would be greatly appreciated.

Cheers!

Hey, two questions:

Is it considered to provide browser-consumable (<ES6) output of most.js library?

Or is it already available/possible with existing bundling tools and lies on application-construction side (I understand that this question is out of most.js development scope, sorry...)?

We're probably inconsistent in a few places, e.g. Stream.fromArray vs. List.from. We should just decide which we prefer and make them match.

With:

stream:            -1-2-2-1-4-4-5->

Some might expect:

stream.distinct(): -1-2-----4---5->

But distinct only concerns adjacent duplicates. Perhaps you could use distinct for removing lifetime duplicates, and distinctUntilChanged (cf. RxJS) for removing adjacent duplicates?

The main motivation for most.fromEventWhere was that prior to 0.10, event propagation was async, and so most.fromEvent(...).tap(e => e.preventDefault()) did not work.

Since 0.10, event propagation is synchronous whenever possible. In particular, in constructs like most.fromEvent(...).filter(f), or most.fromEvent(...).tap(f), f will always be called in the call stack in which events originate. That means, for example, most.fromEvent(...).tap(e => e.preventDefault()) will work as expected.

Should we deprecate most.fromEventWhere?

Under tap in the API docs it says most.map(f, stream) -> Stream shouldn't it
read most.tap(f, stream) -> Stream ?

This turns out to be quite a useful construct:

most.fromEvent('change', node).map((e) => e.target.value).startWith(node.value);

It creates a signal representing the current value of a node. I think we should wrap this up in a convenience function (see also #24).

It may be more appropriate for such a thing to live in a new package, maybe most-dom? I can see other things fitting in there, such as MutationObserver-based attribute and content signals.

Currently, most provides both prototype and "static" versions of each combinator. The static versions should be curried to make them nicely composable. We can curry at the public API boundary (ie in the main most.js module) to avoid using curried functions internally.

It seems at best useless, and at worst dangerous :)

It's basically a continuous constant signal, but implemented as discrete events, which is problematic: infinitely many occurrences of x at infinitely small time intervals O.o ... it will happily blast out xs as fast as it possibly can.

Some better options might be:

1. Change repeat to take a time interval: repeat(period, value)
2. Change periodic to take a value (or an optional value, :/ )
3. Invent a new operation that takes a time interval and value

I am running into a problem creating a stream in a unit test.

I am trying to use create to gain access to the add function so that I can trigger events on the stream for unit testing a module that uses most streams.

I think I have a chicken and egg problem. I don't have access to add until someone register's a forEach, but I don't want to start the test until I know I have a reference to the add function. I expected the publisher function to be called synchronously when the stream was created. But the publisher function isn't invoked until someone registers interest in the stream.

Am I missing something in the API to invoke the publisher function and be sure that I have access to add before proceeding with the unit test? How can I create a stream and get reference to add and be sure that I am in a state where I can proceed with the test?

I'm looking for a way of using Most with React, in particular to handle DOM events via React assignment style. Similar solution is presented with RxJS.

My implementation using Most:

function handlerToStream (stream, event) {
  try {
    stream.source.sink.event(performance.now(), event)
  } catch (err) {
    stream.source.sink.error(performance.now(), err);
  }
}

let Button = React.createClass({

  _onClick: handlerToStream.bind(null, (() => {

    let stream = most.create(() => {});
    stream.forEach(event => console.log(event.target));

    return stream;
  })()),
  render() {

    return <button onClick={this._onClick}>click</button>
  }
});

Is there a cleaner way to make this work?

I think it's time to start exploring some other/newer tools. ESLint seems like it's quite usable now, so we should at least try it.

We need a true debounce, as it is more useful/appropriate in many situations than throttle.

There's most.fromPromise, which is the manual way of dealing with promises, so you can do:

stream.map(functionThatReturnsAPromise).flatMap(most.fromPromise);
// or
stream.flatMap(function(x) {
   return most.fromPromise(functionThatReturnsAPromise(x));
});

Both of which probably get unwieldy if you have lots of promise-returning functions. We should brainstorm some ways to make it easier.

Some thoughts:

1. RxJS automatically treats promises as observables in flatMap, iow you can pass a promise-returning function directly to flatMap. We could do something similar, probably in join(). I'm not opposed to that, just want to think through options.
2. Add a specific stream.mapAsync(promiseReturningFunction) API. Does this open the door to folks requesting other *Async variants?
3. Add a convenience API, something like stream.map(promiseReturningFunction).await() that awaits promises in the stream at that point.

Other ideas?

As a compliment to periodic(period), which lets you specify the period/interval, rate(eventsPerSecond) would let you specify the number of events per second. It'd basically just be:

function rate(eventsPerSecond) {
    return periodic(1000 / eventsPerSecond);
}

Two sides of the same coin, but might be more convenient in some cases? Not sure if rate is a good name. Other names that come to mind: eventsPerSecond, eps, fps

To be able to preventDefault correctly on DOM events, preventDefault needs to be called synchronously on the event--that is, before the event handler function returns. Because of the async nature of streams, using streamOfDomEvents.tap(preventDefault) may not do what you'd expect, since preventDefault will be called too late.

One solution would be to allow fromEvent (or a variant thereof) to accept a function that will be called on the events synchronously before they are actually added to the stream.

The name ap, while it's very short and matches the fantasy-land proposal, doesn't help with learnability or readability, imho. A better name might be applyTo, since that's what it does: it takes a stream of functions and applies them to a stream of values.

For two related reasons:

1. Consistency w/JS Array API, and
2. Learnability

Even though I think the name forEach is not great, consistency should probably win. We could potentially provide an alias that is more "stream-like", but I haven't come up with anything I really like (maybe observe or consume, but that's another discussion)

Most.js takes a functional/declarative approach, and so streams don't produce any events when there is no demand, that is when no one is observing them. We should try to be more explicit in the docs about this, since it may be confusing for folks who are accustomed to imperative.

See #64

As a way to stop observing a stream based on another signal--ie when an event appears in another stream. IMHO, this is a nicer alterative to imperative "unsubscribe" functionality. It's streams all the way down ...

Seems like a small typo
https://github.com/cujojs/most/blob/master/docs/api.md#concatmap

stream.flatMap(f): -1--2--31----2----31-2-3|

should be

stream.concatMap(f): -1--2--31----2----31-2-3|

So given this:
most.from([1, 2, 3, 4]).scan(sum, 0).forEach(console.log.bind(console))
which results in 1, 3, 6, 10 in the console but shouldn't it also include 0 which is current or initial value?

What's your view on supporting this?

With just "fromEvent" you can quickly have a chicken and egg problem where your view (re)renders when a stream made of data streams and input streams change, but you don't have a DOM Element reference the first time nor do you want to recreate the stream if it does change later.

With a fromDelegatedEvent stream, you listen at the level of a stable DOM node (some parent container) and the whole DOM hierarchy underneath it can re-render whenever the stream changes. Working with explicit DOM Element references is also not very natural when using virtual DOM libraries.

There doesn't seem to be an explicit way to close a stream. I'm thinking something like stream.close() or stream.end() would make sense, no?

 widgets
    .reduce(function(acc, widget) {
      acc.collection.push(widget);
      console.log(acc);
      return acc;
    }, {collection: []})
    .then(function(result) {
      console.log('I am KING NACKJICHOLSON behold my Widgets!');
    }, console.error);

That's what I have and the below is the console output.

{ collection: [ { id: 1, widgetId: 'nackjicholson-1', owner: 'nackjicholson' } ] }
{ collection: 
   [ { id: 2, widgetId: 'nackjicholson-2', owner: 'nackjicholson' },
     { id: 2, widgetId: 'nackjicholson-2', owner: 'nackjicholson' } ] }
{ collection: 
   [ { id: 3, widgetId: 'nackjicholson-3', owner: 'nackjicholson' },
     { id: 3, widgetId: 'nackjicholson-3', owner: 'nackjicholson' },
     { id: 3, widgetId: 'nackjicholson-3', owner: 'nackjicholson' } ] }
{ collection: 
   [ { id: 4, widgetId: 'nackjicholson-4', owner: 'nackjicholson' },
     { id: 4, widgetId: 'nackjicholson-4', owner: 'nackjicholson' },
     { id: 4, widgetId: 'nackjicholson-4', owner: 'nackjicholson' },
     { id: 4, widgetId: 'nackjicholson-4', owner: 'nackjicholson' } ] }
{ collection: 
   [ { id: 5, widgetId: 'nackjicholson-5', owner: 'nackjicholson' },
     { id: 5, widgetId: 'nackjicholson-5', owner: 'nackjicholson' },
     { id: 5, widgetId: 'nackjicholson-5', owner: 'nackjicholson' },
     { id: 5, widgetId: 'nackjicholson-5', owner: 'nackjicholson' },
     { id: 5, widgetId: 'nackjicholson-5', owner: 'nackjicholson' } ] }
I am KING NACKJICHOLSON behold my Widgets!

Is that just the intended behavior and I'm missing something? The reduce docs are a little thin. If you tell me how it works I'd be willing to make a pull to update the docs.

I am using browserify to require this module on the client and I am getting a error module ./makePromise not found from most/node_modules/when/es6-shim/Promise.js.

Any ideas?

Thank you
-David

Hi, quick question

So here's my setup

var rest = require('rest');
var mime = require('rest/interceptor/mime');
var most = require('most');
var client = rest.wrap(mime);

var stream1 = most.fromPromise(client({path:'http://reqr.es/api/users?page=3'}));
var stream2 = most.fromPromise(client({path:'http://reqr.es/api/users/23'}));

stream1.concat(stream2)
.flatMapError(function(e){
   return most.of('hello');
})
.observe(console.log.bind(console));

So stream2 will fail with a 404 error not found.

Shouldn't my console.log then show the successful data from stream1 and hello ?

Thank you for the help in advanced

The perf test infrastructure is pretty good, but it would be good to use a well-known package, like benchmarkjs. It may also reduce some maintenance on perf tests.

Users can use most.create, but it might be worth adding a helper module. The helper could probably rely on the fairly simple interface provided by stomp.js, ie bring your own stomp client as long as it supports that interface.

This seems like it belongs in its own repo, eg most-stomp or something. So you'd bower install --save most most-stomp if you want to use streams + stomp. In #31, we're adding WebSocket support in the most package (granted, in a separate module). However, STOMP seems a little farther removed, hence maybe a separate package.

FYI, looks like both Kefir and Bacon are adding support for transducers based on the potential performance gains.

baconjs/bacon.js#444

Apparently Rx already had support.

https://github.com/Reactive-Extensions/RxJS/blob/master/doc/gettingstarted/transducers.md

Just wanted to put this on your radar if it wasn't already.

See #74. Similar to the proposed periodic(period, value), delayed(delayTime, value) is kind of the single-value equivalent. I can see it being very useful with until, since, and during.

One potential problem is that we have most.delay(delayTime, stream) and stream.delay(delayTime) which are similarly named, but do something different: they timeshift a whole stream "to the right" by delayTime.

It seems as though streams which are nested using flatMap are not closed out correctly. Code sample:

var most = require('most'),
  jQuery = require('jquery'),
  socket = require('sockjs')('/api/ws'),
  client = require('stomp').over(socket),
  connect = jQuery.Deferred();

client.connect({}, function() {
  connect.resolve();
});

function getResource(resource, headers) {
  return most.fromPromise(connect).flatMap(function() {
    return getSnapshot(resource, headers).flatMap(function(data) {
      return getUpdates(resource, data, headers);
    });
  });
}

function getSnapshot(resource, headers) {
  return stream('/resource/' + resource, headers)
    .take(1);
}

function getUpdates(resource, data, headers) {
  return stream('/topic/' + resource, headers)
    .startWith([])
    .scan(patch, data);
}

function patch(data, patch) {
  return jiff.patch(patch, data);
}

function stream(destination, headers) {
  return most.create(function(add) {
    var subscription = client.subscribe(destination, function(msg) {
      add(JSON.parse(msg.body));
    }, headers);
    return subscription.unsubscribe.bind(subscription);
  })
}

When running code such as:

getResource('resource').take(1).observe(console.log.bind(console));

I expect both streams which are created (getSnapshot and getUpdates) to be disposed of. Currently only the stream generated by getSnapshot is disposed of correctly while the stream generated by getUpdates remains un-disposed.

The JSDoc is pretty comprehensive, but we need to publish real, readable docs soon.

The ones that come to mind immediately are numeric ranges, ie all integers from 0 to 99, or all decimal numbers from 0 to 9, stepping by 0.25. They're easy to create using most.iterate, but maybe slightly verbose without arrow functions:

// ES5
most.iterate(function(x) { return x + 1; }, 0).take(100); // integers 0 - 99
most.iterate(function(x) { return x + .25; }, 0).take(10 / .25); // 0, .25, .5, .75, 1.0, 1.25 ...

// ES6
most.iterate(x => x + 1, 0).take(100); // arrows ftw
most.iterate(x => x + .25, 0).take(10 / .25); // yay arrows, division is still annoying

In writing that out, I also noticed how annoying it is to have to divide to do the take correctly :(. So, for example, maybe we could provide something like most.range that hides the details:

most.range(0, 100); // integers 0-99
most.range(0, 10, .25); // 0, .25, .5, .75, 1.0, 1.25 ...

There may be other useful streams, but numbers are the ones the came to mind first, and numeric ranges are generally useful in lots of situations.

Are there other predefined streams that seem like they'd be broadly useful? Should we just provide numeric ranges? Would it be better to stick with base building blocks for now? Is this the slippery slope that leads to a huge, unwieldy API?

This is related to cujojs/when#373. It seems as though commit 25bc8b7 broke my Browserify build:

Error: Cannot find module 'when/lib/decorators/unhandledRejection' from '.../app/bower_components/most/lib'
Error: Cannot find module 'when/lib/Promise' from '.../app/bower_components/most/lib'

Prior to 0.8.5, I was able to include mostJS as part of my build by specifying the following in package.json:

  "browser": {
    "most": "./app/bower_components/most/most.js"
    "when/es6-shim/Promise": "./app/bower_components/when/es6-shim/Promise.browserify-es6.js"
  }

Now I can get around the Promise requirement by stubbing when/lib/Promise, but I can't stub unhandledRejection because there does not seem to be a non-browserified copy of it in the bower repository. Would installing via NPM be a viable workaround for now?

Any thoughts?

Streams are inert until there is some indication of demand. Because they are demand driven, no events flow through a stream until you either observe or reduce it. Both require you to provide a function to do something with the events.

Sometimes you may want to create demand without providing a function, especially when dealing with the DOM which almost always requires some side-effectual steps upstream. For example, the circles example creates a whole network of streams that move the circles via DOM side-effects. Those streams get flattened into a single stream using join. The stream needs to be consumed with a noop function to make the example "go".

So, should we add an API for this? If so, I can see two options:

1. Add a new API (maybe named drain?) to create demand without needing to provide a function, or
2. Refactor observe to be varargs. That is, make observe(), act like observe(noop)

If I had to pick, I'd pick 1 because I dislike varargs. Honestly, though, I'm not sure this is even needed ... maybe there's a better way to write the circles example that avoids the silly noop.

When the outer stream is forcibly ended, eg using take, join current waits for inner "inflight" events (represented by promises) to arrive before ending. Instead, it should abandon those inflight events and dispose the inner streams immediately.

See #41

For 1-argument functions, most.map can act as lift if it is curried (see #30). But for multi-arg functions, combine can be used to lift. For example:

function add(a, b) {
    return a + b;
}

var z = add(x, y);

That could be a function that takes two numbers and returns the result of adding them (or two strings, etc. but let's stick with numbers for now). Here's how you can get a function that adds two time-varying numbers and returns a time-varying results that changes whenever either of its inputs changes:

// Suppose x and y are streams of numbers, aka time-varying numbers
var z = most.combine(add, x, y);

So, lift could simply wrap that up:

function lift(f) {
    return function() {
        return most.combine.apply(void 0, coerceToArrayAndPrepend(f, arguments));
    };
}

Then you can use it to program with time-varying values:

var liftedAdd = most.lift(add);

// Suppose x and y are streams of numbers, aka time-varying numbers
var z = liftedAdd(x, y);

So i have this

most.fromEvent('click', plusBtn).map(function() {
    return 1;
});

but wouldn't it be nice to do

most.fromEvent('click', plusBtn).mapTo(1);

which would be shorthand for the above or allow other than function similar to how bacon does it

most.fromEvent('click', plusBtn).map(1);

Right now we have stream.sampleWith(sampler), which works nicely. It's fairly narrow in that it allows sampling one stream, and passing through the value directly.

One question this raises is how to sample 2 streams or N streams. Say you wanted to add the values of two inputs, and display the result only when the user clicks a button. You could combine the inputs, and then sample the resulting stream:

most.combine(add, s1, s2).sampleWith(buttonClick);

To cut down the noise, you might do:

most.combine(add, s1.distinct(), s2.distinct()).sampleWith(buttonClick);

But that's still doing extra, unnecessary adds: each time s1 or s2 changes, add will be called, even while the user isn't clicking the button. If add is trivial, then you probably don't care. On the other hand, it is doing extra work and throwing it away, which could be especially bad if add is nontrivial.

Another option might be to flip the whole thing around in a "sample that's like combine" combinator. Like:

buttonClick.sample(add, s1, s2);

That allows sampling multiple inputs, and would only call add when the button is actually clicked, no matter how often s1 and s2 change.

Interestingly, it seems like both combine and sampleWith could be reimplemented using sample, although it's not clear the perf impact this may have on combine.

function combine(f, ...streams) {
    return merge(streams).sample(f, streams);
}

function sampleWith(sampler, stream) {
    return sampler.sample(identity, stream);
}

Personally, I prefer flatMap, but in recent discussions, TC39 seems to be preferring to use the name chain, and fantasy-land uses chain, too.

I currently have 2 small examples of using most.js to do some interesting things. I'll probably create more, and it'd be great it other folks did too :)

One question, though, is where to keep them. Seems like there are at least three options:

1. Keep them all in the most.js repo, eg in an "examples/" folder, or
2. Keep them all in a separate most-examples repo, or
3. Keep each example in its own repo

In my mind, these examples are all relatively small, so I feel like option 3 spreads them out too much. I'm leaning toward 1 or 2, but want to get other thoughts and opinions.

So i know there are talks about depreciating fromEventWhere #86 which i am +1, if you can have a smaller api that is compose-able rather than all inclusive that seems better especially if fromEventWhere seemed to be a temp solution.

Nonetheless in the API you have

var form = document.querySelector('.container');
most.fromEventWhere(function(e) {
        return e.matches('.toggle-button') && e.preventDefault();
    }, 'click', container)
    .forEach(doSomething);

I believe you mean the var form to be var container
and i do not believe it's possible to do element.matches since the argument e is event correct?

Brian, why cujo? What makes it special over, say, bacon.js and others?

Thanks :-)

My experience with buster has been very good. Even so, it's worth exploring other options, like mocha and tape.

I'm kind of keen on tape because it seems a bit lighter weight.

There is already most.of which contains a single event whose time is basically "min". IOW, as soon as you ask for it (eg via observe) it will arrive. It might be useful to be able to create an event that will arrive at a specific future time. Maybe something like:

var s = most.at(time, value);

We already have stream.delay:

var s = most.of(value).delay(delayTime);

Though this is slightly different: It gives you an event that always arrives delayTime after you ask for it. It does not give you a single event anchored at a particular time. That's what I'm proposing here: a single event, anchored at a particular time.

Open questions:

Should the stream be "hot"? That is, observers that start observing before or at time will see the event once time arrives, but those that start observing after the time will not--to them, the stream will appear to be empty and end immediately. Or should the stream be "cold", in that all observers will see the event: those who start observing before or at time will see the event once time arrives, and those who start observing at or after time will always see the event immediately (even though it's .time field will be in the past).

The latter seems odd, but does give slightly more control--for better or worse 😄. We already have an internal operation that prunes past events. We could expose that to allow devs to choose: most.at(time, value) or eg most.at(time, value).skipOldEvents(olderThanTimeDelta).

Right now stream.ap is defined as a cross-product. This is a classic definition for finite lists, but is kind of useless for infinite streams: given fs.ap(xs) where fs is an infinite stream of functions, and xs is an infinite stream of values, you simply end up with the same result as xs.map(f0) where f0 is the first function in fs. So, the cross-product is basically impossible to observe:

fs.ap(xs) -> [f0(x0), f0(x1), f0(x2), ...]

A potentially more useful definition might be a zip-like applicative, ie pair-wise applying the fs to the xs. That is:

fs.ap(xs) -> [f0(x0), f1(x1), f2(x2), ...]

Many pairs, triples, etc. patterns of combinators can be simplified algebraically. For example, s.map(f).map(g) is equivalent to s.map(g•f), where • is function composition. Thus a sequence of N map operations can be reduced to a single map. Similarly, a sequence of N filter operations can be reduced to a single filter using predicate composition (ie p(x) && q(x)).

More sophisticated simplifications are possible as well. For example, s.filter(p).map(f) can be simplified to a "filtermap" operation, which itself then composes on the left with additional filter ops and on the right with additional map ops. IOW filtermap.map can be reduced again to a single filtermap.

Rather than have each combinator know how to combine itself with any other combinator, I think these rules should be applied from the outside somehow. For example, some sort of "compiler" which can analyze and "rewrite" combinator sequences. It could be built from an extensible set of matching rules, where each rule is able to recognize a specific combinator sequence, such as .map.map and substitute a simplified/optimized replacement.

The ability to imperatively push items into a stream is useful in adapting sources like DOM events, EventEmitters, etc.