A question about signal processing in the Connectable.Connect method. about rxgo HOT 6 CLOSED

Can you paste the permalink to the line in the source code you are referring to? Also, can you try the following:

sub := co.Connect()
if v, ok := <-sub; ok {
        fmt.Println("v --->", v)
}
assert.Equal(t, 6, num)

from rxgo.

This is the permalink to the first piece of code:

And I had tried to do it:

func TestDoOperator(t *testing.T) {
	co := Just(1, 2, 3)
	num := 0

	nextf := func(item interface{}) {
		num += item.(int)
	}

	co = co.Do(nextf)
	sub := co.Connect()
	if v, ok := <-sub; ok {
		fmt.Println("v --->", v)
	}
	fmt.Println("no vvvvvvvvv")
	<-sub

	assert.Equal(t, 6, num)
}

The test results are like this:

❯ go test
no vvvvvvvvv
PASS

I'm still confused about it:

go func() {
	temp <- sub // Because no one reads sub, so it will be blocked here
	fmt.Println("test") // The code can not be executed here.
	done <- temp // So temp will not be sent to done, it is impossible for others to read the temp from done.
}()

I guess done <- temp can not be executed.
Thanks!

from rxgo.

This code:

if v, ok := <-sub; ok {
        fmt.Println("v --->", v)
}

does not block/wait. It reads from sub and prints if it's immediately available, which it never is at that point.

In order to make sure the goroutine in the pipeline finishes before doing anything else, you must block-wait for it.

// ...
sc := co.Connect()
for subs := range sc {
        for sub := range subs {
                fmt.Println("v ---->", sub)
        }
}

Also note that co.Connect() returns a <-chan (chan subscription.Subscription), because a connectable emits a resulting Subscription object for every goroutine it creates for each observer it subscribes to, hence the nested loops.

from rxgo.

Sorry, I still don't understand. Maybe my question is not clear.

I try this code, but it doesn't work.

sc := co.Connect()
for sub := range <-sc {
        if s, ok := <-sub; ok {
                fmt.Printf("%v is done", s)
        }
}

This is the result:

./connectable_test.go:66:15: invalid operation: <-sub (receive from non-chan type subscription.Subscription)
FAIL    _/Users/LLLeon/Golang/RxGo/connectable [build failed]

I think it's because sub is nill. Just like the last test result.

v ---> <nil>

And what puzzles me is this piece of code:

go func() {
	temp <- sub // There is no buffer in temp, so sub can only be put in when someone receives from temp, it'll blocked here.
	done <- temp // This line of code will not be executed.
}()

Is my explanation correct?
@jochasinga

from rxgo.

I have the same issue and i think you are right.

go func() {
	temp <- sub
	fmt.Println("hahaha")
	done <- temp
}()

the word can never be printed.

and the test did not succeed.

connectable_test.go:228
for sub := range subs {
		for s := range sub {
                        // actually can't run
			assert.Equal("bang", s.Error.Error())
		}
	}

from rxgo.

Hey! I'm cleaning the backlog by removing the issues without updates for more than a year. Meanwhile, please check at the v2 to see if the issue is still relevant.

from rxgo.