golang源码分析：uber/cadence（2）

package main

import (
  "flag"
  "time"

  "github.com/pborman/uuid"
  "go.uber.org/cadence/client"
  "go.uber.org/cadence/worker"

  common "exp1/common"
)

// This needs to be done as part of a bootstrap step when the process starts.
// The workers are supposed to be long running.
func startWorkers(h *common.SampleHelper) {
  // Configure worker options.
  workerOptions := worker.Options{
    MetricsScope: h.WorkerMetricScope,
    Logger:       h.Logger,
    FeatureFlags: client.FeatureFlags{
      WorkflowExecutionAlreadyCompletedErrorEnabled: true,
    },
  }
  h.StartWorkers(h.Config.DomainName, ApplicationName, workerOptions)
}

func startShadower(h *common.SampleHelper) {
  workerOptions := worker.Options{
    MetricsScope:       h.WorkerMetricScope,
    Logger:             h.Logger,
    EnableShadowWorker: true,
    ShadowOptions: worker.ShadowOptions{
      WorkflowTypes:  []string{helloWorldWorkflowName},
      WorkflowStatus: []string{"Completed"},
      ExitCondition: worker.ShadowExitCondition{
        ShadowCount: 10,
      },
    },
  }
  h.StartWorkers(h.Config.DomainName, ApplicationName, workerOptions)
}

func startWorkflow(h *common.SampleHelper) {
  workflowOptions := client.StartWorkflowOptions{
    ID:                              "helloworld_" + uuid.New(),
    TaskList:                        ApplicationName,
    ExecutionStartToCloseTimeout:    time.Minute,
    DecisionTaskStartToCloseTimeout: time.Minute,
  }
  h.StartWorkflow(workflowOptions, helloWorldWorkflowName, "Cadence")
}

func registerWorkflowAndActivity(
  h *common.SampleHelper,
) {
  h.RegisterWorkflowWithAlias(helloWorldWorkflow, helloWorldWorkflowName)
  h.RegisterActivity(helloWorldActivity)
}

func main() {
  var mode string
  flag.StringVar(&mode, "m", "trigger", "Mode is worker, trigger or shadower.")
  flag.Parse()

  var h common.SampleHelper
  h.SetupServiceConfig()

  switch mode {
  case "worker":
    registerWorkflowAndActivity(&h)
    startWorkers(&h)

    // The workers are supposed to be long running process that should not exit.
    // Use select{} to block indefinitely for samples, you can quit by CMD+C.
    select {}
  case "shadower":
    registerWorkflowAndActivity(&h)
    startShadower(&h)

    select {}
  case "trigger":
    startWorkflow(&h)
  }
}

package main

import (
  "context"
  "time"

  "go.uber.org/cadence/activity"
  "go.uber.org/cadence/workflow"
  "go.uber.org/zap"
)

/**
 * This is the hello world workflow sample.
 */

// ApplicationName is the task list for this sample
const ApplicationName = "helloWorldGroup"

const helloWorldWorkflowName = "helloWorldWorkflow"

// helloWorkflow workflow decider
func helloWorldWorkflow(ctx workflow.Context, name string) error {
  ao := workflow.ActivityOptions{
    ScheduleToStartTimeout: time.Minute,
    StartToCloseTimeout:    time.Minute,
    HeartbeatTimeout:       time.Second * 20,
  }
  ctx = workflow.WithActivityOptions(ctx, ao)

  logger := workflow.GetLogger(ctx)
  logger.Info("helloworld workflow started")
  var helloworldResult string
  err := workflow.ExecuteActivity(ctx, helloWorldActivity, name).Get(ctx, &helloworldResult)
  if err != nil {
    logger.Error("Activity failed.", zap.Error(err))
    return err
  }

  // Adding a new activity to the workflow will result in a non-determinstic change for the workflow
  // Please check https://cadenceworkflow.io/docs/go-client/workflow-versioning/ for more information
  //
  // Un-commenting the following code and the TestReplayWorkflowHistoryFromFile in replay_test.go
  // will fail due to the non-determinstic change
  //
  // If you have a completed workflow execution without the following code and run the
  // TestWorkflowShadowing in shadow_test.go or start the worker in shadow mode (using -m shadower)
  // those two shadowing check will also fail due to the non-deterministic change
  //
  // err := workflow.ExecuteActivity(ctx, helloWorldActivity, name).Get(ctx, &helloworldResult)
  // if err != nil {
  //   logger.Error("Activity failed.", zap.Error(err))
  //   return err
  // }

  logger.Info("Workflow completed.", zap.String("Result", helloworldResult))

  return nil
}

func helloWorldActivity(ctx context.Context, name string) (string, error) {
  logger := activity.GetLogger(ctx)
  logger.Info("helloworld activity started")
  return "Hello " + name + "!", nil
}

