diff --git a/internal/graphql/services/executors.go b/internal/graphql/services/executors.go index e2254cddf2..a515b42851 100644 --- a/internal/graphql/services/executors.go +++ b/internal/graphql/services/executors.go @@ -32,7 +32,7 @@ func (s *executorsService) List(selector string) ([]testkube.ExecutorDetails, er } func (s *executorsService) SubscribeList(ctx context.Context, selector string) (<-chan []testkube.ExecutorDetails, error) { - return HandleSubscription(ctx, "events.executor.>", s, func() ([]testkube.ExecutorDetails, error) { + return HandleSubscription(ctx, "agentevents.executor.>", s, func() ([]testkube.ExecutorDetails, error) { return s.List(selector) }) } diff --git a/internal/graphql/services/executors_test.go b/internal/graphql/services/executors_test.go index aae22813a4..df45d7f7ea 100644 --- a/internal/graphql/services/executors_test.go +++ b/internal/graphql/services/executors_test.go @@ -169,7 +169,7 @@ func TestExecutorsService_SubscribeList(t *testing.T) { assert.NoError(t, err) <-ch client.Client = getMockExecutorClient(k8sObjects2).Client - assert.NoError(t, srvMock.BusMock().PublishTopic("events.executor.create", testkube.Event{ + assert.NoError(t, srvMock.BusMock().PublishTopic("agentevents.executor.create", testkube.Event{ Type_: testkube.EventCreated, Resource: testkube.EventResourceExecutor, })) diff --git a/pkg/api/v1/testkube/model_event_extended.go b/pkg/api/v1/testkube/model_event_extended.go index e9415a386e..af7a0d2e3c 100644 --- a/pkg/api/v1/testkube/model_event_extended.go +++ b/pkg/api/v1/testkube/model_event_extended.go @@ -8,8 +8,8 @@ import ( ) const ( - TestStartSubject = "events.test.start" - TestStopSubject = "events.test.stop" + TestStartSubject = "agentevents.test.start" + TestStopSubject = "agentevents.test.stop" ) // check if Event implements model generic event type @@ -270,14 +270,14 @@ func (e Event) Topic() string { } if e.Resource == nil { - return "events.all" + return "agentevents.all" } if e.ResourceId == "" { - return "events." + string(*e.Resource) + return "agentevents." + string(*e.Resource) } - return "events." + string(*e.Resource) + "." + e.ResourceId + return "agentevents." + string(*e.Resource) + "." + e.ResourceId } // GetResourceId implmenents generic event trigger diff --git a/pkg/api/v1/testkube/model_event_extended_test.go b/pkg/api/v1/testkube/model_event_extended_test.go index cf0da3bb5c..bb6845a63b 100644 --- a/pkg/api/v1/testkube/model_event_extended_test.go +++ b/pkg/api/v1/testkube/model_event_extended_test.go @@ -339,20 +339,20 @@ func TestEvent_Topic(t *testing.T) { t.Parallel() evt := Event{Type_: EventStartTest, Resource: nil} - assert.Equal(t, "events.all", evt.Topic()) + assert.Equal(t, "agentevents.all", evt.Topic()) }) t.Run("should return event topic with resource name and id if set", func(t *testing.T) { t.Parallel() evt := Event{Type_: EventStartTest, Resource: EventResourceExecutor, ResourceId: "a12"} - assert.Equal(t, "events.executor.a12", evt.Topic()) + assert.Equal(t, "agentevents.executor.a12", evt.Topic()) }) t.Run("should return event topic with resource name when id not set", func(t *testing.T) { t.Parallel() evt := Event{Type_: EventStartTest, Resource: EventResourceExecutor} - assert.Equal(t, "events.executor", evt.Topic()) + assert.Equal(t, "agentevents.executor", evt.Topic()) }) } diff --git a/pkg/event/bus/nats.go b/pkg/event/bus/nats.go index f652882cbb..e10fc563e2 100644 --- a/pkg/event/bus/nats.go +++ b/pkg/event/bus/nats.go @@ -23,7 +23,7 @@ var ( const ( SubscribeBuffer = 1 - SubscriptionName = "events" + SubscriptionName = "agentevents" InternalPublishTopic = "internal.all" InternalSubscribeTopic = "internal.>" ) diff --git a/pkg/event/emitter.go b/pkg/event/emitter.go index b814b0f22a..232537c983 100644 --- a/pkg/event/emitter.go +++ b/pkg/event/emitter.go @@ -155,7 +155,7 @@ func (e *Emitter) Listen(ctx context.Context) { } func (e *Emitter) startListener(l common.Listener) { - err := e.Bus.SubscribeTopic("events.>", l.Name(), e.notifyHandler(l)) + err := e.Bus.SubscribeTopic("agentevents.>", l.Name(), e.notifyHandler(l)) if err != nil { e.Log.Errorw("error while starting listener", "error", err) } diff --git a/pkg/logs/client/interface.go b/pkg/logs/client/interface.go index d942682f6c..d07c902df9 100644 --- a/pkg/logs/client/interface.go +++ b/pkg/logs/client/interface.go @@ -9,8 +9,8 @@ import ( const ( StreamPrefix = "log" - StartSubject = "events.logs.start" - StopSubject = "events.logs.stop" + StartSubject = "agentevents.logs.start" + StopSubject = "agentevents.logs.stop" ) //go:generate mockgen -destination=./mock_stream.go -package=client "github.com/kubeshop/testkube/pkg/logs/client" Stream diff --git a/pkg/logs/events.go b/pkg/logs/events.go index 712a5df1a5..7a2bfb3d56 100644 --- a/pkg/logs/events.go +++ b/pkg/logs/events.go @@ -27,8 +27,8 @@ const ( StartQueue = "logsstart" StopQueue = "logsstop" - LogStartSubject = "events.logs.start" - LogStopSubject = "events.logs.stop" + LogStartSubject = "agentevents.logs.start" + LogStopSubject = "agentevents.logs.stop" ) var (