core/flow: clean up observers once every 3h

Author: ignoramous

intra/core/volatileflow.go

@@ -9,6 +9,7 @@ import (
 	"context"
 	"slices"
 	"sync"
+	"time"
 )
 
 type FlowFunc[T any] func(v T)
@@ -32,6 +33,15 @@ func (f FlowOn[T]) flow(v T) (flowed bool) {
 	return false
 }
 
+func (f FlowOn[T]) obsolete() bool {
+	select {
+	case <-f.ctx.Done():
+		return true
+	default:
+	}
+	return false
+}
+
 type Flow[T any] struct {
 	ctx context.Context
 
@@ -77,6 +87,16 @@ func (f *Flow[T]) stream() {
 				}
 				f.removeFinallys(notflowing)
 			}()
+		case <-time.Tick(3 * time.Hour):
+			go func() {
+				notflowing := make(map[FlowOn[T]]struct{})
+				for _, o := range f.observers() {
+					if o.obsolete() {
+						notflowing[o] = struct{}{}
+					}
+				}
+				f.removeFinallys(notflowing)
+			}()
 		}
 	}
 }