type (
  // SampleHelper class for workflow sample helper.
  SampleHelper struct {
    Service            workflowserviceclient.Interface
    WorkerMetricScope  tally.Scope
    ServiceMetricScope tally.Scope
    Logger             *zap.Logger
    Config             Configuration
    Builder            *WorkflowClientBuilder
    DataConverter      encoded.DataConverter
    CtxPropagators     []workflow.ContextPropagator
    workflowRegistries []registryOption
    activityRegistries []registryOption
    Tracer             opentracing.Tracer


    configFile string
  }    

// Configuration for running samples.
  Configuration struct {
    DomainName      string                    `yaml:"domain"`
    ServiceName     string                    `yaml:"service"`
    HostNameAndPort string                    `yaml:"host"`
    Prometheus      *prometheus.Configuration `yaml:"prometheus"`
  }

registryOption struct {
    registry interface{}
    alias    string
  }

func (h *SampleHelper) SetupServiceConfig() {
          if h.configFile == "" {
    h.configFile = defaultConfigFile
        if err := yaml.Unmarshal(configData, &h.Config); err != nil {
        logger, err := zap.NewDevelopment()
        if h.Config.Prometheus != nil {
    reporter, err := h.Config.Prometheus.NewReporter(
      prometheus.ConfigurationOptions{
        Registry: prom.NewRegistry(),
        h.Builder = NewBuilder(logger).
    SetHostPort(h.Config.HostNameAndPort).
    SetDomain(h.Config.DomainName).
    SetMetricsScope(h.ServiceMetricScope).
    SetDataConverter(h.DataConverter).
    SetTracer(h.Tracer).
    SetContextPropagators(h.CtxPropagators)
        service, err := h.Builder.BuildServiceClient()
        h.Service = service
        domainClient, _ := h.Builder.BuildCadenceDomainClient()
  _, err = domainClient.Describe(context.Background(), h.Config.DomainName)
          h.workflowRegistries = make([]registryOption, 0, 1)
  h.activityRegistries = make([]registryOption, 0, 1)  

const (
  defaultConfigFile = "config/development.yaml"

h.RegisterWorkflowWithAlias(helloWorldWorkflow, helloWorldWorkflowName)

func (h *SampleHelper) RegisterWorkflowWithAlias(workflow interface{}, alias string) {
          registryOption := registryOption{
    registry: workflow,
    alias:    alias,
  }
        h.workflowRegistries = append(h.workflowRegistries, registryOption)
    h.RegisterActivity(helloWorldActivity)

func (h *SampleHelper) RegisterActivity(activity interface{}) {
  h.RegisterActivityWithAlias(activity, "")
}      

func (h *SampleHelper) RegisterActivityWithAlias(activity interface{}, alias string) {
          registryOption := registryOption{
    registry: activity,
    alias:    alias,
  }
        h.activityRegistries = append(h.activityRegistries, registryOption)

func helloWorldWorkflow(ctx workflow.Context, name string) error {
          ao := workflow.ActivityOptions{
    ScheduleToStartTimeout: time.Minute,
    StartToCloseTimeout:    time.Minute,
    HeartbeatTimeout:       time.Second * 20,
  }
        err := workflow.ExecuteActivity(ctx, helloWorldActivity, name).Get(ctx, &helloworldResult)  

func helloWorldActivity(ctx context.Context, name string) (string, error) {

func ExecuteActivity(ctx Context, activity interface{}, args ...interface{}) Future {
  return internal.ExecuteActivity(ctx, activity, args...)
}

workflowOptions := client.StartWorkflowOptions{
    ID:                              "helloworld_" + uuid.New(),
    TaskList:                        ApplicationName,
    ExecutionStartToCloseTimeout:    time.Minute,
    DecisionTaskStartToCloseTimeout: time.Minute,
  }
    h.StartWorkflow(workflowOptions, helloWorldWorkflowName, "Cadence")

func startWorkers(h *common.SampleHelper) {
      workerOptions := worker.Options{
    MetricsScope: h.WorkerMetricScope,
    Logger:       h.Logger,
    FeatureFlags: client.FeatureFlags{
      WorkflowExecutionAlreadyCompletedErrorEnabled: true,
    },
  }
      h.StartWorkers(h.Config.DomainName, ApplicationName, workerOptions)

func (h *SampleHelper) StartWorkers(domainName string, groupName string, options worker.Options) {
        worker := worker.New(h.Service, domainName, groupName, options)
        h.registerWorkflowAndActivity(worker)
        err := worker.Start()

func (h *SampleHelper) registerWorkflowAndActivity(worker worker.Worker) {
        for _, w := range h.workflowRegistries {
    if len(w.alias) == 0 {
      worker.RegisterWorkflow(w.registry)
    } else {
      worker.RegisterWorkflowWithOptions(w.registry, workflow.RegisterOptions{Name: w.alias})
    }
  }
  for _, act := range h.activityRegistries {
    if len(act.alias) == 0 {
      worker.RegisterActivity(act.registry)
    } else {
      worker.RegisterActivityWithOptions(act.registry, activity.RegisterOptions{Name: act.alias})
    }
  }

func New(
  service workflowserviceclient.Interface,
  domain string,
  taskList string,
  options Options,
) Worker {
  return internal.NewWorker(service, domain, taskList, options)
}

type (
  // Worker hosts workflow and activity implementations.
  // Use worker.New(...) to create an instance.
  Worker interface {
    Registry


    // Start starts the worker in a non-blocking fashion
    Start() error
    // Run is a blocking start and cleans up resources when killed
    // returns error only if it fails to start the worker
    Run() error
    // Stop cleans up any resources opened by worker
    Stop()
  }