Removing Kefir.emitter()

[rpominov]

TLDR; (For people who came here from "How do I create Emitter/Subject/Bus in Kefir?" questions)

Kefir once had Kefir.emitter() that had both Stream and Emitter interfaces. Then it was removed, because:

• We prefer to do things in declarative/functional style rather than imperative in Kefir and FRP
• Kefir.emitter() is an imperative pattern
• In 90% of cases Kefir.emitter() can be replaced with some declarative/functional style pattern (you can find some examples bellow in this thread)
• Lack of Kefir.emitter() should inspire people to find that better patterns
• In remaining 10% cases one can extract emitter from Kefir.stream(emitter => ...) (see https://github.com/rpominov/kefir/issues/88#issuecomment-92512136)

Also note that Kefir.fromBinder() mentioned often bellow is renamed to Kefir.stream() now.

If you have more questions, please continue discussion here rather that where from you have been redirected to this thread. So we have all answers in one place.

Below goes original issue body:

---

Kefir.emitter() looks like the most basic method for creating general purpose streams, but I think it actually an antipattern. Beginners tend to overuse it, but people who get FRP right, use it very rarely or even never.

I think we should remove (deprecate) Kefir.emitter(), and advertise Kefir.fromBinder() as the main method for creating general purpose streams. And we should probably rename Kefir.fromBinder() to something looking less scary and more like a main method for creating streams — Kefir.stream() or something. Also Kefir.bus() should be removed, as it mostly same as Kefir.emitter().

This proposition also aligns well with idea of minimizing API surface area https://github.com/pozadi/kefir/issues/71

---

Here is the issues one can run into when using Kefir.emitter():

1) Emitting before anyone listening:

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

// The callback obviously won't be called with `1`, but sometimes it
// not so obvious for some people.
emitter.onValue(function(x) {
  console.log(x);
});

// Right way to achieve this
var stream = Kefir.fromBinder(function(emitter) {
  emitter.emit(1);
});

// will log `1`
stream.onValue(function(x) {
  console.log(x);
});

2) Resources are not managed correctly:

// bad
function getStream() {
  var emitter = Kefir.emitter();
  // subscriber will never be removed
  target.addEventListener('click', function(event) {
    emitter.emit(event);
  });
}

// good
function getStream() {
  return Kefir.fromBinder(function(emitter) {
    target.addEventListener('click', emitter.emit);
    return function() {
      // free resources
      target.removeEventListener('click', emitter.emit);
    }
  });
}

3) A property not getting values from an emitter while not active (see http://pozadi.github.io/kefir/#active-state)

4) Methods for emitting events are exposed for all users of emitter-srteam:

var stream = getStream(); // using function from example above

// no one stops me to doing:
stream.emit(1);

And I guess there is more.

[Macil]

I agree that emitters are pretty redundant with buses and should be removed, and that Kefir.fromBinder needs to be shown first in the documentation instead of emitter, but I often find buses (including their emitter subset) important.

The plug method of buses makes them invaluable when you need a stand-in for a stream that doesn't exist yet. Here's an example of a pattern that's cropped up a couple of times in a codebase I'm working on:

var label = new Kefir.Bus();
var widget = new Widget(label.toProperty('default'));
label.plug(
  widget.events.clicks.map(...)
);

Buses/emitters are also handy when you're internally working with streams already but also need to expose a not-stream API:

function makeFoo() {
  var someStream = ...
  var emitter = new Kefir.Emitter();
  Kefir.merge([
    someStream, emitter
  ]).onValue(...);
  return {
    foo: function(x) {
      emitter.push(x);
    }
  };
}

I consider myself pretty familiar with stream libraries, but I'm pretty sure if I had to do either of those without Buses then I'd get something subtly wrong each time.

[Macil]

4) Buses could have a stream property that contains the read-only stream. The bus object itself would just have emit, plug, and stream properties. You couldn't subscribe to it directly. This would make buses work like Promise.defer() used to. (It could be viewed as an argument against my case since Promise.defer() was deprecated. I find that function useful still on occasion, but when necessary it's easy enough to translate to a new Promise call and assign the resolve/reject values to an outer scope since the callback is always synchronously called immediately. Similarly translating Kefir.emitter() calls to use Kefir.fromBinder is awkwardly more complicated and likely error-prone.)

