Retry Observable FromChannel

[aibolorazbekov]

Hi All.

I'm trying to call Retry for Observable that returned FromChannel.
I want make retry 10 times, but it did it only one time.

I tried an example from documentation (Observable from Just operation) - making retry 10 times - and it did it 10 times.

Here is the code that i'm working:

pipeline := asObservable(streamCh)
<-pipeline.
	Map(func(c context.Context, i interface{}) (interface{}, error) {
		return nil, errors.New("foo")
	}). // here i want to make retry 10 times if error foo returned 
	Retry(10, func(e error) bool {
		return e.Error() == "foo"
	})
	Run(rxgo.WithErrorStrategy(errorStrategy))

func asObservable(streamCh chan rxgo.Item) rxgo.Observable {
	go func() {
		for {
			select {
			case msg := <-o.msgs:
				streamCh <- rxgo.Of(msg)
			}
		}
	}()

	return rxgo.FromChannel(o.streamCh)
}

So, it seems like, i there is something that i don't understand mechanics of Observable that returned FromChannel...

Can someone please, help me to solve the issue?