Add ability to pass in a mergewith to opset

lytics · Oct 16, 2024 · 03fb736 · 03fb736
1 parent d175fff
commit 03fb736
Show file tree

Hide file tree

Showing 3 changed files with 7 additions and 4 deletions.
diff --git a/opqueue.go b/opqueue.go
@@ -49,7 +49,7 @@ func NewOpQueue(depth, width int) *OpQueue {
 		entries: map[ID]*OpSet{},
 
 		reducer: func(opset *OpSet, op *Op) {
-			opset.append(op)
+			opset.MergeWith(append, op)
 		},
 	}
 	q.cond.L = &q.mu

diff --git a/opset.go b/opset.go
@@ -12,8 +12,11 @@ func newOpSet(op *Op) *OpSet {
 	}
 }
 
-func (os *OpSet) append(op *Op) {
-	os.set = append(os.set, op)
+func appendOp(ops []*Op, op *Op) []*Op {
+	return append(ops, op)
+}
+func (os *OpSet) MergeWith(fn func([]*Op, *Op) []*Op, op *Op) {
+	os.set = fn(os.set, op)
 }
 
 // Ops get the list of ops in this set.

diff --git a/opwindow.go b/opwindow.go
@@ -46,7 +46,7 @@ func NewOpWindow(depth, width int, windowedBy time.Duration) *OpWindow {
 		windowedBy:    windowedBy,
 		m:             make(map[ID]*queueItem),
 		reducer: func(opset *OpSet, op *Op) {
-			opset.append(op)
+			opset.MergeWith(appendOp, op)
 		},
 	}
 	q.q.Init()