Strange behaviour with two and more pipes about rxgo HOT 6 CLOSED

Thank you for help! Yes, i guess i was trying implmenting pub-sub using reactive channels, which is probably not the right mix. Will rethink the solution to map patterns correctly.

from rxgo.

From the first look, it seems like because the goroutine which scans and chuck the text into lines channel doesn't block, by the time the program reaches the line where you create an iterable lines is still empty.

You can probably do the following:

• Print len(lines) just before creating the iterable
• If this is the case, you will have to block/wait for lines to receive something before creating an iterable from it and start processing.

Let me know if that works out.

from rxgo.

It does not seem to work. If you simply run the app and click Enter several times, it does apply only one filer per enter. So it will be Enter -> isOne, Enter -> isTwo, Enter -> isOne, Enter -> isTwo .... and so one. It does not seem to ever try checking both filters for each new value in the channel.

from rxgo.

@arkadyb that's because you're using a single source which to two separate goroutines. The data distributes "evenly" to both routines, not copied. If instead of inputting the text manually to os.Stdin you create a for loop that sends "one" and "two" alternately to lines very quickly you'll see the indeterminate behavior rather than a pattern. The source channel doesn't know if the data emitted on the source is "one" or "two" and there's no way it can send to the correct goroutine (hence "one" is sent to pipe2 at times and thus not printed, and vice versa).

To get a predictable behavior, either use a single stream with a more inclusive filter function:

isOneOrTwo := func(i interface{}) bool {
        return i == "one" || i == "two"
}
<-src.Filter(isOneOrTwo).Subscribe(process)

or send scanner.Text() to two source channels i.e. lines1 and lines2:

lines1 := make(chan interface{})
lines2 := make(chan interface{})

go func() {
        scanner := bufio.NewScanner(os.Stdin)
        for scanner.Scan() {
                lines1 <- scanner.Text()
                lines2 <- scanner.Text()
        }
}()

src1 := observable.Observable(lines1)
src2 := observable.Observable(lines2)

p1 := src1.Filter(isOne).Subscribe(process)
p2 := src2.Filter(isTwo).Subscribe(process)

select {
case <-p1:
case <-p2:
}

from rxgo.

@jochasinga, thanks for answer!

Would not be correct to send data to all the subscribed sources?

from rxgo.

Sorry I don't understand the question. Basically, a source, may it be a channel, slice, array, or an observable, can only be "streamed" down a single pipeline, or get distributed to more. You can't do something like you would do in a pub-sub pattern where a publisher emits an event, and all subscribers that subscribe to that event gets the message (which might be something you were trying to do?).

So if you start using operators on a single source to create many pipelines, you are distributing the processing task among all of them concurrently.

The closest you can do to make several pipelines processing the same data from a single source is through a Connectable:

src := connectable.From(it)
sc := src.Subscribe(process1).Subscribe(process2).Subscribe(process3).Connect()
<-sc

When Connect() is called, each Subscribe creates a new goroutine that use the subscribed process to process the source stream. It returns a <-chan (chan subscription.Subscription).

from rxgo.