Comments (6)
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.
Related Issues (20)
- when to support the generic type HOT 1
- Why I can not find "Throttle" operator?
- [Support] Leak goroutine with rxgo.Merge HOT 1
- SwitchMap operator
- New maintainer? HOT 11
- Sometimes First().Get() not returning a context canceled error HOT 2
- Go 1.18 Generics Support
- RX driver for SQL databases
- Counting items of a Connectable Observable
- Cold & Hot Observables HOT 1
- Add `Subject` to rxgo
- There are some examples.
- Custom Producer HOT 1
- How cancel subscription ?
- How to do recursion?
- Is this still mantained? HOT 2
- RxGo Rest api example HOT 1
- mapOperator missing res when return error in "map" function HOT 1
- Adding Subjects
- Real-World Example in doc HOT 1
Recommend Projects
-
React
A declarative, efficient, and flexible JavaScript library for building user interfaces.
-
Vue.js
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
-
Typescript
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
-
TensorFlow
An Open Source Machine Learning Framework for Everyone
-
Django
The Web framework for perfectionists with deadlines.
-
Laravel
A PHP framework for web artisans
-
D3
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
-
Recommend Topics
-
javascript
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
-
web
Some thing interesting about web. New door for the world.
-
server
A server is a program made to process requests and deliver data to clients.
-
Machine learning
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
-
Visualization
Some thing interesting about visualization, use data art
-
Game
Some thing interesting about game, make everyone happy.
Recommend Org
-
Facebook
We are working to build community through open source technology. NB: members must have two-factor auth.
-
Microsoft
Open source projects and samples from Microsoft.
-
Google
Google ❤️ Open Source for everyone.
-
Alibaba
Alibaba Open Source for everyone
-
D3
Data-Driven Documents codes.
-
Tencent
China tencent open source team.
from rxgo.