feat(01-03): implement ticker-driven window aggregator with TDD

- Add Aggregate function reading ClassifiedPackets, emitting WindowSnapshots per time window
- Add DefaultWindowMs=500 constant for configurable window duration
- Flush partial window on done channel close or closed events channel
- Call optional onSnapshot callback per window for verbose mode
- Add 5 tests: EmitsSnapshot, MultipleClasses, DoneFlushesPartial, EmptyWindow, WindowIndex

aggregate/window.go

@@ -0,0 +1,63 @@
+package aggregate
+
+import (
+	"time"
+
+	"github.com/netsynth/netsynth/classify"
+)
+
+// DefaultWindowMs is the default time window duration in milliseconds.
+const DefaultWindowMs = 500
+
+// Aggregate reads ClassifiedPackets from events, accumulates counts per TrafficClass
+// in time windows of windowMs duration, and emits a WindowSnapshot per window.
+// When done is closed, flushes the current partial window and closes the output channel.
+// If onSnapshot is non-nil, it is called with each snapshot before sending to the output channel
+// (used for --verbose stderr output).
+func Aggregate(done <-chan struct{}, events <-chan classify.ClassifiedPacket,
+	windowMs int, onSnapshot func(classify.WindowSnapshot)) <-chan classify.WindowSnapshot {
+
+	out := make(chan classify.WindowSnapshot, 8)
+	go func() {
+		defer close(out)
+		ticker := time.NewTicker(time.Duration(windowMs) * time.Millisecond)
+		defer ticker.Stop()
+
+		counts := make(map[classify.TrafficClass]int64)
+		var totalPackets int64
+		windowIndex := 0
+
+		flush := func() {
+			snap := classify.WindowSnapshot{
+				Counts:       counts,
+				TotalPackets: totalPackets,
+				WindowIndex:  windowIndex,
+			}
+			if onSnapshot != nil {
+				onSnapshot(snap)
+			}
+			out <- snap
+			counts = make(map[classify.TrafficClass]int64)
+			totalPackets = 0
+			windowIndex++
+		}
+
+		for {
+			select {
+			case <-done:
+				flush() // flush final partial window
+				return
+			case <-ticker.C:
+				flush()
+			case ev, ok := <-events:
+				if !ok {
+					flush()
+					return
+				}
+				counts[ev.Class]++
+				totalPackets++
+			}
+		}
+	}()
+	return out
+}

aggregate/window_test.go

