CLOUDP-368428-Stream-Processor Constants and Aligning with Flex Cluster

Author: ParthasarathyV

cfn-resources/stream-processor/cmd/resource/callbacks.go

@@ -147,15 +147,15 @@ func handleCreateCallback(ctx context.Context, client *util.MongoDBClient, curre
 			if peErr := startStreamProcessor(ctx, client.AtlasSDK, callbackCtx.ProjectID, callbackCtx.WorkspaceOrInstanceName, callbackCtx.ProcessorName); peErr != nil {
 				return *peErr
 			}
-			return createInProgressEvent("Starting stream processor", currentModel, callbackContext)
+			return createInProgressEvent(constants.Pending, currentModel, callbackContext)
 		}
-		return finalizeModel(streamProcessor, currentModel, "Create Completed")
+		return finalizeModel(streamProcessor, currentModel, constants.Complete)
 
 	case StartedState:
-		return finalizeModel(streamProcessor, currentModel, "Create Completed")
+		return finalizeModel(streamProcessor, currentModel, constants.Complete)
 
 	case InitiatingState, CreatingState:
-		return createInProgressEvent(fmt.Sprintf("Creating stream processor (current state: %s)", currentState), currentModel, callbackContext)
+		return createInProgressEvent(constants.Pending, currentModel, callbackContext)
 
 	case FailedState:
 		return handler.ProgressEvent{
@@ -214,14 +214,14 @@ func handleUpdateCallback(ctx context.Context, client *util.MongoDBClient, curre
 			if peErr := startStreamProcessor(ctx, client.AtlasSDK, callbackCtx.ProjectID, callbackCtx.WorkspaceOrInstanceName, callbackCtx.ProcessorName); peErr != nil {
 				return *peErr
 			}
-			return createInProgressEvent("Starting stream processor", currentModel, callbackContext)
+			return createInProgressEvent(constants.Pending, currentModel, callbackContext)
 		}
 
-		return finalizeModel(streamProcessorResp, currentModel, "Update Completed")
+		return finalizeModel(streamProcessorResp, currentModel, constants.Complete)
 
 	case StartedState:
 		if desiredState == StartedState {
-			return finalizeModel(streamProcessor, currentModel, "Update Completed")
+			return finalizeModel(streamProcessor, currentModel, constants.Complete)
 		}
 
 		_, err := client.AtlasSDK.StreamsApi.StopStreamProcessorWithParams(ctx,
@@ -237,7 +237,7 @@ func handleUpdateCallback(ctx context.Context, client *util.MongoDBClient, curre
 				Message:         fmt.Sprintf("Error stopping stream processor: %s", err.Error()),
 			}
 		}
-		return createInProgressEvent("Stopping stream processor", currentModel, callbackContext)
+		return createInProgressEvent(constants.Pending, currentModel, callbackContext)
 
 	case FailedState:
 		return handler.ProgressEvent{
@@ -246,6 +246,6 @@ func handleUpdateCallback(ctx context.Context, client *util.MongoDBClient, curre
 		}
 
 	default:
-		return createInProgressEvent(fmt.Sprintf("Updating stream processor (current state: %s)", currentState), currentModel, callbackContext)
+		return createInProgressEvent(constants.Pending, currentModel, callbackContext)
 	}
 }

cfn-resources/stream-processor/cmd/resource/share.go

@@ -28,13 +28,13 @@ import (
 	"github.com/mongodb/mongodbatlas-cloudformation-resources/util/constants"
 )
 
-func isCallback(req *handler.Request) bool {
+func IsCallback(req *handler.Request) bool {
 	_, found := req.CallbackContext["callbackStreamProcessor"]
 	return found
 }
 
 func HandleCreate(req *handler.Request, client *util.MongoDBClient, model *Model) handler.ProgressEvent {
-	if isCallback(req) {
+	if IsCallback(req) {
 		callbackCtx := getCallbackData(*req)
 		if peErr := validateCallbackData(callbackCtx); peErr != nil {
 			return *peErr
@@ -107,7 +107,7 @@ func HandleCreate(req *handler.Request, client *util.MongoDBClient, model *Model
 
 	return handler.ProgressEvent{
 		OperationStatus:      handler.InProgress,
-		Message:              "Creating stream processor",
+		Message:              constants.Pending,
 		ResourceModel:        inProgressModel,
 		CallbackDelaySeconds: defaultCallbackDelaySeconds,
 		CallbackContext: buildCallbackContext(projectID, workspaceOrInstanceName, processorName, map[string]any{
@@ -160,13 +160,13 @@ func HandleRead(req *handler.Request, client *util.MongoDBClient, model *Model)
 
 	return handler.ProgressEvent{
 		OperationStatus: handler.Success,
-		Message:         "Read Completed",
+		Message:         constants.ReadComplete,
 		ResourceModel:   resourceModel,
 	}
 }
 
 func HandleUpdate(req *handler.Request, client *util.MongoDBClient, prevModel *Model, model *Model) handler.ProgressEvent {
-	if isCallback(req) {
+	if IsCallback(req) {
 		callbackCtx := getCallbackData(*req)
 		if peErr := validateCallbackData(callbackCtx); peErr != nil {
 			return *peErr
@@ -249,7 +249,7 @@ func HandleUpdate(req *handler.Request, client *util.MongoDBClient, prevModel *M
 
 		return handler.ProgressEvent{
 			OperationStatus:      handler.InProgress,
-			Message:              "Stopping stream processor",
+			Message:              constants.Pending,
 			ResourceModel:        inProgressModel,
 			CallbackDelaySeconds: defaultCallbackDelaySeconds,
 			CallbackContext: buildCallbackContext(projectID, workspaceOrInstanceName, processorName, map[string]any{
@@ -295,7 +295,7 @@ func HandleUpdate(req *handler.Request, client *util.MongoDBClient, prevModel *M
 
 		return handler.ProgressEvent{
 			OperationStatus:      handler.InProgress,
-			Message:              "Starting stream processor",
+			Message:              constants.Pending,
 			ResourceModel:        inProgressModel,
 			CallbackDelaySeconds: defaultCallbackDelaySeconds,
 			CallbackContext: buildCallbackContext(projectID, workspaceOrInstanceName, processorName, map[string]any{
@@ -304,7 +304,7 @@ func HandleUpdate(req *handler.Request, client *util.MongoDBClient, prevModel *M
 		}
 	}
 
-	return finalizeModel(streamProcessorResp, model, "Update Completed")
+	return finalizeModel(streamProcessorResp, model, constants.Complete)
 }
 
 func HandleDelete(req *handler.Request, client *util.MongoDBClient, model *Model) handler.ProgressEvent {
@@ -337,7 +337,7 @@ func HandleDelete(req *handler.Request, client *util.MongoDBClient, model *Model
 
 	return handler.ProgressEvent{
 		OperationStatus: handler.Success,
-		Message:         "Delete Completed",
+		Message:         constants.Complete,
 	}
 }
 
@@ -374,7 +374,7 @@ func HandleList(req *handler.Request, client *util.MongoDBClient, model *Model)
 
 	return handler.ProgressEvent{
 		OperationStatus: handler.Success,
-		Message:         "List Completed",
+		Message:         constants.Complete,
 		ResourceModels:  response,
 	}
 }