Feat add cloud job (#43)

cmds/protoc-gen-cloud-job/internal/gen.go

@@ -0,0 +1,135 @@
+package internal
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/dave/jennifer/jen"
+	"github.com/pubgo/lava/pkg/proto/cloudjobpb"
+	"google.golang.org/protobuf/compiler/protogen"
+	"google.golang.org/protobuf/proto"
+)
+
+const jobPkg = "github.com/pubgo/lava/component/cloudjobs"
+
+type eventInfo struct {
+	srv *protogen.Service
+	mth *protogen.Method
+}
+
+// GenerateFile generates a .errors.pb.go file containing service definitions.
+func GenerateFile(gen *protogen.Plugin, file *protogen.File) *protogen.GeneratedFile {
+	filename := file.GeneratedFilenamePrefix + ".pb.cloud_job.go"
+	genFile := jen.NewFile(string(file.GoPackageName))
+	genFile.HeaderComment("Code generated by protoc-gen-cloud-job. DO NOT EDIT.")
+	genFile.HeaderComment("versions:")
+	genFile.HeaderComment(fmt.Sprintf("- protoc-gen-cloud-job %s", version))
+	genFile.HeaderComment(fmt.Sprintf("- protoc                 %s", protocVersion(gen)))
+	if file.Proto.GetOptions().GetDeprecated() {
+		genFile.HeaderComment(fmt.Sprintf("%s is a deprecated file.", file.Desc.Path()))
+	} else {
+		genFile.HeaderComment(fmt.Sprintf("source: %s", file.Desc.Path()))
+	}
+
+	g := gen.NewGeneratedFile(filename, file.GoImportPath)
+	g.Skip()
+
+	if len(file.Services) == 0 {
+		return g
+	}
+
+	var events = make(map[string]map[string]*eventInfo)
+	for _, srv := range file.Services {
+		name, ok := proto.GetExtension(srv.Desc.Options(), cloudjobpb.E_JobName).(string)
+		if !ok || name == "" {
+			continue
+		}
+
+		if events[name] == nil {
+			events[name] = map[string]*eventInfo{}
+		}
+
+		for _, m := range srv.Methods {
+			jobSubjectName, ok := proto.GetExtension(m.Desc.Options(), cloudjobpb.E_SubjectName).(string)
+			if !ok || jobSubjectName == "" {
+				continue
+			}
+
+			if events[name][jobSubjectName] != nil {
+				gen.Error(fmt.Errorf("cloud job:%s subject:%s exists", name, jobSubjectName))
+				return g
+			}
+
+			events[name][jobSubjectName] = &eventInfo{srv: srv, mth: m}
+		}
+	}
+
+	if len(events) == 0 {
+		return g
+	}
+
+	g.Unskip()
+	for name, subjects := range events {
+		for subName, info := range subjects {
+			code := strings.ReplaceAll(info.srv.GoName, "InnerService", "")
+			code = strings.ReplaceAll(code, "Service", "")
+			var keyName = fmt.Sprintf("%s%sKey", code, info.mth.GoName)
+			genFile.Commentf("%s %s/%s", keyName, info.srv.GoName, info.mth.GoName)
+			genFile.Const().
+				Id(keyName).
+				Op("=").
+				Lit(subName)
+			genFile.Var().Id("_").Op("=").Qual(jobPkg, "RegisterSubject").
+				Call(jen.Id(keyName), jen.New(jen.Id(info.mth.Input.GoIdent.GoName))).Line()
+
+			genFile.
+				Func().
+				Id(fmt.Sprintf("Register%s%sAsyncJob", code, info.mth.GoName)).
+				Params(
+					jen.Id("jobCli").Op("*").Qual(jobPkg, "Client"),
+					jen.Id("handler").Func().Params(
+						jen.Id("ctx").Op("*").Qual(jobPkg, "Context"),
+						jen.Id("req").Op("*").Id(info.mth.Input.GoIdent.GoName),
+					).Error(),
+				).
+				Block(jen.Qual(jobPkg, "RegisterJobHandler").Call(jen.Id("jobCli"), jen.Lit(name), jen.Id(keyName), jen.Id("handler")))
+			genFile.Line()
+
+			var prefix = fmt.Sprintf("Push%s", code)
+			var mthName = fmt.Sprintf("%sAsyncJob", info.mth.GoName)
+			mthName = handlerPushEventName(mthName, prefix)
+			genFile.
+				Func().
+				Id(mthName).
+				Params(
+					jen.Id("jobCli").Op("*").Qual(jobPkg, "Client"),
+					jen.Id("ctx").Qual("context", "Context"),
+					jen.Id("req").Op("*").Id(info.mth.Input.GoIdent.GoName),
+				).
+				Error().
+				Block(jen.Return().Id("jobCli").Dot("Publish").Call(jen.Id("ctx"), jen.Id(keyName), jen.Id("req")))
+		}
+	}
+
+	g.P(genFile.GoString())
+	return g
+}
+
+func protocVersion(gen *protogen.Plugin) string {
+	v := gen.Request.GetCompilerVersion()
+	if v == nil {
+		return "(unknown)"
+	}
+	var suffix string
+	if s := v.GetSuffix(); s != "" {
+		suffix = "-" + s
+	}
+	return fmt.Sprintf("v%d.%d.%d%s", v.GetMajor(), v.GetMinor(), v.GetPatch(), suffix)
+}
+
+func handlerPushEventName(name string, prefix string) string {
+	if strings.HasPrefix(name, prefix) {
+		return name
+	}
+	return fmt.Sprintf("%s%s", prefix, name)
+}

cmds/protoc-gen-cloud-job/internal/version.go

@@ -0,0 +1,3 @@
+package internal
+
+const version = "v0.0.1"

cmds/protoc-gen-cloud-job/main.go

@@ -0,0 +1,24 @@
+package main
+
+import (
+	"flag"
+
+	"github.com/pubgo/lava/cmds/protoc-gen-cloud-job/internal"
+	"google.golang.org/protobuf/compiler/protogen"
+	"google.golang.org/protobuf/types/pluginpb"
+)
+
+func main() {
+	flag.Parse()
+
+	protogen.Options{ParamFunc: flag.CommandLine.Set}.
+		Run(func(gp *protogen.Plugin) error {
+			gp.SupportedFeatures = uint64(pluginpb.CodeGeneratorResponse_FEATURE_PROTO3_OPTIONAL)
+
+			for _, name := range gp.Request.FileToGenerate {
+				internal.GenerateFile(gp, gp.FilesByPath[name])
+			}
+
+			return nil
+		})
+}

component/cloudjobs/aaa.go

@@ -0,0 +1,14 @@
+package cloudjobs
+
+import (
+	"github.com/pubgo/funk/log"
+	"google.golang.org/protobuf/proto"
+)
+
+var logger = log.GetLogger("cloud_jobs")
+
+type Register interface {
+	RegisterCloudJobs(jobCli *Client)
+}
+
+type JobHandler[T proto.Message] func(ctx *Context, args T) error