[rooftopsparrow]

I think emitters and buses are very valuable from a a discovery and testing point of view. Its very valuable to write a simple emitter or bus for testing an idea or an interface. Its also a nice utility belt in a testing library to be able to control when things are emitted explicitly. I think it has great power and should just be "frowned upon" for general use. But sometimes the emitter or bus succinctly describes a lower level problem and I don't think they should be removed.

[rpominov]

@AgentME in your first example you can use Kefir.pool(), so no problem here.

The second one without Kefir.emitter() could look something like this.

function makeFoo() {
  var someStream = ...

  var emitter;
  var stream = Kefir.fromBinder(function(_emitter) {
    emitter = _emitter;
  });

  Kefir.merge([
    someStream, stream
  ]).onValue(...);

  return {
    foo: function(x) {
      emitter && emitter.emit(x);
    }
  };
}

A little more boilerplate, but not too much. And if you need to use such pattern often you can define your own Kefir.emitter. Also, although it looks more complicated with .fromBinder, now it more obvious what really happening under the hood, and if a developer write that code it means that he or she understand what going on. And it will be more obvious for other people who will read it.

@rooftopsparrow I agree this methods are valuable for testing ideas. But the cases when they describes a lower level problem is very rare in my experience, and in this cases one can use .fromBinder as in example above. Also if one need to use Kefir.emitter() often in app code or in unit tests, he or she can implement it in user land.

Here is one of possible implementations btw:

Kefir.emitter = function() {

  var emitter;
  var stream = Kefir.fromBinder(function(_emitter) {
    emitter = _emitter;
    return function() {
      emitter = null;
    };
  });

  stream.emit = function(x) {
    emitter && emitter.emit(x);
  };

  // TODO: other methods .error, .end, .emitEvent (or perhaps you don't need them)

  return stream;
};

---

To summarize (as I see it), we have: 1) having to write more boilerplate in rare cases (which might be not a bad thing) 2) lack of convenience methods for testing ideas VS at least 1-3 of my original points.

[rpominov]

Another case of misusing Kefir.emitter() I see pretty often looks something like this:

var stream1 = ...
var stream2 = Kefir.emitter();
stream1.onValue(function(x) {
  stream2.emit(x + 1);
});

While correct way to do this is using .map, or in more complex case .withHandler.

[rpominov]

So I think this should be done. Not having emitter will force programmers to do things in a right way from the beginning. And also will force them to understand how activation/deactivation works, which is important to use this lib properly.

[rpominov]

Going to deprecate in 2.0.0 and remove in 3.0.0

[rpominov]

Related: Bus of Doom

[rpominov]

For those who still want to use emitter and/or bus, here are complete implementations using public API:

Kefir.emitter = function() {
  var emitter;
  var stream = Kefir.stream(function(_emitter) {
    emitter = _emitter;
    return function() {
      emitter = undefined;
    }
  });

  stream.emit = function(x) {
    emitter && emitter.emit(x);
    return this;
  }

  stream.error = function(x) {
    emitter && emitter.error(x);
    return this;
  }

  stream.end = function() {
    emitter && emitter.end();
    return this;
  }

  stream.emitEvent = function(x) {
    emitter && emitter.emitEvent(x);
    return this;
  }

  return stream.setName('emitter');
}

Kefir.bus = function() {
  var pool = Kefir.pool();

  var emitter;
  var stream = Kefir.stream(function(_emitter) {
    emitter = _emitter;
    pool.onAny(_emitter.emitEvent);
    return function() {
      emitter = undefined;
      pool.offAny(_emitter.emitEvent);
    }
  });

  stream.emit = function(x) {
    emitter && emitter.emit(x);
    return this;
  }

  stream.error = function(x) {
    emitter && emitter.error(x);
    return this;
  }

  stream.end = function() {
    emitter && emitter.end();
    return this;
  }

  stream.emitEvent = function(x) {
    emitter && emitter.emitEvent(x);
    return this;
  }

  stream.plug = function(x) {
    pool.plug(x);
    return this;
  }

  stream.unplug = function(x) {
    pool.unplug(x);
    return this;
  }

  return stream.setName('bus');
}

