Proposal: GroupByWithTimeout

[webfrank]

When grouping by high cardinality key it would be useful to close observable if no data is coming for a configured period, something like: GroubByWithinWindow(..., d time.Duration)

After d time close the observable, and any subscriber will be completed, if the key reappears a new group observable will be created.

[samber]

You can add a ro.Timeout(duration) operator to each produced observable.

[webfrank]

Hi, I've tried with something like this:

ro.FromChannel(chanIn),
ro.DistinctBy(func(e *types.Event) string {
	return fmt.Sprintf("%s-%s-%d", e.Serial, e.Type, e.Oid)
}),
ro.Timeout[*types.Event](time.Minute),

but after timeout occurs no more events are processed, I think that raising an error should not close the source, it should be handled in the subscribe or inside error handling it should be possible to decide if error should close the source or not

[samber]

If you apply the timeout on each observable returned by GroupBy, it must work:

ro.FromChannel(chanIn),
ro.GroupBy(func(e *types.Event) string {
	return fmt.Sprintf("%s-%s-%d", e.Serial, e.Type, e.Oid)
}),
ro.Map(func(obs ro.Observable[*types.Event]) ro.Observable[*types.Event] {
        return ro.Timeout[*types.Event](time.Minute)
}),

I agree, this should be much easier to declare.