kefirjs / kefir Goto Github PK

This fiddle demonstrates the problem:

http://jsfiddle.net/mjr12grq/2/

Bacon and Rx define a fromArray method that works exactly like Kefir.once, but emits multiple items passed in as an array.

Kefir.fromArray([1, 2, 3]).log();
<value> 1
<value> 2
<value> 3
<end>

Would be nice to have.

Currently a property might have a current value and a current error at the same time. I think this is wrong. When property gets an error, this error should became current error, but the current value should be removed (if property has one), and vise versa.

I've tried many things but I can't seem to be able to introduce an error into a stream.

For example, if I get a weird value in a stream that I want to handle up the stream, I'd like to create an error and continue. There seems to be no obvious way to do this without flatMaping a error stream.

Some things I tried:

Kefir.constant('hello').
map(function(x) { throw new Error(x); }).
onError(function(e) { console.error('GOT ERROR', e) });
// Above doesn't work, works just like a throw anywhere else

Kefir.constant('hello').
map(function(x) { return new Error(x); }).
onError(function(e) { console.error('GOT ERROR', e) });
// Does not work, get an Error object in _current 

Kefir.constant('hello').
flatMap(function(x) { return Kefir.constantError(new Error(x)); }).
onError(function(e) { console.error('GOT ERROR', e) }).
onValue(function(v) { console.log('got value', v); });
// Works in general

Would love a method or some way to trigger this without having to flatMap a stream. Any advice?

Also, in the above flatMap scenario, if I don't attach a onValue handler, I still don't get the error. I'm assuming that because of the lazy nature of Kefir the _active property only is true when there is a onValue handler?

Thanks!

I am thinking of removing support of special form of callbacks "array functions". It should simplify lib code and improve performance a bit.
Probably no one ever used them anyway except for me :)

Any thoughts on the subject?

Is there a way to do something like the following?

var x = Kefir.constant(1);
var y = Kefir.never();

Kefir.isProperty(x); // returns: true
Kefir.isStream(y);  // returns true

Kefir.isProperty(y); // returns: false
Kefir.isStream(x); // returns: false

Kefir.isProperty({}); // returns: false
Kefir.isStream({}); // returns: false

If this worked regardless of which instance of Kefir the stream of property comes from I think that would be really useful for testing.

What I am doing at the moment is testing the .__proto__._name of the stream and or property.

Hi @pozadi

I've created a way to map a stream to a WebWorker like this:

In your main file:

Kefir.emitter()
     .mapWithWorker(new Worker('worker.js'))

In your worker.js:

Kefir
    .fromMessage()
    // do stuff
    .toMessage();

Might be an interesting thing to add to the core. (Or do you prefer to keep the core mean and lean?)

See https://github.com/rpominov/kefir/issues/59#issuecomment-77556464

I'm exploring the use of Kefir to create an isomorphic Flux implementation. My Kefir property will be used on a JS server, serialized to JSON, then deserialized back into a Kefir property on the client.

I see that Kefir implements onValue and offValue. The presence of both methods leads me to believe there's room for a memory leak if someone calls onValue without calling offValue. Is this correct?

It'd be nice to have the ability to pull the last value from a property without adding a listener - something like this:

var serializedState = JSON.stringify(myProperty.toValue());

I'd get a single sync lookup of the property's last value and could forget about the property after that on the server.

When using scan, I can pass a seed value as the second argument. In most cases, the seed appears to be emitted as the first event. However if the input stream contains a 'current' value, the seed is only used on the first stateful transition and never emitted.

As an example:

function add(x,y){return x+y};
var head = Kefir.sequentially(0, [1])
var tail = Kefir.sequentially(0, [10,20])
Kefir.concat( [head, tail] ).scan( add, 100 ).log();

This scan function outputs the seed '100', and then accumulates the rest of the values correctly.

[concat.scan] <value:current> 100
[concat.scan] <value> 101
[concat.scan] <value> 111
[concat.scan] <value> 131
[concat.scan] <end>

However when the input already has a value marked as 'current'

 var head = Kefir.constant(1)
 ....

The seed value is used as the initial state, but is never emitted