[greglearns]

@pozadi I'm glad that you provided example code for emitter, but I hope that you either 1) reconsider removing Kefir.emitter, or 2) add the example code to the main documentation page at http://pozadi.github.io/kefir, or 3) suggest a way to solve a (very strong, I believe) use-case that I will describe.

I use Kefir on both the server and front-end. On the front-end, I've been using one Kefir.emitter for each action a user can take on the UI (click a button, change a value in a field, click a link, etc.) as a way to decouple the logic of the UI and how it works from the UI DOM-manipulating code.

example:

(in Kefir) module.export.buttonClicked = new Kefir.emitter()
(in Kefir) module.export.onButtonClicked = emitter.buttonClicked
.flatMap(...amazing stuff...)
.filter(..more amazing stuff...)
.debounce(...Kefir is awesome...)

(in the UI) ...user does something; buttonClicked.emit({ data: ... }); return
(elsewhere in the UI) onButtonClicked.onValue(function(stuff) { ... update the state of the UI })

This way all of the UI / UX flow logic is decoupled from the UI -- I even have unit tests of incredibly cool, complex UI interactions all because of Kefir.emitter. I've been converting people to use Kefir, but without Kefir.emitter, I don't see how people can easily use Kefir for the UI without Kefir.emitter (unless I wrap Kefir and add emitter back in, and call it "Postum" and tell people to use "Postum" (I am joking :-) ). I know that there are wrappers for jQuery events, and maybe there is a way to easily create a stream for each user event, but I don't think that it would be as easy as just saying emitter.emit(data).

Also, I don't see how Kefir.fromEvents would help in this case -- I saw that you used that in http://jsfiddle.net/mw18nbrr/3/.

By the way, loving Kefir. You're doing a great job with it.

[Macil]

In your case, you might find that Kefir pools work well in a more stream-oriented fashion:

(in Kefir) module.export.buttonClicked = new Kefir.pool();
...
(in the UI) buttonClicked.plug(Kefir.fromEvents(buttonNode, 'click'));

As a bonus, if everyone unsubscribes from buttonClicked, then all of the streams plugged into it can be deactivated.

Though I did put kefir-bus onto npm, which is pretty much the given example code plus some fixes to make the bus actually persist its ended state, and passes Kefir's old tests for buses.

[greglearns]

Thanks @AgentME for the suggestion, but I don't want to use (can't use) Kefir.fromEvents(buttonNode, 'click') because I'm using a UI framework that doesn't emit events that way. And using Kefir.stream(function(emitter) { emitter.emit(1); }) doesn't really work because it creates a stream that only emits one event. I'd need a way to easily keep emitting new events into that stream.

I guess I really care about this because using emitter makes developing UI's so nice. I see how there are weaknesses with giving people emitter... but I'm sad if emitter is removed, and so I hope there is some easy way to keep the benefits (and simplicity) of emitter for the UI, which really is a good approach (emit kicks off a stream with an event, which for a UI is great for interacting with user-initiated events).

[rpominov]

Hi @greglearns, thanks for your feedback, I hope we will be able to figure something out.

The posting of example implementations serve two purposes:

1. People can use it when migrating a large codebase to v2.0.0
2. Secondly, I agree that sometimes one might actually need something like emitter, and the example can show how to implement it in the user land. Although, at least in my experience, it a pretty rare case when you actually need an emitter, and there usually a better alternative.

Lets look more closely at your use case, what is the framework you are using?

[greglearns]

Hi @pozadi, thanks for looking at this issue.

I'm using http://www.ractivejs.org. You can see where they talk about their events here: http://docs.ractivejs.org/latest/events-overview. I don't see how one could easily add an Kefir.fromEvents in this case.

[rpominov]

From the example:

ractive.on( 'activate', function ( event ) {
  alert( 'Activating!' );
});

it looks that the following should work:

var activateStream = Kefir.fromEvents(ractive, 'activate');

