add generate mqtt worker from plugin

Author: agungdwiprasetyo

cmd/candi/plugin.go

@@ -11,6 +11,7 @@ type plugin struct {
 const (
 	pluginGCPPubSubWorker = "GCPPubSubWorker"
 	pluginSTOMPWorker     = "STOMPWorker"
+	pluginMQTTWorker      = "MQTTWorker"
 )
 
 var (
@@ -54,7 +55,7 @@ var (
 				`import (`: `import (
 	stompbroker "github.com/golangid/candi-plugin/stomp-broker"`,
 				"brokerDeps := broker.InitBrokers(": `brokerDeps := broker.InitBrokers(
-			stompbroker.NewSTOMPBroker(stompbroker.InitDefaultConnection("[broker host]", "[username]", "[password]")),`,
+			stompbroker.NewSTOMPBroker(stompbroker.InitDefaultConnection("127.0.0.1:61613", "[username]", "[password]")),`,
 			},
 			editAppFactory: map[string]string{
 				`import (`: `import (
@@ -75,5 +76,43 @@ var (
 		stompbroker.STOMPBroker: workerhandler.NewSTOMPWorkerHandler(usecase.GetSharedUsecase(), deps),`,
 			},
 		},
+
+		pluginMQTTWorker: {
+			name:        pluginMQTTWorker,
+			packageName: "github.com/golangid/candi-plugin/mqtt-broker",
+			editConfig: map[string]string{
+				`import (`: `import (
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+	mqttbroker "github.com/golangid/candi-plugin/mqtt-broker"`,
+				"brokerDeps := broker.InitBrokers(": `brokerDeps := broker.InitBrokers(
+			mqttbroker.NewMQTTBroker(mqtt.NewClientOptions().
+				AddBroker("tcp://127.0.0.1:1883").
+				SetClientID("MQTTClientID").
+				SetUsername("MQTTUsername").
+				SetPassword("MQTTPassword").
+				SetCleanSession(false).
+				SetAutoReconnect(true).
+				SetConnectRetry(true),
+			),`,
+			},
+			editAppFactory: map[string]string{
+				`import (`: `import (
+	mqttbroker "github.com/golangid/candi-plugin/mqtt-broker"`,
+				`return
+}`: `apps = append(apps, mqttbroker.NewMQTTSubscriber(
+		service,
+		service.GetDependency().GetBroker(mqttbroker.MQTTBroker),
+	))
+	return
+}`,
+			},
+			editModule: map[string]string{
+				"import (": `import (
+	mqttbroker "github.com/golangid/candi-plugin/mqtt-broker"
+`,
+				"mod.workerHandlers = map[types.Worker]interfaces.WorkerHandler{": `mod.workerHandlers = map[types.Worker]interfaces.WorkerHandler{
+		mqttbroker.MQTTBroker: workerhandler.NewMQTTWorkerHandler(usecase.GetSharedUsecase(), deps),`,
+			},
+		},
 	}
 )

cmd/candi/project_generator_utils.go

@@ -235,6 +235,11 @@ func filterWorkerHandler(cfg *serviceConfig, flagParam *flagParameter) (wording
 		options = append(options, fmt.Sprintf("%d) AMQ (STOMP) Consumer (plugin)", len(options)+1))
 		handlers[strconv.Itoa(len(options))] = pluginSTOMPWorker
 	}
+	if flagParam.addModule || flagParam.initService || (flagParam.addHandler &&
+		validateDir(flagParam.getFullModuleChildDir("delivery", "workerhandler", strings.ToLower(pluginMQTTWorker)+"_handler.go")) != nil) {
+		options = append(options, fmt.Sprintf("%d) MQTT Subscriber (plugin)", len(options)+1))
+		handlers[strconv.Itoa(len(options))] = pluginMQTTWorker
+	}
 
 	wording = strings.Join(options, "\n")
 	return
@@ -366,18 +371,16 @@ func getNeedFileUpdates(srvConfig *serviceConfig) (fileUpdates []fileUpdate) {
 			})
 		}
 		for _, module := range srvConfig.Modules {
-			if module.Skip {
+			if module.Skip && !srvConfig.flag.addHandler {
 				continue
 			}
 			moduleName := cleanSpecialChar.Replace(module.ModuleName)
 			deliveryPackageDir := fmt.Sprintf(`"%s/internal/modules/%s/delivery`, module.PackagePrefix, moduleName)
-			if !module.IsWorkerActive {
-				fileUpdates = append(fileUpdates, fileUpdate{
-					filepath:   rootDir + "internal/modules/" + moduleName + "/module.go",
-					oldContent: "// " + deliveryPackageDir + "/workerhandler",
-					newContent: deliveryPackageDir + "/workerhandler",
-				})
-			}
+			fileUpdates = append(fileUpdates, fileUpdate{
+				filepath:   rootDir + "internal/modules/" + moduleName + "/module.go",
+				oldContent: "// " + deliveryPackageDir + "/workerhandler",
+				newContent: deliveryPackageDir + "/workerhandler",
+			})
 			for before, after := range pl.editModule {
 				fileUpdates = append(fileUpdates, fileUpdate{
 					filepath:   rootDir + "internal/modules/" + moduleName + "/module.go",
@@ -505,11 +508,15 @@ func getAllModuleHandler(path string) (wording string, handlers map[string]strin
 	}
 	if validateDir(path+"/workerhandler/"+strings.ToLower(pluginGCPPubSubWorker)+"_handler.go") == nil {
 		options = append(options, fmt.Sprintf("%d) GCP PubSub Subscriber (plugin)", len(options)+1))
-		handlers[strconv.Itoa(len(options))] = RabbitmqHandler
+		handlers[strconv.Itoa(len(options))] = pluginGCPPubSubWorker
 	}
 	if validateDir(path+"/workerhandler/"+strings.ToLower(pluginSTOMPWorker)+"_handler.go") == nil {
 		options = append(options, fmt.Sprintf("%d) AMQ (STOMP) Consumer (plugin)", len(options)+1))
-		handlers[strconv.Itoa(len(options))] = RabbitmqHandler
+		handlers[strconv.Itoa(len(options))] = pluginSTOMPWorker
+	}
+	if validateDir(path+"/workerhandler/"+strings.ToLower(pluginMQTTWorker)+"_handler.go") == nil {
+		options = append(options, fmt.Sprintf("%d) MQTT Subscriber (plugin)", len(options)+1))
+		handlers[strconv.Itoa(len(options))] = pluginMQTTWorker
 	}
 
 	wording = strings.Join(options, "\n")

cmd/candi/template_delivery_worker.go

@@ -358,7 +358,7 @@ func (h *{{.WorkerPluginName}}Handler) handleTopic{{upper (camel .ModuleName)}}(
 	trace, ctx := tracer.StartTraceWithContext(eventContext.Context(), "{{upper (camel .ModuleName)}}Delivery{{.WorkerPluginName}}:HandleTopic{{upper (camel .ModuleName)}}")
 	defer trace.Finish()
 
-	fmt.Printf("message consumed by module {{.ModuleName}}. message: %s\n", eventContext.Context())
+	fmt.Printf("message consumed by module {{.ModuleName}}. message: %s\n", eventContext.Message())
 
 	// exec usecase
 	// h.uc.SomethingUsecase()