[concat.scan] value:current 101
[concat.scan] 111
[concat.scan] 131
[concat.scan]

I've been scouring the documentation trying to work out if this is intended behaviour or not - it seems incorrect, and my intuition would be that the seed should never be emitted directly.

(Using version 1.1.0 from NPM)

There was not much time since v1.0.0 was released but there is number of breaking changes that we should make. And although there is not many of them, I think they are important and we shouldn't wait to much before there will be more changes for such "big" release. So I guess we'll release v2.0.0 soon.

Here is current list of changes to be done: https://github.com/pozadi/kefir/milestones/v2.0.0

If you think there is something else to be done, please share before it too late :)

Hi...I'm trying to build a minimal todo list, my first idea was build the todoList like an emitter, then use a newTodo emitter and sampledBy for join both emitter and update the list...

var _todos = Kefir.emitter(); //maybe it must be a property
var _newTodo = Kefir.emitter();

Kefir.sampledBy([ _todos ],[ _newTodo ],function(_t,todo){
    _todos.emit(_t.concat(todo));
});
_newTodo.emit(...)

this is really ugly...the sampledBy has side effect and actually doesnt works, Baconjs has an update method which try to solve this "side effect"...(looks the ufo sample code)

I really like how minimal is kefir, but I'm not sure how can solve it...can you give me a clue about how could you solve it in an elegant way or maybe you can add a TODO example like those from TODO mvc...thanks!!!....

Regular flatMap is like this:

But I want something like this:

The equivalent with Javascript Arrays would be:

var g = function (xs, f) { 
    return xs.map(function (x) { return [x, f(x)]; })
        .reduce(function (x, y) { return x.concat(y); }); 
};

and with Haskell Lists would be:

g :: [a] -> (a -> a) -> [a]
g xs f = concat $ map (\x -> [x, f x]) xs

I think it could be done something like this:

Observable.prototype.mergeFlatMapLatest = function(fn) {
      var other = this.flatMapLatest(fn);
      this.merge(other);
};

I don’t think that is the best way of doing it but it would let you do some really cool things like repeated long polling:

var streamFromPromise = function (promise) {    
    return K.fromBinder(function(emitter) {
        promise.cancellable()
            .then(emitter.emit)
            .catch(emitter.emit);

        return promise.cancel;
    });
};    

streamFromPromise(getChangesSince(0))
    .mergeFlatMapLatest(function getSubsequentChanges(changes) {
        return streamFromPromise(getChangesSince(changes[“last_seq”])
            .mergeFlatMapLatest(getSubsequentChanges);
        });
    })
    .pluck('changes')
    .flatMap(function(xs) { return Kefir.sequentially(0, xs); })
    .log();

Which is a pattern which can be applied to all sorts of asynchronous actions.

Maybe there is a way of making merge more composable or perhaps this can be done with withHandler somehow?

I've watched a great talk by @sebmarkbage http://www.youtube.com/watch?v=4anAwXYqLG8 lately, and also work on a React+Flux project with relatively large amount of boilerplate code. And after thinking about it for some time I'm convinced that boilerplate code is not a problem at all, as far as it stays simple.

Kefir has some methods that are basically short hands for two lines of boilerplate code, and I now think it was a mistake to introduce them. It doesn't worth it to maintain docs/tests for them, and also library users will only benefit from not having them.

For example we have .tap which can be easily replaced by .map. Compare this two code snippet:

stream.tap(function(x) {
  sideEffect(x);
});

stream.map(function(x) {
  sideEffect(x);
  return x;
});

First of all, do we need .tap when .map works just fine for this use case? But also, which snippet is easier to understand if you didn't read docs for neither .map nor .tap? To me it pretty clear what happening in .map version, but for .tap I'd better check the docs.

So I think we should remove some methods from the API, here is the list:

• repeatedly (alternative Kefir.repeat(function() {return Kefir.sequentially(/*...*/)}))
• mapTo
• pluck
• invoke
• not
• timestamp
• tap
• and
• or

Of course the removing won't be fast to not break existing code, I think we'll only deprecate them now, but remove only in v3.0.0 or later.

Currently .changes removes current values/errors only from properties, but for streams it just returns same stream. As streams also might have current values/errors, I think .changes should remove them from streams too.

