fix: Add missing contentType to new Publish APIs

Also added missing UUID and refactored out duplicate code

Signed-off-by: Leonard Goodell <leonard.goodell@intel.com>

internal/app/service.go

@@ -33,6 +33,7 @@ import (
 	bootstrapHandlers "github.com/edgexfoundry/go-mod-bootstrap/v3/bootstrap/handlers"
 	"github.com/edgexfoundry/go-mod-core-contracts/v3/dtos"
 	"github.com/edgexfoundry/go-mod-messaging/v3/pkg/types"
+	"github.com/google/uuid"
 
 	"github.com/edgexfoundry/app-functions-sdk-go/v3/internal"
 	"github.com/edgexfoundry/app-functions-sdk-go/v3/internal/appfunction"
@@ -43,8 +44,6 @@ import (
 	"github.com/edgexfoundry/app-functions-sdk-go/v3/internal/webserver"
 	"github.com/edgexfoundry/app-functions-sdk-go/v3/pkg/interfaces"
 	"github.com/edgexfoundry/app-functions-sdk-go/v3/pkg/util"
-	contractsCommon "github.com/edgexfoundry/go-mod-core-contracts/v3/common"
-
 	clientInterfaces "github.com/edgexfoundry/go-mod-core-contracts/v3/clients/interfaces"
 	"github.com/edgexfoundry/go-mod-core-contracts/v3/clients/logger"
 	coreCommon "github.com/edgexfoundry/go-mod-core-contracts/v3/common"
@@ -742,29 +741,16 @@ func (svc *Service) BuildContext(correlationId string, contentType string) inter
 }
 
 // Publish pushes data to the MessageBus using configured topic
-func (svc *Service) Publish(data any) error {
-	messageClient := bootstrapContainer.MessagingClientFrom(svc.dic.Get)
-	if messageClient == nil {
-		return fmt.Errorf(messageBusDisabledErr)
-	}
-
-	triggertopic := svc.config.Trigger.PublishTopic
-	baseTopic := svc.config.MessageBus.BaseTopicPrefix
-
-	payload, err := json.Marshal(data)
+func (svc *Service) Publish(data any, contentType string) error {
+	err := svc.PublishWithTopic(svc.config.Trigger.PublishTopic, data, contentType)
 	if err != nil {
-		return fmt.Errorf("%v: %v", publishMarshalErr, err)
-	}
-	message := types.NewMessageEnvelope(payload, context.Background())
-	err = messageClient.Publish(message, contractsCommon.BuildTopic(baseTopic, triggertopic))
-	if err != nil {
-		return fmt.Errorf("%v: %v", publishDataErr, err)
+		return err
 	}
 	return nil
 }
 
-// Publish pushes data to the MessageBus using given topic
-func (svc *Service) PublishWithTopic(topic string, data any) error {
+// PublishWithTopic pushes data to the MessageBus using given topic
+func (svc *Service) PublishWithTopic(topic string, data any, contentType string) error {
 	messageClient := bootstrapContainer.MessagingClientFrom(svc.dic.Get)
 	if messageClient == nil {
 		return fmt.Errorf(messageBusDisabledErr)
@@ -776,6 +762,9 @@ func (svc *Service) PublishWithTopic(topic string, data any) error {
 	}
 
 	message := types.NewMessageEnvelope(payload, context.Background())
+	message.CorrelationID = uuid.NewString()
+	message.ContentType = contentType
+
 	err = messageClient.Publish(message, coreCommon.BuildTopic(svc.config.MessageBus.BaseTopicPrefix, topic))
 	if err != nil {
 		return fmt.Errorf("%v: %v", publishDataErr, err)

internal/app/service_test.go

@@ -41,6 +41,7 @@ import (
 	"github.com/edgexfoundry/go-mod-bootstrap/v3/di"
 	clients "github.com/edgexfoundry/go-mod-core-contracts/v3/clients/http"
 	"github.com/edgexfoundry/go-mod-core-contracts/v3/clients/logger"
+	coreCommon "github.com/edgexfoundry/go-mod-core-contracts/v3/common"
 	"github.com/edgexfoundry/go-mod-core-contracts/v3/dtos"
 	messageMocks "github.com/edgexfoundry/go-mod-messaging/v3/messaging/mocks"
 	"github.com/google/uuid"
@@ -1092,7 +1093,7 @@ func TestService_Publish(t *testing.T) {
 				config: tt.config,
 			}
 
-			err := svc.Publish(tt.message)
+			err := svc.Publish(tt.message, coreCommon.ContentTypeJSON)
 			require.Equal(t, tt.expectedError, err)
 		})
 	}
@@ -1177,7 +1178,7 @@ func TestService_PublishWithTopic(t *testing.T) {
 				config: tt.config,
 			}
 
-			err := svc.PublishWithTopic(tt.topic, tt.message)
+			err := svc.PublishWithTopic(tt.topic, tt.message, coreCommon.ContentTypeJSON)
 			require.Equal(t, tt.expectedError, err)
 
 		})

internal/appfunction/context.go

@@ -28,6 +28,7 @@ import (
 	bootstrapContainer "github.com/edgexfoundry/go-mod-bootstrap/v3/bootstrap/container"
 	bootstrapInterfaces "github.com/edgexfoundry/go-mod-bootstrap/v3/bootstrap/interfaces"
 	"github.com/edgexfoundry/go-mod-messaging/v3/pkg/types"
+	"github.com/google/uuid"
 
 	"github.com/edgexfoundry/app-functions-sdk-go/v3/internal/bootstrap/container"
 	"github.com/edgexfoundry/app-functions-sdk-go/v3/pkg/interfaces"
@@ -43,6 +44,7 @@ import (
 const (
 	messageBusDisabledErr = "publish failed due to MessageBus disabled via configuration"
 	publishDataErr        = "failed to publish data to messagebus"
+	topicFormatErr        = "failed to format publish topic"
 )
 
 // NewContext creates, initializes and return a new Context with implements the interfaces.AppFunctionContext interface
@@ -280,44 +282,41 @@ func (appContext *Context) PipelineId() string {
 }
 
 // Publish pushes data to the MessageBus using configured topic
-func (appContext *Context) Publish(data any) error {
-	messageClient := bootstrapContainer.MessagingClientFrom(appContext.Dic.Get)
-	if messageClient == nil {
-		return fmt.Errorf(messageBusDisabledErr)
-	}
+func (appContext *Context) Publish(data any, contentType string) error {
+	topic := container.ConfigurationFrom(appContext.Dic.Get).Trigger.PublishTopic
 
-	triggerConfig := container.ConfigurationFrom(appContext.Dic.Get).Trigger
-	baseTopic := bootstrapContainer.ConfigurationFrom(appContext.Dic.Get).GetBootstrap().MessageBus.BaseTopicPrefix
-
-	payload, err := json.Marshal(data)
-	if err != nil {
-		return fmt.Errorf("failed to marshal data for publishing using configured topic: %v", err)
-	}
-	message := types.NewMessageEnvelope(payload, context.Background())
-	err = messageClient.Publish(message, common.BuildTopic(baseTopic, triggerConfig.PublishTopic))
+	err := appContext.PublishWithTopic(topic, data, contentType)
 	if err != nil {
-		return fmt.Errorf("%v: %v", publishDataErr, err)
+		return err
 	}
 	return nil
 }
 
-// Publish pushes data to the MessageBus for a given topic
-func (appContext *Context) PublishWithTopic(topic string, data any) error {
+// PublishWithTopic pushes data to the MessageBus for a given topic
+func (appContext *Context) PublishWithTopic(topic string, data any, contentType string) error {
 	messageClient := bootstrapContainer.MessagingClientFrom(appContext.Dic.Get)
 	if messageClient == nil {
 		return fmt.Errorf(messageBusDisabledErr)
 	}
 
 	config := bootstrapContainer.ConfigurationFrom(appContext.Dic.Get)
 
-	full_topic := common.BuildTopic(config.GetBootstrap().MessageBus.BaseTopicPrefix, topic)
+	formattedTopic, err := appContext.ApplyValues(topic)
+	if err != nil {
+		return fmt.Errorf("%v: %v", topicFormatErr, err)
+	}
+
+	fullTopic := common.BuildTopic(config.GetBootstrap().MessageBus.BaseTopicPrefix, formattedTopic)
 	payload, err := json.Marshal(data)
 	if err != nil {
 		return fmt.Errorf("failed to marshal data for publishing using given topic: %v", err)
 	}
 
 	message := types.NewMessageEnvelope(payload, context.Background())
-	err = messageClient.Publish(message, full_topic)
+	message.ContentType = contentType
+	message.CorrelationID = uuid.NewString()
+
+	err = messageClient.Publish(message, fullTopic)
 	if err != nil {
 		return fmt.Errorf("%v: %v", publishDataErr, err)
 	}

internal/appfunction/context_test.go

@@ -546,7 +546,7 @@ func TestContext_Publish(t *testing.T) {
 
 			appContext := NewContext("", publishDic, "")
 
-			err := appContext.Publish(tt.message)
+			err := appContext.Publish(tt.message, common.ContentTypeJSON)
 			require.Equal(t, tt.expectedError, err)
 		})
 	}
@@ -634,7 +634,7 @@ func TestService_PublishWithTopic(t *testing.T) {
 
 			appContext := NewContext("", publishDic, "")
 
-			err := appContext.PublishWithTopic(tt.topic, tt.message)
+			err := appContext.PublishWithTopic(tt.topic, tt.message, common.ContentTypeJSON)
 			require.Equal(t, tt.expectedError, err)
 
 		})

pkg/interfaces/context.go

@@ -110,7 +110,7 @@ type AppFunctionContext interface {
 	// PipelineId returns the ID of the pipeline that is executing
 	PipelineId() string
 	// Publish pushes data to the MessageBus using configured topic
-	Publish(data any) error
-	// Publish pushes data to the MessageBus for a given topic
-	PublishWithTopic(topic string, data any) error
+	Publish(data any, contentType string) error
+	// PublishWithTopic pushes data to the MessageBus using given topic
+	PublishWithTopic(topic string, data any, contentType string) error
 }

pkg/interfaces/service.go

@@ -190,7 +190,7 @@ type ApplicationService interface {
 	// BuildContext allows external callers that may need a context (eg background publishers) to easily create one
 	BuildContext(correlationId string, contentType string) AppFunctionContext
 	// Publish pushes data to the MessageBus using configured topic
-	Publish(data any) error
-	// Publish pushes data to the MessageBus using given topic
-	PublishWithTopic(topic string, data any) error
+	Publish(data any, contentType string) error
+	// PublishWithTopic pushes data to the MessageBus using given topic
+	PublishWithTopic(topic string, data any, contentType string) error
 }