Wire event stream source into TUI trace pipeline

2 files changed, 29 insertions, 0 deletions

diff --git a/internal/ior.go b/internal/ior.go
index 55c637c..da5d6db 100644
--- a/internal/ior.go
+++ b/internal/ior.go
@@ -17,6 +17,7 @@ import (
 	"ior/internal/statsengine"
 	"ior/internal/tracepoints"
 	"ior/internal/tui"
+	"ior/internal/tui/eventstream"
 
 	bpf "github.com/aquasecurity/libbpfgo"
 )
@@ -142,7 +143,9 @@ func tuiTraceStarterFromRunTrace(
 		})
 
 		engine := statsengine.NewEngine(64)
+		streamBuf := eventstream.NewRingBuffer()
 		tui.SetDashboardSnapshotSource(engine)
+		tui.SetEventStreamSource(streamBuf)
 
 		startedCh := make(chan struct{})
 		errCh := make(chan error, 1)
@@ -151,6 +154,7 @@ func tuiTraceStarterFromRunTrace(
 			errCh <- startTrace(ctx, startedCh, func(el *eventLoop) {
 				el.printCb = func(ep *event.Pair) {
 					engine.Ingest(ep)
+					streamBuf.Push(eventstream.NewStreamEvent(ep.EnterEv.GetTime(), ep))
 					ep.Recycle()
 				}
 			})
diff --git a/internal/tui/tui.go b/internal/tui/tui.go
index 4433172..d9dd00e 100644
--- a/internal/tui/tui.go
+++ b/internal/tui/tui.go
@@ -8,6 +8,7 @@ import (
 	"ior/internal/flags"
 	"ior/internal/statsengine"
 	dashboardui "ior/internal/tui/dashboard"
+	"ior/internal/tui/eventstream"
 	tuiexport "ior/internal/tui/export"
 	"ior/internal/tui/pidpicker"
 	"os"
@@ -44,6 +45,11 @@ var dashboardSourceState struct {
 	source snapshotSource
 }
 
+var eventStreamSourceState struct {
+	mu     sync.RWMutex
+	source *eventstream.RingBuffer
+}
+
 // SetDashboardSnapshotSource sets the snapshot source used by dashboard mode.
 func SetDashboardSnapshotSource(source snapshotSource) {
 	dashboardSourceState.mu.Lock()
@@ -51,12 +57,25 @@ func SetDashboardSnapshotSource(source snapshotSource) {
 	dashboardSourceState.mu.Unlock()
 }
 
+// SetEventStreamSource sets the event stream source used by dashboard mode.
+func SetEventStreamSource(source *eventstream.RingBuffer) {
+	eventStreamSourceState.mu.Lock()
+	eventStreamSourceState.source = source
+	eventStreamSourceState.mu.Unlock()
+}
+
 func getDashboardSnapshotSource() snapshotSource {
 	dashboardSourceState.mu.RLock()
 	defer dashboardSourceState.mu.RUnlock()
 	return dashboardSourceState.source
 }
 
+func getEventStreamSource() *eventstream.RingBuffer {
+	eventStreamSourceState.mu.RLock()
+	defer eventStreamSourceState.mu.RUnlock()
+	return eventStreamSourceState.source
+}
+
 // Run starts the TUI program in alternate screen mode.
 func Run() error {
 	return RunWithTraceStarter(defaultTraceStarter)
@@ -329,6 +348,12 @@ func (lateBoundDashboardSource) Snapshot() *statsengine.Snapshot {
 	return source.Snapshot()
 }
 
+type lateBoundEventStreamSource struct{}
+
+func (lateBoundEventStreamSource) Source() *eventstream.RingBuffer {
+	return getEventStreamSource()
+}
+
 func exportSnapshotCSV(snap *statsengine.Snapshot) (string, error) {
 	filename := fmt.Sprintf("ior-snapshot-%s.csv", time.Now().Format("20060102-150405"))
 	f, err := os.Create(filename)