iop/apps/edge/internal/events/bus_test.go

package events_test

import (
	"testing"

	edgeevents "iop/apps/edge/internal/events"
	iop "iop/proto/gen/iop"
)

func TestBusPublishesRunToSpecificAndAllSubscribers(t *testing.T) {
	bus := edgeevents.NewBus()
	specific, unsubscribeSpecific := bus.SubscribeRun("run-1", 1)
	defer unsubscribeSpecific()
	all, unsubscribeAll := bus.SubscribeAllRuns(1)
	defer unsubscribeAll()

	bus.PublishRun(&iop.RunEvent{RunId: "run-1", Type: "start"})

	select {
	case event := <-specific:
		if event.GetRunId() != "run-1" {
			t.Fatalf("specific run id: %q", event.GetRunId())
		}
	default:
		t.Fatal("expected specific run subscriber event")
	}
	select {
	case event := <-all:
		if event.GetRunId() != "run-1" {
			t.Fatalf("all run id: %q", event.GetRunId())
		}
	default:
		t.Fatal("expected all run subscriber event")
	}
}

func TestBusCountsDroppedRunEvents(t *testing.T) {
	bus := edgeevents.NewBus()
	_, unsubscribe := bus.SubscribeRun("run-1", 1)
	defer unsubscribe()

	// First event fills the buffer; subsequent ones are dropped until drained.
	bus.PublishRun(&iop.RunEvent{RunId: "run-1", Type: "start"})
	bus.PublishRun(&iop.RunEvent{RunId: "run-1", Type: "chunk-1"})
	bus.PublishRun(&iop.RunEvent{RunId: "run-1", Type: "chunk-2"})

	stats := bus.Stats()
	if stats.DroppedRunEvents != 2 {
		t.Fatalf("DroppedRunEvents: got %d want 2", stats.DroppedRunEvents)
	}
	if stats.DroppedNodeEvents != 0 {
		t.Fatalf("DroppedNodeEvents: got %d want 0", stats.DroppedNodeEvents)
	}
}

func TestBusCountsDroppedNodeEvents(t *testing.T) {
	bus := edgeevents.NewBus()
	_, unsubscribe := bus.SubscribeNode("node-1", 1)
	defer unsubscribe()

	bus.PublishNode(&iop.EdgeNodeEvent{NodeId: "node-1", Type: "node.connected"})
	bus.PublishNode(&iop.EdgeNodeEvent{NodeId: "node-1", Type: "node.disconnected"})
	bus.PublishNode(&iop.EdgeNodeEvent{NodeId: "node-1", Type: "node.reconnected"})

	stats := bus.Stats()
	if stats.DroppedNodeEvents != 2 {
		t.Fatalf("DroppedNodeEvents: got %d want 2", stats.DroppedNodeEvents)
	}
	if stats.DroppedRunEvents != 0 {
		t.Fatalf("DroppedRunEvents: got %d want 0", stats.DroppedRunEvents)
	}
}

func TestBusReplaysRecentRunEvents(t *testing.T) {
	bus := edgeevents.NewBus()

	bus.PublishRun(&iop.RunEvent{RunId: "run-1", Type: "start"})
	bus.PublishRun(&iop.RunEvent{RunId: "run-2", Type: "start"})
	bus.PublishRun(&iop.RunEvent{RunId: "run-1", Type: "complete"})

	replay := bus.ReplayRun("run-1")
	if len(replay) != 2 {
		t.Fatalf("ReplayRun len: got %d want 2", len(replay))
	}
	if replay[0].GetType() != "start" || replay[1].GetType() != "complete" {
		t.Fatalf("replay order: got [%q,%q] want [start,complete]", replay[0].GetType(), replay[1].GetType())
	}

	if got := bus.ReplayRun("missing"); got != nil {
		t.Fatalf("expected nil replay for unknown run, got %v", got)
	}
}

func TestBusReplayIsBounded(t *testing.T) {
	const cap = 3
	bus := edgeevents.NewBusWithReplayCapacity(cap)

	for i := 0; i < cap+5; i++ {
		bus.PublishRun(&iop.RunEvent{RunId: "run-1", Type: "tick"})
		bus.PublishNode(&iop.EdgeNodeEvent{NodeId: "node-1", Type: "tick"})
	}

	if got := bus.ReplayRun("run-1"); len(got) != cap {
		t.Fatalf("ReplayRun len: got %d want %d", len(got), cap)
	}
	if got := bus.ReplayNode("node-1"); len(got) != cap {
		t.Fatalf("ReplayNode len: got %d want %d", len(got), cap)
	}
}

func TestBusPublishesNodeToSpecificAndAllSubscribers(t *testing.T) {
	bus := edgeevents.NewBus()
	specific, unsubscribeSpecific := bus.SubscribeNode("node-1", 1)
	defer unsubscribeSpecific()
	all, unsubscribeAll := bus.SubscribeAllNodes(1)
	defer unsubscribeAll()

	bus.PublishNode(&iop.EdgeNodeEvent{NodeId: "node-1", Type: "node.disconnected"})

	select {
	case event := <-specific:
		if event.GetNodeId() != "node-1" {
			t.Fatalf("specific node id: %q", event.GetNodeId())
		}
	default:
		t.Fatal("expected specific node subscriber event")
	}
	select {
	case event := <-all:
		if event.GetNodeId() != "node-1" {
			t.Fatalf("all node id: %q", event.GetNodeId())
		}
	default:
		t.Fatal("expected all node subscriber event")
	}
}