That's because .fromEvents also supports on/off methods pair, and the target object can be basically any object that has one of supported methods pair.

[greglearns]

@pozadi :+1: -- that solved my issues.

And I think I can solve my other issue, which was related to testing complex combinations of Kefir streams that start from Kefir.fromEvent, is to replace (for the test) the Kefir.fromEvent with either a Kefir.constant or Kefir.sequentially stream.

Thanks again!

[rpominov]

Cool :+1:

If you'll have other troubles related to this, I'll be happy to look at them too.

Btw, another interesting pattern I saw people using, is to use pool instead of emitter:

// With emitter
var em = Kefir.emitter();
em.emit(1);

// With pool
var p = Kefir.pool();
p.plug(Kefir.constant(1));

Basically, if you want to expose a stream from a module, in a way so module users could push events to that stream, consider exposing a pool.

I'm not sure yet what I think about it, but it has some advantages over emitter:

1. You can "emit" events before anyone is subscribed, and events won't be lost.
2. You have more control, as you can do something like p.plug(Kefir.later(1000, 1)) to schedule a later emit. Or, of course, something like p.plug(Kefir.fromEvents(...)), but this is already looks more like a normal use of a pool.

And I did some optimizations recently, which also spread to this pattern, allowing it to work about 10x faster than before.

[greglearns]

Looks like Kefir.pool would solve the other edge case that I had, so thanks for that. It also solves the problem of if you forget to activate the stream before putting values into it :-)

[dpren]

@pozadi I've been following this issue and figured I'd jump in to see if you're willing to take a look at a related question I have on stackoverflow, since I don't know how many Kefir experts are actually out there. :neckbeard:

"How to stream events from a callback function?" http://stackoverflow.com/questions/30249010/kefir-js-how-to-stream-events-from-a-callback-function

[greglearns]

@dpren I answered your stackoverflow question and included a jsfiddle in the answer that demonstrates adding the keys.

[rpominov]

@dpren See this answer http://stackoverflow.com/a/30255824/478603 and my comment to it.

[sheerun]

It was really confusing for me until I found this thread. This should be probably documented (as well as how to solve use cases people are mentioning here)

[frankandrobot]

Great examples @rpominov, but how would you hook up React inputs without a bus? This is the current pattern we use:

<textarea value={that.state.content}
                  onChange={evt => that.contentBus.push(evt.target.value)}/>