This suggestion may be more practical since kefir is all about composing functors, filters, and other transformations.

Transducers allow transformations to be pipelined so that no intermediate results have to be created. According to benchmarks of one implementation it beats out a lazy approach, at least possibly because Transducer transforms are very friendly to being inlined by a JavaScript engine.

First off benchmarks:
http://jlongster.com/Transducers.js-Round-2-with-Benchmarks

Second where they came from:
http://blog.cognitect.com/blog/2014/8/6/transducers-are-coming

Another JS implementation from the clojure folks that announced their Transducer pattern this year:
https://github.com/cognitect-labs/transducers-js

Kefir.js, like Bacon.js, doesn't offer an easy way to get the current value from a property:

property.currentValue()

This is for reasons of purism, I believe. I.e., we're not supposed to want to get the current value that way. Is that correct? If so, could you explain the benefits of this philosophy?

In any case, I would like to request this as a feature. There is already a way to twist the public interface to do it anyway:

var current;
var fn = function (v) { current = v };
property.onValue(fn);
property.offValue(fn);

Why not provide a method or (read-only) property to do this the easy way?

Any thoughts on adding functionality similar to combineTemplate from Baconjs?

Test code:

'use strict';

var Kefir = require('kefir');

var thing = Kefir.fromBinder(function (emitter) {
    console.log("subscribed");
    emitter.emit(1);

    var timer = setTimeout(function () {
        emitter.emit(2);
    }, 1000);

    return function () {
        console.log("unsubscribed");
        clearTimeout(timer);
    };
});

thing.take(1).onValue(function (val) {
    console.log(val);
});

The unsubscribe callback is never called

According to this issue, the subscriber to 'thing' should detach automatically after 1 item, so maybe this is a bug with fromBinder?

Interestingly, if I change it to take(2), the callback is called

It would be quite useful, for example, in a .withHandler handler, to be able to pass an event directly, and / or to plug a stream into an emitter:

stream.withHandler(function (emitter, event) {
    if (event.type === 'end') {
        emitter.plug(Kefir.later(1000, 1)); // a 'last minute concat'
    } else {
        emitter.send(event); // just pass the event
    }
})

Without this, some tedious branching-code would need to be written.

Is it possible to get current value of Property?
Also, is there any way to check if Observable is Property?
It'll be very useful to have.
Just created very simple extension for React.js. But it depends on ._current on Property. And I can't think up solution that will avoid this with public methods on observable

Consider this example:

A:                ----1-----e-----2-----
B:                -------3-----4-----5--
combine(A + B):   -------4--e--5--6--7--

The 5 value after the error combined as 1 + 4. The 1 is the latest value from A, and combine uses it
ignoring the fact that there was a error after it.

I think this behavior is not correct, combine should continue to emit errors on new values from B until there is a value in A. In other words third value should be combined from e and 4 which result in e. So the correct output should be:

combine(A + B):   -------4--e--e--6--7--

The .sampledBy method also affected by this:

A:                ----1-----e-------2----
B:                -------•-----•--•---•--
A sampled by B:   -------1--e--1--1---2--

should be:        -------1-----e--e---2--

I think we should fix this, but here is a question "is the current behavior useful for anybody?", because when it will be fixed it will be tricky to create current behavior.

I am going to release 1.0.0 in near future. There won't be much changes from current latest version, I just think the API is stable enough to start using real SemVer.

The current list of changes for 1.0.0 is:

• move jQuery plugin to a separate repo
• remove deprecated .withDefault method

If you think there is something else that should be done, please comment on this issue.

... so properties will not store potentially outdated current value. This was originally suggested by @raimohanska here baconjs/bacon.js#536 (comment)

Have you thought about adding a bunch of aliases and such so that kefir becomes a drop-in replacement for baconjs?

We're using Bacon heavily in our codebase right now and running into some performance issues. I'd love to try out kefir, but the API differences makes that a pain. I don't know for sure that kefir will resolve the perf issues, so having to do the work of migrating some code over in an exploratory way is a big barrier.

I think this is something that lodash did extremely well. They had a "pure" version, and then an underscore compatible version - that made the migration extremely smooth and they're benefiting greatly for it.

