Resume from error

[webfrank]

Hi,
I have pipeline where I want to handle missing events. I'm using ro.Timeout() and if events are not coming within the specified interval I have an error I can handle but the pipeline stops, I would like to resume normal operations until next timeout occurs.

[webfrank]

This is the sample code I have

observable := ro.Pipe[*proto.Event, ro.Observable[*proto.Event]](
	ro.FromChannel(p.channelIn),
	ro.GroupBy(func(e *proto.Event) string {
		return e.Serial
	}),
)

p.sub = observable.SubscribeWithContext(p.Ctx, ro.OnNext(func(source ro.Observable[*proto.Event]) {
	var last *proto.Event
	ro.Pipe[*proto.Event, *proto.Event](
		source,
		ro.Timeout[*proto.Event](duration),
		ro.Tap(
			func(e *proto.Event) {
				last = e
			},
			func(err error) {
				if strings.HasPrefix(err.Error(), "ro.Timeout") {
					p.Emit(last)
				}
			},
			func() {},
		),
		ro.Retry[*proto.Event](),
	).SubscribeWithContext(p.Ctx, ro.NoopObserver[*proto.Event]())
}))

As you can see I've a Timeout but instead of handling errors in subscribe I had to use Tap and Retry to restart the pipeline.