We don't add an event listener directly to the DOM because that's an anti-pattern in React. Instead, we use a bus (we're BaconJS users but the idea is the same).

[iofjuupasli]

@frankandrobot look at #110 or https://github.com/AgentME/kefir-bus I'd recomend to use the last one. Using pool with constants is hack, that should be encapsulated. kefir-bus uses hack internally, but in your app code it will look the same as with old bus.

[aj0strow]

I want a bus / emitter for unit testing. @iofjuupasli thanks for the link!

[aksonov]

I'm trying to use reactive approach instead of classic MVC and found that most resources propose to do it via Subject for rxjs which is equivalent of emitter. So what is right way to replace MVC with kefir? If bus is anti pattern then redux, ffux, flux are also antipatterns, aren't?

[frankandrobot]

@aksonov So the consensus is "buses are usually an anti-pattern" (emphasis "usually"). The react-flux (one-way data flow) approach is probably one exception. Another way of looking at it is that you can code the Views imperatively and Model-Controller declaratively---the bus (the dispatcher) serves as the boundary between the imperative system and the declarative system. That buys you the ability to hire "normal" UX developers to polish the app while you work on the "Core".... well that's the theory at least :-)

In practice, everything is fine as long as developers stick to Flux best practices...which happens most of the time. Otherwise, when using a bus as the dispatcher, you will get hit with all the issues mentioned at the beginning of this thread. For example,

1. To prevent the listener-created-after-the-event issue, we push every bus message in the event loop queue (so it's always called at a later time). But this opens up another can of worms where mixing imperative code with dispatcher updates results in stuff running in the wrong order. On the other hand, if you follow the Flux (Redux?) practice of doing everything through the dispatcher, this won't be a problem (because it never happens).
2. To prevent the resource management issue, we do all side effects through a redux-saga inspired system. In other words, you never create "one-time" listeners.
3. We also have a Container Component class that automagically attaches/removes listeners to properties that are Observables and maps the values to React state.

This system obviously isn't perfect but it works a good chunk of the time.

Now that I've verbalized this, I wouldn't mind seeing if others have different approaches for doing one-way data flow.

UPDATE: Or you could just use Kefir.pool instead of an emitter...maybe. This would solve (1) if I'm understanding it's usage correctly. @rpominov would we get a memory leak if we don't unplug Kefir.constants from the pool?

const AppDispatcher = Kefir.pool()
AppDispatcher.plug(Kefir.constant({
  type: 'some-action',
  payload: 'some-payload'
})

[rpominov]

@rpominov would we get a memory leak if we don't unplug Kefir.constants from the pool?

No, it's safe. All observables are unplugged automatically at their completion.

Also plug(constant(...)) case is highly optimized, the observable won't be even added to the pool: https://github.com/rpominov/kefir/blob/3.2.3/src/many-sources/abstract-pool.js#L73-L81

[frankandrobot]

@rpominov pardon my french, but that is freakin' sweet!

One last question @rpominov, say we have a scenario where we already have a listener on a pool. If we plug a Kefir.constant to the pool, and then immediately attach a second listener, do we still observe the constant? i.e.,

const AppDispatcher = Kefir.pool()
AppDispatcher.onValue(//...) //first listener
AppDispatcher.plug(Kefir.constant(//...)) 
AppDispatcher.onValue(//...) //second listener: do we still observe the constant?  

[rpominov]

No, the second listener won't observe that constant in this scenario.

You can turn the pool to a property, and add listeners to property, then it should work:

const AppDispatcher = Kefir.pool()
const prop = AppDispatcher.toProperty()
prop.onValue(//...) //first listener
AppDispatcher.plug(Kefir.constant(//...))
prop.onValue(//...) //second listener: it will observe the constant 

Not sure if this helps though 😄

[frankandrobot]

@rpominov Yea I'm not sure if this helps. By using a property, you trade "missing events" to "getting the wrong event"...which is just as potentially bad and hard to debug.

Thinking about this a little more, is the real issue that FRP isn't designed for dynamically created listeners? It seems that the correct pattern should be "setup your listeners, then do stuff". At least when analyzing the first and second gotchas (at the start of this thread), violating this pattern causes the issues. The same is true even if we use a pool (as shown above).

On a related note, I just finished reading this reddit and that guy seconded my notion that Buses were invented for exactly a use case like ReactJS.

[irskep]

I came across this issue while working on a Typescript code base and thought I'd drop in a type-safe implementation for future google-users. It's just a slightly tweaked version of the JS version above.

I needed this because I'm introducing FRP into a codebase that didn't initially have it, and I don't want to have to refactor the whole thing at once.

import * as kefir from "kefir";

export default class KefirBus<T, S> {
  private pool: kefir.Pool<T, S>;

  private emitter?: kefir.Emitter<T, S>;

  stream: kefir.Stream<T, S>;

  constructor(name: string) {
    this.pool = kefir.pool<T, S>();

    this.stream = kefir.stream((_emitter) => {
      this.emitter = _emitter;
      this.pool.onAny(_emitter.emitEvent);
      return () => {
        this.emitter = undefined;
        this.pool.offAny(_emitter.emitEvent);
      };
    });

    this.stream.setName(name);
  }

  emit(x: T): KefirBus<T, S> {
    this.emitter?.emit(x);
    return this;
  }

  error(x: S): KefirBus<T, S> {
    this.emitter?.error(x);
    return this;
  }

  end(): KefirBus<T, S> {
    this.emitter?.end();
    return this;
  }

  emitEvent(x: kefir.Event<T, S>): KefirBus<T, S> {
    this.emitter?.emitEvent(x);
    return this;
  }

  plug(x: kefir.Stream<T, S>): KefirBus<T, S> {
    this.pool.plug(x);
    return this;
  }

  unplug(x: kefir.Stream<T, S>): KefirBus<T, S> {
    this.pool.unplug(x);
    return this;
  }
}