Update xp plugin to use message queue config from manager instead of …

…xp management

plugins/turing/config/config.go

@@ -2,6 +2,7 @@ package config
 
 import (
 	"fmt"
+	common_mq_config "github.com/caraml-dev/xp/common/messagequeue"
 	"sync"
 
 	"github.com/caraml-dev/mlp/api/pkg/instrumentation/newrelic"
@@ -49,6 +50,7 @@ type TreatmentServicePluginConfig struct {
 	AssignedTreatmentLogger       config.AssignedTreatmentLoggerConfig `json:"assigned_treatment_logger"`
 	DebugConfig                   config.DebugConfig                   `json:"debug_config"`
 	DeploymentConfig              config.DeploymentConfig              `json:"deployment_config"`
+	MessageQueueConfig            common_mq_config.MessageQueueConfig  `json:"message_queue_config"`
 	ManagementService             config.ManagementServiceConfig       `json:"management_service"`
 	MonitoringConfig              config.Monitoring                    `json:"monitoring_config"`
 	SwaggerConfig                 config.SwaggerConfig                 `json:"swagger_config"`

plugins/turing/manager/experiment_manager.go

@@ -14,7 +14,6 @@ import (
 
 	xpclient "github.com/caraml-dev/xp/clients/management"
 	"github.com/caraml-dev/xp/common/api/schema"
-	common_mq_config "github.com/caraml-dev/xp/common/messagequeue"
 	_config "github.com/caraml-dev/xp/plugins/turing/config"
 	"github.com/caraml-dev/xp/treatment-service/config"
 	"github.com/go-playground/validator/v10"
@@ -147,6 +146,7 @@ func (em *experimentManager) MakeTreatmentServicePluginConfig(
 		AssignedTreatmentLogger:       em.TreatmentServicePluginConfig.AssignedTreatmentLogger,
 		DebugConfig:                   em.TreatmentServicePluginConfig.DebugConfig,
 		DeploymentConfig:              em.TreatmentServicePluginConfig.DeploymentConfig,
+		MessageQueueConfig:            em.TreatmentServicePluginConfig.MessageQueueConfig,
 		ManagementService:             em.TreatmentServicePluginConfig.ManagementService,
 		MonitoringConfig:              em.TreatmentServicePluginConfig.MonitoringConfig,
 		SwaggerConfig:                 em.TreatmentServicePluginConfig.SwaggerConfig,
@@ -155,25 +155,6 @@ func (em *experimentManager) MakeTreatmentServicePluginConfig(
 		SegmenterConfig:               *treatmentServiceConfig.SegmenterConfig,
 		ManagementServicePollerConfig: em.TreatmentServicePluginConfig.ManagementServicePollerConfig,
 	}
-	messageQueueKind := *treatmentServiceConfig.MessageQueueConfig.Kind
-	switch messageQueueKind {
-	case schema.MessageQueueKindPubsub:
-		pluginConfig.MessageQueueConfig = common_mq_config.MessageQueueConfig{
-			Kind: "pubsub",
-			PubSubConfig: &common_mq_config.PubSubConfig{
-				Project:              *treatmentServiceConfig.MessageQueueConfig.PubSub.Project,
-				TopicName:            *treatmentServiceConfig.MessageQueueConfig.PubSub.TopicName,
-				PubSubTimeoutSeconds: em.TreatmentServicePluginConfig.PubSubTimeoutSeconds,
-			},
-		}
-	case schema.MessageQueueKindNoop:
-		pluginConfig.MessageQueueConfig = common_mq_config.MessageQueueConfig{
-			Kind: "",
-		}
-	default:
-		return nil, fmt.Errorf("invalid message queue kind (%s) was provided", messageQueueKind)
-	}
-
 	return pluginConfig, nil
 }
 

plugins/turing/manager/experiment_manager_test.go

@@ -62,6 +62,14 @@ func TestNewExperimentManager(t *testing.T) {
 						"environment_type": "dev",
 						"max_go_routines": 200
 					},
+					"message_queue_config": {
+						"kind": "dev",
+						"pub_sub_config": {
+							"project":"dev",
+							"topic_name":"xp-update",
+							"pub_sub_timeout_seconds": 30
+						}
+					},
 					"management_service": {
 						"authorization_enabled": true,
 						"url": "http://xp-management.global.io/api/xp/v1"