Add .flatMapEnd?

[rpominov]

We already have .flatMap & .flatMapError, adding .flatMapEnd seems like a logical continuation. It will be a more powerful alternative to .beforeEnd, which allows only to insert a value before end.

Inserting a value before end

// Before
foo.beforeEnd(() => 1)

// After
foo.flatMapEnd(() => Kefir.constant(1))

Inserting a error before end

// Before 
// (will work only if foo does't contain other values, 
// otherwise it will be even trickier)
foo.beforeEnd(() => 1).flatMap(Kefir.constantError)

// After
foo.flatMapEnd(() => Kefir.constantError(1))

Inserting multiple values

// Before 
// (Will work only if foo does't contain other values, 
// otherwise it will be even trickier)
foo.beforeEnd(() => 1).flatMap(() => Kefir.sequentially(1000, [1, 2, 3]))

// After
foo.flatMapEnd(() => Kefir.sequentially(1000, [1, 2, 3]))

Ignoring the end

// Before
foo.ignoreEnd()

// After
// (Suppose we'll have Kefir.neverEnd() 
// similar to Kefir.never() but that also doesn't end)
foo.flatMapEnd(Kefir.neverEnd)

The 1 & 4 examples aren't very convincing as current alternative is easier, but I still include them to better illustrate potential of .flatMapEnd.

[mcmathja]

Would this be substantially different from just calling .concat with the new observable?

[rpominov]

Yeah, you are right. I didn't thought about it this way. There is a minor difference that with .concat we must create new observable up front, while with .flatMapEnd we'll be able to create it later at the moment end fires. This is similar to .flatMap vs .merge. Although unlike in flatMap/merge case Kefir won't provide any new information in .flatMapEnd, so it can be useful only if we get that information from outside, e.g.:

x = ... // changes in time
stream.flatMapEnd(() => Kefir.later(x, 1))

But all examples above can be done using .concat indeed.

Need to think more about it...

[mAAdhaTTah]

I think this would be really useful for something I'm working on (I think). Here's what I'm trying to do:

I've got a stream of states (from the Redux observable). That stream of states should produce a "render stream", a stream that updates the element for a given component, providing an async abstraction for animations, for example. The idea would be that, once the render stream ends, it would flatMapEnd an event stream, which the outside could subscribe to for DOM events.

The idea would be that each state produces a render stream, which on end, produces a flatMapLatest events stream for the updated state of the DOM, so the stream consumer would always be receiving the events from the latest state of the DOM.

So it would look like this:

state$.flatMapEnd(state => {
  return Kefir.stream(emitter => {
    // do some async animation
    emitter.end();
  });
})
  .flatMapLatest(() => {
    // Provide new events stream
    return Kefir.fromEvents(document.getElementById('my-button'), 'click');
  });

I didn't see an easy way to do this with the current API, although it's possible beforeEnd could do it, assuming that would "short-circuit" the stream cleanup process.

[rpominov]

@mAAdhaTTah Hm, I don't get it. If state$ is a stream of redux states it should never end, right? If it never ends state$.flatMapEnd() will be basically a noop...

Anyway as @mcmathja mentioned flatMapEnd essentially very similar to concat, at least in your example you should be able to replace it with concat without any changes in behavior:

state$.concat(
  Kefir.stream(emitter => {
    // do some async animation
    emitter.end();
  })
)
  .flatMapLatest(() => {
    // Provide new events stream
    return Kefir.fromEvents(document.getElementById('my-button'), 'click');
  });

Seems like we might have something different in mind when thinking about how flatMapEnd would work.

[mAAdhaTTah]

state$ doesn't end, the render stream ends, and a new render stream needs to be created (and eventually terminated) every time a new state is emitted.

Perhaps I'm misunderstsanding flatMap. Maybe it needs to be more like this?

state$.flatMapLatest(state => {
    return Kefir.stream(emitter => {
        // do some async animation
        emitter.end();
    });
})
    .flatMapEnd(() => {
        // Provide new events stream
        return Kefir.fromEvents(document.getElementById('my-button'), 'click');
    });

[rpominov]

Still can't quite grasp your example.

Perhaps I'm misunderstsanding flatMap.

Maybe. Essentially flatMap works like this (arrays are streams, v() are values, e() are errors, and c are completions):

[e(0), v(1), v(2), c].flatMap(x => [v(x), e(x), c]) // [e(0), v(1), e(1), v(2), e(2), c]
[v(0), e(1), e(2), c].flatMapError(x => [v(x), e(x), c]) // [v(0), v(1), e(1), v(2), e(2), c]
[v(0), e(0), c].flatMapEnd(() => [v(1), e(2), c]) // [v(0), e(0), v(1), e(2), c]

So basically when we flatMap* we substitute v(), e(), or c with the stuff that mapping function returns. And in case of flatMapEnd we can do the same using concat:

