staltz/xstream

An extremely intuitive, small, and fast functional reactive stream library for JavaScript

Churchill Lee

@FeliciousX

Max Battcher

@WorldMaker

I've got a map(...fromArray(Stream<any>[])).flatten().compose(flattenConcurrently) and I remove a Stream from the array but its handlers never get removed and it's dispose()/cancel() is never called.

Is there some better way I'm missing for such a dynamic set of streams with xstream?

Or did I discover some edge case bug? I don't recall having an issue like this with Rx, but this particular code is brand new so I don't have a direct Rx equivalent to compare so I could be forgetful.

Max Battcher

@WorldMaker

Worse, the final Listeners down the line seem to somehow get rebound every time the top level Stream[] changes, so start to see duplicate notifications, etc.

The answer could also be my brain is mush today and I'm missing something obvious.

Max Battcher

@WorldMaker

Ah, yeah: It's probably that I want map(...fromArray().compose(flattenConcurrently)).flatten(), instead, isn't it? Another case where writing the problem down makes it more obvious.

Max Battcher

@WorldMaker

Huh, but now I need to prune already completed ones to keep them from resubscribing... Fun.

Max Battcher

@WorldMaker

Ah yes, I already have an out-of-band signal to help with that. Thanks for listening mostly inactive room.

André Staltz

@staltz

Thomas Grainger

@graingert

is there a .debug that's just for errors?

@staltz ^

André Staltz

@staltz

I think you'd have to use replaceError @graingert

Thomas Grainger

@graingert

@staltz yeah I did that and it was way better. I can put the error message in the virtual Dom

And a progress indicator for submitting to sentry!

Michal Vanko

@michalvankodev

Is it wanted behaviour for xs.combine to not emit anything when you only combine one stream? eg. xs.combine(match$).map....

Not crucial.. just wasn't expecting it to not work

Nick Johnstone

@Widdershin

@michalvankodev seems to be working for me:

 λ node                                                    
> const xs = require('xstream').default
>  xs.combine(xs.of('hi')).addListener({next: console.log})
[ 'hi' ]

(tested against the latest version of xstream)

Perhaps there's something else going on

Maybe the stream is never being subscribed to

λ • Geovani de Souza

@geovanisouza92

I have this use case, but I'm not totally sure how to achieve this: the source observable/stream will produce events sequentially, then I want to queue those events on a limited count (similar to what bufferCount does), but instead emitting arrays with those events, I want to "resolve" (think on Promise) each event and emit the result (similar to map+flatten), and this "queue operator" queue the source events, then, when one event resolves, it pick another event from the queue and continue, it will not wait for the given count to be emitted from the source, but will limit concurrent events to that count. Am I making sense? Well, I would like something similar to https://www.npmjs.com/package/throat, but for Observable/Stream. I'm thinking on a custom operator, but do you have another idea?

λ • Geovani de Souza

@geovanisouza92

well, it seems that the past @staltz already answered me :joy: https://stackoverflow.com/a/26126750/3415659

André Staltz

@staltz

Sudarsan Balaji

@artfuldev

I wanna create a diff stream with time between its events. How do I do that?

const source$ = xs.fromArray([[1,2],[1],[1,2,3,4,5]]);
const required$ =
  source$
    .startWith([])
    . // what should I do here, to get
      // -[]--[1]--[1,2]--[1]--[1,2]--[1,2,3]--[1,2,3,4]--[1,2,3,4,5]-|

The -- is constant time between the diff events

trying to modify a list of items slowly (one-by-one) even though I get all of them at a time

Michael Maier

@maiermic

how can I duplicate the first element of a stream?

Michael Maier

@maiermic

well it seems I have another issue
at least Stream.merge(elements$, elements$.take(1)) is working

this example works as expected

const xs = xstream.Stream

var stream = xs.periodic(1000).take(3)

xs.merge(stream, stream.take(1))
  .addListener({
    next: i => console.log(i),
    error: err => console.error(err),
    complete: () => console.log('completed'),
  })

Output:

0
0
1
2
"completed"

Michael Maier

@maiermic

if I add a debug to stream I get unexpected output

var stream = xs.periodic(1000).take(3).debug('test')

Output:

"test:"
0
0
0
"test:"
1
1
"test:"
2
2
"completed"

bug or feature?

Steve Lee

@SteveALee

seems good to me. Why would it be a bug? Is printing your 'test' label and then the value - all be it on separate lines.

Michael Maier

@maiermic

I expect

"test:"
0
0
"test:"
1
"test:"
2
"completed"

oh, this might be a misunderstanding of mine caused by JSBin

it's console prints

"test:"
0

instead of

test: 0

Steve Lee

@SteveALee

Yeah, that was my point. I wondered why was on separate line - so it's JSBin!

Alex

@whitecolor

@staltz is it correct that operators convert MemoryStream to Stream?

André Staltz

@staltz

@whitecolor some of them do, like filter, because filter "removes" events and unless we buffer all events from the past, we cant guarantee that the output stream after filter would remain as a memorystream

Alex

@whitecolor

I think this should be stated in docs

Michal Vanko

@michalvankodev

Hello, I'm reading documentation for flattenSequentially

I think the Marble diagram is bit misleading.

* In essence, `flattenSequentially` concatenates all nested streams.
 *
 * Marble diagram:
 *
 * ```text
 * --+--------+-------------------------
 *   \        \
 *    \       ----1----2---3--|
 *    --a--b----c----d--|
 *          flattenSequentially
 * -----a--b----c----d------1----2---3--
 *

```

Shouldn't it be

 * --+--------+-------------------------
 *   \        \
 *    \       ----1----2---3--|
 *    --a--b----c----d--|
 *          flattenSequentially
 * -----a--b----c----d--1-2-3-----

Basically the events which were buffered shouldn't they be emmited right away?

Alex

@whitecolor

why? just after first streams completes it will subscribe to the second and emit the values

Michal Vanko

@michalvankodev

But those buffered events shouldn't be postponed by the time it took them to be emmited in the original stream

Alex

@whitecolor

it will not buffer, it will just subscribe, the example assumes that second stream has no subscribers yet

Where communities thrive

People

Repo info

Activity