I've seen the tree demo. Quite impressive. I'd be curious to know what makes Kefir.js fast / Bacon.js slow.

What do Kefir users sacrifice for this increased speed, if anything?

I'm having an issue with the scan method but I am not sure if it just my lack of understanding or if it is a bug.

Here is a stripped down version of what I am trying to achieve:

var Kefir = require('kefir'),
    emitter = Kefir.emitter(),
    scan = emitter.scan(function (acc, value) {
      acc.push(value);
      return acc;
    }, []),
    merged;

emitter.emit(1);
emitter.emit(2);

merged = scan.take(1).concat(emitter);

setTimeout(function () {
  emitter.emit(3);
  emitter.emit(4);
}, 10);

merged.log('merged');

From this I would expect the following log:

// merged <value:current> [ 1, 2 ]
// merged <value> 3
// merged <value> 4

But instead I get:

// merged <value:current> []
// merged <value> [ 3 ]
// merged <value> 4

If I add a subscription to scan with scan.log(); or scan.onValue(function () {}); before emitting anything I get the expected result.

Hi,

Why is Emitter not available from Kefir namespace? I'm building a library which defines objects which inherit from Stream and Property. Today I wanted to extend a new object using Emitter.prototype, however it turned out it's not available.

Would you consider a PR which adds Emitter to Kefir object?

Regards,
Adam

I suppose that Kefir was start as a funny hobby. So thats why it has this name.
Since for now its a serious project it requires serious name. There are two reasons for it:

1. Kefir is looking ugly (sorry, but extremely ugly)
2. Its russian word so it means nothing for most part of projects audience

I suppose that you can choose more nice looking and more semantic heavy name. Thank you :)

I'm trying to gather data from an html document; some of the data is static or periodic, but at the lowest level I want to associate these values with items in a list. I've created some helpers to give me streams and data, but I'm having some difficulty when combining them.

Example: in the header of the page is a span containing a location. It occurs only once on the page, but it occurs before any of the list items I'm interested in. I do something like this:

var location = select(tokenized, '#header span.city')
    .flatMap(innerText)
    .toProperty();

var item = select(tokenized, 'ul.results a')
    .flatMap(function (obs) {
        var hrefs = obs.flatMap(attr('href')),
            innerText = obs.flatMap(innerText);

        return Kefir.zip([location, href, innerText]);
    });

item.onValue(function (a) {
    console.log(a);
});

Run as-is, I get nothing. But if I call location.onValue(function () { }); it works as expected. I would expect Kefir.zip to act as a subscriber on 'location', but it appears not to. Is this a bug or intentional? What should I be doing instead, if intentional?

In Bacon the .mapError method converts errors to values (i.e. Error-events to Next-events in bacon terminology).

I'm not sure it a right behavior, in my opinion .mapError should just map one error to another error. And maybe we should have another method that converts errors to values.

Any thoughts?

Is there any functionality like Rx's .subscribe() or streams .pipe()? Essentially I want to be able to write Observables in a way that a consumer can compose them together easily.

I'm going to add ability to omit the seed argument in operations that requires a seed value (.diff, .scan, .reduce) so the first value of a stream will be used as a seed, and all rest will be handled normally.

But there is an issue, .diff(seed, fn) method allows to omit the fn argument, so if only one argument passed it will be hard to recognize if it seed or fn (since a function may be used as seed, and an array may be used as fn). Other operations may will also alow to omit fn in the future.

The solution might be to add new methods like .diffWithoutSeed but it certainly not very good name for a method, and here is where I need help! Any ideas on names for those new methods?

Lets say we have some simple high-order function and Emitter. And we want to use them together like:

function apply(f) {
  return function(x) {
    return f(x);
  }
}

var emitter = Kefir.emitter();
apply(emitter.emit)(1);

But it fails with exception:

Uncaught TypeError: Cannot read property '_send' of undefined kefir.js:2234(anonymous function) @ VM80:4(anonymous function) @ VM80:9InjectedScript._evaluateOn @ VM61:888InjectedScript._evaluateAndWrap @ VM61:821InjectedScript.evaluate @ VM61:687