[v(0), e(0), c].concat([v(1), e(2), c]) // [v(0), e(0), v(1), e(2), c]

[mAAdhaTTah]

Yeah, so I think this is what I'm looking for. Let me step back and try and explain the big picture; short version is I don't think concat will work because I need to keep generating new streams, rather than stuffing in one stream that I'm waiting to end. But you tell me:

The basic idea is to transform an element and a stream of states into a stream of DOM events. In the middle of all of this, the element itself needs to be rendered as a result of the new stream and the "metastream" of DOM events needs to be swapped out with a new version. My understanding is the "DOM event stream swapping" can be done with flatMap and friends.

However, the event stream shouldn't just be swapped out every time we get a new state; it needs to be swapped out only after rendering has completed. So my thought is to use flatMapEnd, which if I understand correctly, would call the callback and switch to the new stream after the current stream ends. Hence this code:

const rendering$ = state$.flatMapLatest(state => {
    return Kefir.stream(emitter => {
        // do some async animation
        emitter.end();
    });
});

would return a stream that, when the original state$ stream emits a value, swaps out the current rendering stream with a new stream that will end once the rendering is complete.

From there, we need to return a stream that, when rendering$ ends, swaps out the DOM events stream, hence:

rendering$.flatMapEnd(() => {
    // Provide new events stream
    return Kefir.fromEvents(document.getElementById('my-button'), 'click');
});

rendering$ is a "chain" of streams that will keep ending, so right as each of them end, that's when we swap out the DOM events stream.

Does that make sense? Do I have the use of flatMapEnd / flatMapLatest correct? If so, I'll definitely need this method for what I'm trying to do and will (also definitely) open a PR to add it.

Let me know what you think.

[rpominov]

short version is I don't think concat will work because I need to keep generating new streams, rather than stuffing in one stream that I'm waiting to end.

This doesn't sounds right. flatMapEnd won't have more power than concat, it also can only stuff in one stream at the end.

would return a stream that, when the original state$ stream emits a value, swaps out the current rendering stream with a new stream that will end once the rendering is complete.

In const rendering$ = state$.flatMapLatest(something) the rendering$ stream will end only after state$ end (but it never ends). So later when you do rendering$.flatMapEnd(callback), callback will never be called because rendering$ never ends.

Maybe repeat is closer to what you want? Although not sure, what you are describing also reminds a bit cycle.js. Either way doesn't seem like a use case for flatMapEnd.

[mAAdhaTTah]

In const rendering$ = state$.flatMapLatest(something) the rendering$ stream will end only after state$ end (but it never ends). So later when you do rendering$.flatMapEnd(callback), callback will never be called because rendering$ never ends.

oooooh! Because rendering$ only ends when all of its sources end. got it.

What I'm working on is inspired by Cycle.js for sure, but there are 2 things I'm doing differently:

1. The DOM rendering and the DOM events aren't so decoupled. I think that part is kinda weird, that the stream of DOM events and the stream of vdoms are separate things, whereas I feel they should be more tied together.
2. I'm using Handlebars + morphdom for rendering, since we're not using Node in production, we need a way of getting isomorphic rendering without it.

Right now, I'm trying to work out how to tie the rendering & events together into one place, in streams. I've got the idea conceptually, but I'm still struggling to figure out how to actually put it together.

Not sure repeat is the way to go, since the rendering stream still needs to be kicked off every time a new state is emitted, and the previous rendering stream needs to be cleaned up before the new one starts. I'll have to play with it and see what I can do.

Anyway, thanks for your help!

[rpominov]

@mAAdhaTTah Yeah, when you have DOM updates as side effects of your streams, and at the same time these streams depend on events from the DOM that gets updated thing get very complicated. I personally don't know of a good way to manage that complexity. But I know that cycle.js works well for some people. I for one just use React for complex UIs.

Closing this issue, because proposed flatMapEnd probably won't be added since it's basically the same as concat.

[mAAdhaTTah]

Here's how I ended up handling it:

function makeEventSwapper(events : EventsConfig, el : HTMLElement) {
    let sub : Subscription = { closed: true };
    // bindEvents returns a stream of DOM events.
    let bind = () => bindEvents(events, el) : Kefir.Observable;

    return (emitter, event) => {
        if (event.type === 'value') {
            if (!sub.closed) {
                sub.unsubscribe();
            }

            const render$ : Kefir.Observable = event.value;
            sub = render$.observe({ complete: () => emitter.next(bind()) });
        }
    };
}

render$$
    .withEventHandler(makeEventSwapper(events, el))
    .flatMapLatest();

Can you see a better way?

[rpominov]

Maybe something like this:

let bind = () => bindEvents(events, el)

render$$
  .map(render$ => render$.ignoreValues().ignoreErrors().beforeEnd(bind))
  .flatMapLatest()
  .flatMapLatest()