feat!: add an additional level in event publish topic (#1313)

BREAKING CHNAGE: event publish topic will be <prefix>/<service-name>/<profile-name>/<device-name>/<source-name>

closes #1261

Signed-off-by: Chris Hung <chris@iotechsys.com>

example/cmd/device-simple/res/configuration.toml

@@ -66,7 +66,7 @@ Type = "redis"
 AuthMode = "usernamepassword"  # required for redis messagebus (secure or insecure).
 SecretName = "redisdb"
   [MessageBus.Topics]
-  PublishTopicPrefix = "edgex/events/device" # /<device-profile-name>/<device-name>/<source-name> will be added to this Publish Topic prefix
+  PublishTopicPrefix = "edgex/events/device" # /<device-service-name>/<device-profile-name>/<device-name>/<source-name> will be added to this Publish Topic prefix
   DeviceCommandRequestTopic = "edgex/device/command/request/device-simple/#"   # subscribing for inbound command requests
   ResponseTopicPrefix = "edgex/response" # /<service-name>/<request-id> will be added to this prefix topic
   MetadataSystemEventTopic = "edgex/system-events/core-metadata/+/+/device-simple/#"

internal/common/utils.go

@@ -87,7 +87,8 @@ func SendEvent(event *dtos.Event, correlationID string, dic *di.Container) {
 	ctx = context.WithValue(ctx, common.ContentType, encoding) // nolint: staticcheck
 	envelope := types.NewMessageEnvelope(bytes, ctx)
 	prefix := configuration.MessageBus.Topics[config.MessageBusPublishTopicPrefix]
-	publishTopic := fmt.Sprintf("%s/%s/%s/%s", prefix, event.ProfileName, event.DeviceName, event.SourceName)
+	serviceName := container.DeviceServiceFrom(dic.Get).Name
+	publishTopic := fmt.Sprintf("%s/%s/%s/%s/%s", prefix, serviceName, event.ProfileName, event.DeviceName, event.SourceName)
 	err = mc.Publish(envelope, publishTopic)
 	if err != nil {
 		lc.Errorf("Failed to publish event to MessageBus: %s", err)

internal/common/utils_test.go

@@ -20,6 +20,7 @@ import (
 	"github.com/edgexfoundry/go-mod-core-contracts/v3/dtos"
 	"github.com/edgexfoundry/go-mod-core-contracts/v3/dtos/requests"
 	"github.com/edgexfoundry/go-mod-core-contracts/v3/dtos/responses"
+	"github.com/edgexfoundry/go-mod-core-contracts/v3/models"
 	msgMocks "github.com/edgexfoundry/go-mod-messaging/v3/messaging/mocks"
 
 	"github.com/edgexfoundry/device-sdk-go/v3/internal/cache"
@@ -66,6 +67,7 @@ func NewMockDIC() *di.Container {
 	configuration := &config.ConfigurationStruct{
 		Device: config.DeviceInfo{MaxCmdOps: 1},
 	}
+	deviceService := &models.DeviceService{Name: TestDeviceService}
 
 	devices := responses.MultiDevicesResponse{
 		Devices: []dtos.Device{
@@ -121,12 +123,15 @@ func NewMockDIC() *di.Container {
 	pwcMock.On("ProvisionWatchersByServiceName", context.Background(), TestDeviceService, 0, -1).Return(responses.MultiProvisionWatchersResponse{}, nil)
 
 	return di.NewContainer(di.ServiceConstructorMap{
-		bootstrapContainer.LoggingClientInterfaceName: func(get di.Get) interface{} {
-			return logger.NewMockClient()
-		},
 		container.ConfigurationName: func(get di.Get) interface{} {
 			return configuration
 		},
+		container.DeviceServiceName: func(get di.Get) any {
+			return deviceService
+		},
+		bootstrapContainer.LoggingClientInterfaceName: func(get di.Get) interface{} {
+			return logger.NewMockClient()
+		},
 		bootstrapContainer.DeviceClientName: func(get di.Get) interface{} {
 			return dcMock
 		},