The workaround is to bind this for emit, but this kind of annoying:

apply(emitter.emit.bind(emitter))(1);

Bacon has a limitation that multiple versions of Bacon don't work together, does kefir have this same limitation?

An example:

• a library uses kefir streams internally
• the library accepts streams (rx, bacon and kefir) and "munges" them together
• obviously the rx and bacon streams need to be converted to kefir, but do the kefir streams have to be converted to the library's internal version of kefir?

Just curious if there is any use in building off the foundation of the excellent immutable.js? Would require an internal re-write to do this, but maybe there's a big benefit to leveraging the immutable and equality implications provided by immutable.js.

Thoughts?

I've just started taking kefir for a spin, and I have a question about ending streams derived from DOM events. there appear to be a couple of methods of attaching event handlers (no $ for me) - K.fromBinder and K.fromEvent. How do I detach the event listener when using K.fromEvent? I've inspected the proto chain, and I've tried firing various "off" methods - but the stream still emits the events from the element:

https://gist.github.com/sunwukung/69c1be92215112e93a87

I realize this is a breaking change, and that it's entirely possible my use-case isn't quite in-line with what I should be doing (read: I'm a newb at this style of programming!).

However, I don't really see the utility of onValue being a chainable interface- wouldn't most of the desirable use-cases for it be better suited by tap instead? (discounting that tap doesn't subscribe)

I guess what I'm really looking for is an easier way to use ES6 style lambdas to subscribe to streams in my view components. E.g. I have a stream that's shared by multiple views- when the view mounts, I attach an onValue, and when a view unmounts, I need to unsubscribe so the stream isn't active anymore. take and takeWhile don't really seem to fit that use-case, and lambdas don't have names, so offValue isn't really an option either.

My proposed change would let me do something along the lines of:

/// ...
// view setup
componentDidMount() {
  this.offStream = MyStream
    .onValue(e => this.setState({data:e});
},
// view teardown
componentWillUnmount() {
  this.offStream();
}
/// ...

Am I missing something obvious? Does this make sense for how Kefir ought to be used? Or is there a valid reason for onValue to keep the current fluent interface that I'm not appreciating?

Any thoughts of creating a fromPromise to support event streams for promise objects?

I'm somewhat new to FRP, and I only have experience with ReactiveX (specifically RxJS). I'm looking to make the move to Kefir. The main advantage over RxJS is that Kefir allows to convert error events to values, and that error events do not halt the stream.

However, when porting some of my code to Kefir, I realized that the functions passed to map, flatMap, etc. do not get the index alongside the value. In RxJS, you can do the following:

observable.flatMap(function(value, index) {
  // index is now 0, 1, etc.
});

Any reason for this design decision? Could it be added? Or is there any known workaround that you could suggest? The only "solution" I see is to keep an index variable from the closure around that call, and increment it myself:

var index = 0;
observable.flatMap(function(value) {
  // do your stuff
  index++;
  // return result
});

But besides the lack of elegance, I'm worried about any potential issues with thread safety.

Currently, the behavior of pool/bus is to plug in all observables, regardless of whether the instance was previously added or not:

var pool = Kefir.pool().log()
var obs = Kefir.emitter()
pool.plug(obs)
pool.plug(obs)
pool.plug(obs)
obs.emit('test')

Outputs:

[pool] <value> data
[pool] <value> data
[pool] <value> data

I'm new to Kefir/RP, so this might as well be the desired behavior, not a bug. In any case, consider allowing pools to only maintain a unique set of source observables, perhaps via a separate Kefir.poolSet function?

var pool = Kefir.poolSet().log()
var obs = Kefir.emitter()
pool.plug(obs)
pool.plug(obs)
pool.plug(obs)
obs.emit('test')

Outputs:

[pool] <value> data

Kefir has feature that allows lazy subscription to observable sources http://pozadi.github.io/kefir/#active-state Other libraries also has this feature, Bacon works same way, and in RxJS one is able to turn it on.

This feature is pretty great, it allows to achieve good performance in case of creating a lot of observables for later use. For example one can safely create a mouseMoves stream, then attach to it a .map(heavyWork), but until result stream has at least one subscriber the heavyWork function won't be called (we won't even subscribe to mousemove DOM event).

Although this feature creates some annoying issues with stateful observables. Here is some links describing the problem:

• baconjs/bacon.js#536
• https://github.com/pozadi/kefir/issues/33
• http://pozadi.github.io/kefir/#active-state (the ☝️ part)

One of possible solutions is to make all observables active by default (maybe with ability to optionally turn on lazy subscribing). This of course a big change to how library works, and for now it feels like lazy subscribing is basically a good thing, and we shouldn't disable it. But maybe disabling it is actually the way to go.

I created this issue to discuss all possible pros and cons, and to collect all information on the topic here.

Kefir releases Zalgo. See here for why that is bad: https://github.com/oren/oren.github.io/blob/master/posts/zalgo.md

This test case demonstrates the problem:

var s = kefir.fromBinder(function (emitter) {
    emitter.emit('test');
});

var value;


// Could be async, therefore MUST be async
s.onValue(function (v) {
    value = v;
});

value = 'initial';


setTimeout(function () {
    value.should.equal('test');
}, 10);

I just spent the best part of 2 hours tracking down a bug caused by this.

I was trying to do something like https://github.com/niklasvh/react-bacon-flux-poc/blob/master/lib/store.mixin.js in Kefir (see the blog post) and ran into trouble. A minor problem is that Kefir lacks combineTemplate. The major problem, though, is that Kefir's combine, unlike Bacon's, is always a stream, which means that I don't get the initial values of the properties that go into it -- which means I can't get an initial component state until one of the source streams yields an event.

What's the rationale for having combine be a stream? And any ideas on how to implement something like that state-management pattern (or something even better) in Kefir?

I am new to this stuff so take any suggestions with a grain of salt. I have noticed that skipDuplicates is implemented like:

withOneSource('skipDuplicates', {
  _init: function(args) {
    this._fn = args[0] || strictEqual;
    this._prev = NOTHING;
  },
  _free: function() {
    this._fn = null;
    this._prev = null;
  },
  _handleValue: function(x, isCurrent) {
    if (this._prev === NOTHING || !this._fn(this._prev, x)) {
      this._send(VALUE, x, isCurrent);
      this._prev = x;
    }
  }
});

This can possibly lead to cycles in the subscribers. for example i have a start and an end time and an hours field. I wanted to recalculate the hours when you enter a new end time and recalculate the end when you update the hours. I was trying to use skipDuplicates to prevent this cycle. I believe this would fix it:

withOneSource('skipDuplicates', {
  _init: function(args) {
    this._fn = args[0] || strictEqual;
    this._prev = NOTHING;
  },
  _free: function() {
    this._fn = null;
    this._prev = null;
  },
  _handleValue: function(x, isCurrent) {
    if (this._prev === NOTHING || !this._fn(this._prev, x)) {
      this._prev = x; // set value first to prevent cycle
      this._send(VALUE, x, isCurrent);
    }
  }
});

Any thoughts? BTW: thanks for the great library

For first thanks for nice library. I want to use it instead of bacon in my project. But library does not support errors handling right now.
If you do not mind I can add errors handling like bacon: Error object, onError, errors, skipErrors, retry methods for Observables.

Any thoughts?

Should I unbind jquery event or stop the stream observation on context destruction? Can be a stream created with asKefirStream function unbound manually?
... or https://github.com/pozadi/kefir/blob/master/dist/addons/kefir-jquery.js#L29 does all work?

In order to do mimic toProperty in RxJS is rather easy, just call .publishValue(1).refCount()

I understand there is some inspiration from Baconjs.

Is there an interest to implement the "fromNodeCallback" method, so nodejs projects can integrate it with nodejs style async methods?

Looking at bacon, it seems straightforward:

Bacon.fromNodeCallback = liftCallback "fromNodeCallback", (f, args...) ->
    Bacon.fromBinder (handler) ->
     makeFunction(f, args)(handler)
     nop
   , (error, value) ->
     return [new Error(error), end()] if error
     [value, end()]

Is there in Kefir an equivalente to Rx's switch? I couldn't find from the docs. The closest one was flatten, but seems to be for arrays only.