@@ -0,0 +1,158 @@
+package aggregate
+
+import (
+	"testing"
+	"time"
+
+	"github.com/netsynth/netsynth/classify"
+)
+
+// sendPackets sends a slice of ClassifiedPackets to the events channel.
+func sendPackets(events chan<- classify.ClassifiedPacket, packets []classify.ClassifiedPacket) {
+	for _, p := range packets {
+		events <- p
+	}
+}
+
+func TestAggregateEmitsSnapshot(t *testing.T) {
+	done := make(chan struct{})
+	events := make(chan classify.ClassifiedPacket, 20)
+	defer close(done)
+
+	// Use short window to trigger ticks quickly.
+	out := Aggregate(done, events, 20, nil)
+
+	// Send 5 ICMP packets.
+	for i := 0; i < 5; i++ {
+		events <- classify.ClassifiedPacket{Class: classify.ClassICMP}
+	}
+
+	// Wait for at least one snapshot from a tick.
+	select {
+	case snap := <-out:
+		if snap.TotalPackets < 5 {
+			// Packets may span windows; total across first snapshot should be <= 5.
+			// Just verify snapshot is emitted and counts are non-negative.
+		}
+		_ = snap
+	case <-time.After(500 * time.Millisecond):
+		t.Fatal("timeout: no snapshot emitted within 500ms")
+	}
+}
+
+func TestAggregateMultipleClasses(t *testing.T) {
+	done := make(chan struct{})
+	events := make(chan classify.ClassifiedPacket, 20)
+
+	out := Aggregate(done, events, 20, nil)
+
+	events <- classify.ClassifiedPacket{Class: classify.ClassICMP}
+	events <- classify.ClassifiedPacket{Class: classify.ClassICMP}
+	events <- classify.ClassifiedPacket{Class: classify.ClassDNS}
+	events <- classify.ClassifiedPacket{Class: classify.ClassHTTPS}
+	events <- classify.ClassifiedPacket{Class: classify.ClassHTTPS}
+	events <- classify.ClassifiedPacket{Class: classify.ClassHTTPS}
+
+	// Close done to get the final flush.
+	// But first drain any tick-based snapshots.
+	time.Sleep(10 * time.Millisecond)
+	close(done)
+
+	// Accumulate all snapshots to verify totals.
+	totals := make(map[classify.TrafficClass]int64)
+	for snap := range out {
+		for class, count := range snap.Counts {
+			totals[class] += count
+		}
+	}
+
+	if totals[classify.ClassICMP] != 2 {
+		t.Errorf("expected ICMP=2, got %d", totals[classify.ClassICMP])
+	}
+	if totals[classify.ClassDNS] != 1 {
+		t.Errorf("expected DNS=1, got %d", totals[classify.ClassDNS])
+	}
+	if totals[classify.ClassHTTPS] != 3 {
+		t.Errorf("expected HTTPS=3, got %d", totals[classify.ClassHTTPS])
+	}
+}
+
+func TestAggregateDoneFlushesPartial(t *testing.T) {
+	done := make(chan struct{})
+	// Use very large window so no tick fires.
+	events := make(chan classify.ClassifiedPacket, 20)
+
+	out := Aggregate(done, events, 100_000, nil)
+
+	events <- classify.ClassifiedPacket{Class: classify.ClassSSH}
+	events <- classify.ClassifiedPacket{Class: classify.ClassSSH}
+
+	// Give goroutine time to process the two events.
+	time.Sleep(20 * time.Millisecond)
+
+	// Close done to trigger flush.
+	close(done)
+
+	var snap classify.WindowSnapshot
+	select {
+	case snap = <-out:
+	case <-time.After(500 * time.Millisecond):
+		t.Fatal("timeout: no snapshot on done-flush")
+	}
+
+	if snap.Counts[classify.ClassSSH] != 2 {
+		t.Errorf("expected SSH=2 in flush, got %d", snap.Counts[classify.ClassSSH])
+	}
+	if snap.TotalPackets != 2 {
+		t.Errorf("expected TotalPackets=2, got %d", snap.TotalPackets)
+	}
+}
+
+func TestAggregateEmptyWindow(t *testing.T) {
+	done := make(chan struct{})
+	events := make(chan classify.ClassifiedPacket, 20)
+
+	out := Aggregate(done, events, 20, nil)
+
+	// Don't send any packets, just wait for a tick.
+	select {
+	case snap := <-out:
+		if snap.TotalPackets != 0 {
+			t.Errorf("expected TotalPackets=0 for empty window, got %d", snap.TotalPackets)
+		}
+		if len(snap.Counts) != 0 {
+			t.Errorf("expected empty Counts for empty window, got %v", snap.Counts)
+		}
+	case <-time.After(500 * time.Millisecond):
+		t.Fatal("timeout: no snapshot for empty window")
+	}
+
+	close(done)
+}
+
+func TestAggregateWindowIndex(t *testing.T) {
+	done := make(chan struct{})
+	events := make(chan classify.ClassifiedPacket, 5)
+
+	out := Aggregate(done, events, 20, nil)
+
+	// Collect 3 snapshots.
+	var indices []int
+	timeout := time.After(2 * time.Second)
+	for len(indices) < 3 {
+		select {
+		case snap := <-out:
+			indices = append(indices, snap.WindowIndex)
+		case <-timeout:
+			t.Fatalf("timeout: only collected %d snapshots", len(indices))
+		}
+	}
+	close(done)
+
+	// WindowIndex should increment: 0, 1, 2.
+	for i, idx := range indices {
+		if idx != i {
+			t.Errorf("expected WindowIndex[%d]=%d, got %d", i, i, idx)
+		}
+	}
+}