Update plugin

api/schema.yaml

@@ -9,13 +9,26 @@ components:
         topic_name:
           type: string
           description: Topic name of the PubSub subscription
+    MessageQueueKind:
+      description: Kind of message queue
+      type: string
+      enum:
+        - noop
+        - pubsub
+    MessageQueueConfig:
+      type: object
+      properties:
+        kind:
+          $ref: '#/components/schemas/MessageQueueKind'
+        pub_sub:
+          $ref: '#/components/schemas/PubSub'
     SegmenterConfig:
       type: object
     TreatmentServiceConfig:
       type: object
       properties:
-        pub_sub:
-          $ref: '#/components/schemas/PubSub'
+        message_queue_config:
+          $ref: '#/components/schemas/MessageQueueConfig'
         segmenter_config:
           $ref: '#/components/schemas/SegmenterConfig'
     SelectedTreatmentData:

management-service/services/configuration_service.go

@@ -16,15 +16,30 @@ type configurationService struct {
 func NewConfigurationService(cfg *config.Config) ConfigurationService {
 	var segmenterConfig schema.SegmenterConfig = cfg.SegmenterConfig
 
-	return &configurationService{
+	var messageQueueKind schema.MessageQueueKind
+	switch cfg.MessageQueueConfig.Kind {
+	case "pubsub":
+		messageQueueKind = schema.MessageQueueKindPubsub
+	case "":
+		messageQueueKind = schema.MessageQueueKindNoop
+	}
+
+	configurationSvc := &configurationService{
 		treatmentServiceConfig: schema.TreatmentServiceConfig{
-			PubSub: &schema.PubSub{
-				Project:   &cfg.MessageQueueConfig.PubSubConfig.Project,
-				TopicName: &cfg.MessageQueueConfig.PubSubConfig.TopicName,
+			MessageQueueConfig: &schema.MessageQueueConfig{
+				Kind: &messageQueueKind,
 			},
 			SegmenterConfig: &segmenterConfig,
 		},
 	}
+	if cfg.MessageQueueConfig.Kind == "pubsub" {
+		configurationSvc.treatmentServiceConfig.MessageQueueConfig.PubSub = &schema.PubSub{
+			Project:   &cfg.MessageQueueConfig.PubSubConfig.Project,
+			TopicName: &cfg.MessageQueueConfig.PubSubConfig.TopicName,
+		}
+	}
+
+	return configurationSvc
 }
 
 func (svc configurationService) GetTreatmentServiceConfig() schema.TreatmentServiceConfig {

management-service/services/configuration_service_test.go

@@ -19,6 +19,7 @@ func (s *ConfigurationServiceTestSuite) SetupSuite() {
 
 	cfg := config.Config{
 		MessageQueueConfig: &config.MessageQueueConfig{
+			Kind: "pubsub",
 			PubSubConfig: &config.PubSubConfig{
 				Project:   "dev",
 				TopicName: "xp-update",
@@ -41,13 +42,17 @@ func TestConfigurationService(t *testing.T) {
 }
 
 func (s *ConfigurationServiceTestSuite) TestGetTreatmentServicePluginConfig() {
+	messageQueueKind := schema.MessageQueueKindPubsub
 	pubSubConfigProject := "dev"
 	pubSubConfigTopicName := "xp-update"
 
 	expectedConfiguration := schema.TreatmentServiceConfig{
-		PubSub: &schema.PubSub{
-			Project:   &pubSubConfigProject,
-			TopicName: &pubSubConfigTopicName,
+		MessageQueueConfig: &schema.MessageQueueConfig{
+			Kind: &messageQueueKind,
+			PubSub: &schema.PubSub{
+				Project:   &pubSubConfigProject,
+				TopicName: &pubSubConfigTopicName,
+			},
 		},
 		SegmenterConfig: &schema.SegmenterConfig{
 			"s2_ids": map[string]interface{}{

management-service/services/message_queue_service.go

@@ -23,7 +23,7 @@ func NewMessageQueueService(mqConfig *config.MessageQueueConfig) (MessageQueueSe
 	case config.PubSubMQ:
 		mq, err = NewPubSubPublisherService(mqConfig.PubSubConfig)
 	default:
-		return nil, fmt.Errorf("invalid message queue config (%s) was provided", mqConfig.Kind)
+		return nil, fmt.Errorf("invalid message queue kind (%s) was provided", mqConfig.Kind)
 	}
 	if err != nil {
 		return nil, err

plugins/turing/manager/experiment_manager.go

@@ -142,7 +142,7 @@ func (em *experimentManager) MakeTreatmentServicePluginConfig(
 	treatmentServiceConfig *schema.TreatmentServiceConfig,
 	projectID int,
 ) (*config.Config, error) {
-	return &config.Config{
+	pluginConfig := &config.Config{
 		Port:                    em.TreatmentServicePluginConfig.Port,
 		ProjectIds:              []string{strconv.Itoa(projectID)},
 		AssignedTreatmentLogger: em.TreatmentServicePluginConfig.AssignedTreatmentLogger,
@@ -153,13 +153,28 @@ func (em *experimentManager) MakeTreatmentServicePluginConfig(
 		SwaggerConfig:           em.TreatmentServicePluginConfig.SwaggerConfig,
 		NewRelicConfig:          em.TreatmentServicePluginConfig.NewRelicConfig,
 		SentryConfig:            em.TreatmentServicePluginConfig.SentryConfig,
-		PubSub: config.PubSub{
-			Project:              *treatmentServiceConfig.PubSub.Project,
-			TopicName:            *treatmentServiceConfig.PubSub.TopicName,
-			PubSubTimeoutSeconds: em.TreatmentServicePluginConfig.PubSubTimeoutSeconds,
-		},
-		SegmenterConfig: *treatmentServiceConfig.SegmenterConfig,
-	}, nil
+		SegmenterConfig:         *treatmentServiceConfig.SegmenterConfig,
+	}
+	messageQueueKind := *treatmentServiceConfig.MessageQueueConfig.Kind
+	switch messageQueueKind {
+	case schema.MessageQueueKindPubsub:
+		pluginConfig.MessageQueueConfig = config.MessageQueueConfig{
+			Kind: "pubsub",
+			PubSubConfig: config.PubSub{
+				Project:              *treatmentServiceConfig.MessageQueueConfig.PubSub.Project,
+				TopicName:            *treatmentServiceConfig.MessageQueueConfig.PubSub.TopicName,
+				PubSubTimeoutSeconds: em.TreatmentServicePluginConfig.PubSubTimeoutSeconds,
+			},
+		}
+	case schema.MessageQueueKindNoop:
+		pluginConfig.MessageQueueConfig = config.MessageQueueConfig{
+			Kind: "",
+		}
+	default:
+		return nil, fmt.Errorf("invalid message queue kind (%s) was provided", messageQueueKind)
+	}
+
+	return pluginConfig, nil
 }
 
 func NewExperimentManager(configData json.RawMessage) (manager.CustomExperimentManager, error) {

plugins/turing/runner/experiment_runner.go

@@ -249,9 +249,9 @@ func (er *experimentRunner) startBackgroundServices(
 	errChannel chan error,
 ) {
 	backgroundSvcCtx := context.Background()
-	if er.appContext.ExperimentSubscriber != nil {
+	if er.appContext.MessageQueueService != nil {
 		go func() {
-			err := er.appContext.ExperimentSubscriber.SubscribeToManagementService(backgroundSvcCtx)
+			err := er.appContext.MessageQueueService.SubscribeToManagementService(backgroundSvcCtx)
 			if err != nil {
 				errChannel <- err
 			}

treatment-service/config/config.go

@@ -107,9 +107,9 @@ type MessageQueueConfig struct {
 }
 
 type PubSub struct {
-	Project              string `json:"project" default:"dev" validate:"required"`
-	TopicName            string `json:"topic_name" default:"xp-update" validate:"required"`
-	PubSubTimeoutSeconds int    `json:"pub_sub_timeout_seconds" default:"30" validate:"required"`
+	Project              string `json:"project" default:"dev"`
+	TopicName            string `json:"topic_name" default:"xp-update"`
+	PubSubTimeoutSeconds int    `json:"pub_sub_timeout_seconds" default:"30"`
 }
 
 type ManagementServiceConfig struct {