flatMapLatest – add next observable before removing previous #235
Comments
|
Yeah, that makes sense. Although I'm also worry that option (b) can be too dangerous, it can introduce subtle hard to track changes in behavior in many people's programs. Seems like nothing stops us from doing option (a) though. We can then also add an argument to // supports now
()
(fn: Function)
// new
(fn: Function, overlapping: bool)
(overlapping: bool) |
|
That's what I was thinking as well. I'll make a pull request soon with a potential implementation. Thanks!
… On Feb 15, 2017, at 11:43 PM, Roman Pominov ***@***.***> wrote:
Yeah, that makes sense. Although I'm also worry that option (b) can be to dangerous, it can introduce subtle hard to track changes in behavior in many people programs.
Seems like nothing stops us from doing option (a) though. We can then also add an argument to flatMapLatest so it would support following signatures:
// supports now
()
(fn: Function)
// new
(fn: Function, overlapping: bool)
(overlapping: bool)
—
You are receiving this because you authored the thread.
Reply to this email directly, view it on GitHub, or mute the thread.
|
passing an options object of {overlapping:true} will cause the next stream to be added before any old streams are removed.
|
I just got bitten by this non-overlapping behaviour of I do think that the overlapping behaviour should be the default, because that way const combine = (lhs, rhs) =>
lhs.flatMapLatest(lhs => rhs.map(rhs => [lhs, rhs]))Usually when you can reduce the number of primitive operators you end up with simpler and more robust algebraic properties. The use case I have for the overlapping behaviour (currently) looks like this: export const apParallel = I.curry(function apParallel(f, x) {
return flatMapProperty(
f =>
hasFailed(f)
? f
: mapProperty(
x =>
hasSucceeded(f)
? hasSucceeded(x)
? of(result(f)(result(x)), mergeXHRs(0, f, x))
: hasFailed(x)
? x
: mergeXHRs(-1, f, x)
: mergeXHRs(-1, f, x),
x
),
f
)
})The idea above is to have an Applicative that performs XHRs in parallel (above, both Is there some use-case or general property that the current non-overlapping behaviour of The current behaviour seems like an odd discontinuity in an otherwise continuous function. In my case, to work around the non-overlapping behaviour, I'm introducing a hacky import * as I from 'infestines'
import * as K from 'kefir'
const TIMER = 't'
const SOURCE = 's'
const HANDLER = 'h'
const TYPE = 'type'
const VALUE = 'value'
const END = 'end'
const DelayUnsub = I.inherit(
function DelayUnsub(source) {
const self = this
K.Property.call(self)
self[SOURCE] = source
self[HANDLER] = self[TIMER] = 0
},
K.Property,
{
_onActivation() {
const self = this
if (self[TIMER]) {
clearTimeout(self[TIMER])
self[TIMER] = 0
} else {
self[SOURCE].onAny(
(self[HANDLER] = e => {
const t = e[TYPE]
if (t === VALUE) {
self._emitValue(e[VALUE])
} else if (t === END) {
self._emitEnd()
} else {
self._emitError(e[VALUE])
}
})
)
}
},
_onDeactivation() {
const self = this
self[TIMER] = setTimeout(() => {
self[SOURCE].offAny(self[HANDLER])
self[HANDLER] = this[TIMER] = 0
}, 0)
}
}
)
export const delayUnsub = source => new DelayUnsub(source)By wrapping the XHR observables with |
|
There's an open PR to implement this as an option #236. If you're interested in picking that up, I'm definitely still interested in seeing it land. |
This will cause the next stream to be added before any old streams are
removed.
```
const channelA = Kefir.stream((emitter) => {
console.log('connect a');
let count = 0, id = setInterval(() => emitter.value(count++), 250);
return () => { console.log('disconnect a'); clearInterval(id); };
});
const channelB = Kefir.stream((emitter) => {
console.log('connect b');
let count = 0, id = setInterval(() => emitter.value(count++), 250);
return () => { console.log('disconnect b'); clearInterval(id); };
});
const data = {
a: channelA,
b: Kefir.combine([channelA, channelB]),
c: channelB,
};
Kefir.sequentially(1000, ['a', 'b', 'c', undefined])
.flatMapLatest(p => p ? data[p] : Kefir.never())
.log('result');
```
With overlapping option disabled (default):
```
> connect a
> result <value> 0
> result <value> 1
> result <value> 2
> disconnect a
> connect a
> connect b
> result <value> [0, 0]
> result <value> [1, 0]
> result <value> [1, 1]
> result <value> [2, 1]
> result <value> [2, 2]
> disconnect a
> disconnect b
> connect b
> result <value> 0
> result <value> 1
> result <value> 2
> disconnect b
> result <end>
```
With overlapping option enabled:
```
> connect a
> result <value> 0
> result <value> 1
> result <value> 2
> connect b
> result <value> [3, 0]
> result <value> [3, 1]
> result <value> [4, 1]
> result <value> [4, 2]
> disconnect a
> result <value> 3
> result <value> 4
> result <value> 5
> disconnect b
> result <end>
```
Closes: kefirjs#235 kefirjs#236
This will cause the next stream to be added before any old streams are
removed.
```
const channelA = Kefir.stream((emitter) => {
console.log('connect a');
let count = 0, id = setInterval(() => emitter.value(count++), 250);
return () => { console.log('disconnect a'); clearInterval(id); };
});
const channelB = Kefir.stream((emitter) => {
console.log('connect b');
let count = 0, id = setInterval(() => emitter.value(count++), 250);
return () => { console.log('disconnect b'); clearInterval(id); };
});
const data = {
a: channelA,
b: Kefir.combine([channelA, channelB]),
c: channelB,
};
Kefir.sequentially(1000, ['a', 'b', 'c', undefined])
.flatMapLatest(p => p ? data[p] : Kefir.never())
.log('result');
```
With overlapping option disabled (default):
```
> connect a
> result <value> 0
> result <value> 1
> result <value> 2
> disconnect a
> connect a
> connect b
> result <value> [0, 0]
> result <value> [1, 0]
> result <value> [1, 1]
> result <value> [2, 1]
> result <value> [2, 2]
> disconnect a
> disconnect b
> connect b
> result <value> 0
> result <value> 1
> result <value> 2
> disconnect b
> result <end>
```
With overlapping option enabled:
```
> connect a
> result <value> 0
> result <value> 1
> result <value> 2
> connect b
> result <value> [3, 0]
> result <value> [3, 1]
> result <value> [4, 1]
> result <value> [4, 2]
> disconnect a
> result <value> 3
> result <value> 4
> result <value> 5
> disconnect b
> result <end>
```
Closes: kefirjs#235 kefirjs#236
|
I've implemented this on #320, please have a look when you have a chance. The console output is slightly different but that's correct IMO as the intervals inside each stream are different. |
In
AbstractPoolthere is an optiondropthat is used byflatMapLatest(), whendropis set tooldthen the old source is unplugged from the pool before the next source is added:https://github.com/rpominov/kefir/blob/master/src/many-sources/abstract-pool.js#L37
Is there a strong opinion against either a) adding another option, perhaps
overlapping, to softly drop the old stream (add the next stream before removing the eldest) or b) making the default behavior foroldadd the next stream before removing the old stream.Option (b) seems perhaps a little risky and a breaking change. But having control over this would be beneficial in some cases where you are switching between streams, but don't want them to deactivate it unless they are unused in the next streams activation chain.
Example:
Console:
Notice how both
aandbdisconnect and immediately reconnect on the swap. What, I think, I would like in some situations:Is there a different way to achieve this behavior without having to alter the behavior of
AbstractPool. In my real use case, theseKefir.stream()observables wrap over a socket.io connection, so it's less-than-ideal to tear down the connection between sibling projections, but at the same time, its less-than-ideal to keep the socket going for unused/unneeded projections.The text was updated successfully, but these errors were encountered: