Add Reset API with new option of skip signal reapply and fix all lint errors (#1032)

longquanzheng · web-flow · commit 0256258b905b · 2020-12-04T12:27:33.000-08:00
diff --git a/.gen/go/shared/idl.go b/.gen/go/shared/idl.go
diff --git a/.gen/go/shared/types.go b/.gen/go/shared/types.go
diff --git a/activity/activity.go b/activity/activity.go
@@ -46,15 +46,13 @@ type (
 // that could report the activity completed event to cadence server via Client.CompleteActivity() API.
 var ErrResultPending = internal.ErrActivityResultPending
 
+// Register - calls RegisterWithOptions with default registration options.
 // Deprecated: Global activity registration methods are replaced by equivalent Worker instance methods.
 // This method is kept to maintain backward compatibility and should not be used.
-// Register - calls RegisterWithOptions with default registration options.
 func Register(activityFunc interface{}) {
 	internal.RegisterActivity(activityFunc)
 }
 
-// Deprecated: Global activity registration methods are replaced by equivalent Worker instance methods.
-// This method is kept to maintain backward compatibility and should not be used.
 // RegisterWithOptions registers the activity function with options
 // The user can use options to provide an external name for the activity or leave it empty if no
 // external name is required. This can be used as
@@ -72,6 +70,8 @@ func Register(activityFunc interface{}) {
 // If function implementation returns activity.ErrResultPending then activity is not completed from the
 // calling workflow point of view. See documentation of activity.ErrResultPending for more info.
 // This method calls panic if activityFunc doesn't comply with the expected format.
+// Deprecated: Global activity registration methods are replaced by equivalent Worker instance methods.
+// This method is kept to maintain backward compatibility and should not be used.
 func RegisterWithOptions(activityFunc interface{}, opts RegisterOptions) {
 	internal.RegisterActivityWithOptions(activityFunc, opts)
 }
diff --git a/client/client.go b/client/client.go
@@ -326,6 +326,14 @@ type (
 		//  - QueryFailError
 		QueryWorkflowWithOptions(ctx context.Context, request *QueryWorkflowWithOptionsRequest) (*QueryWorkflowWithOptionsResponse, error)
 
+		// ResetWorkflow reset a given workflow execution and returns a new execution
+		// See ResetWorkflowRequest and ResetWorkflowResponse for more information.
+		// The errors it can return:
+		//  - BadRequestError
+		//  - InternalServiceError
+		//  - EntityNotExistError
+		ResetWorkflow(ctx context.Context, request *s.ResetWorkflowExecutionRequest) (*s.ResetWorkflowExecutionResponse, error)
+
 		// DescribeWorkflowExecution returns information about the specified workflow execution.
 		// - runID can be default(empty string). if empty string then it will pick the last running execution of that workflow ID.
 		//
diff --git a/idls b/idls
@@ -1 +1 @@
-Subproject commit 54a04b87e05c733cd9f083b42e9ed226ede24f59
+Subproject commit 5216c3d51ce4a29b91bc77f427fbc97348387b4f
diff --git a/internal/activity.go b/internal/activity.go
@@ -127,8 +127,6 @@ type (
 	}
 )
 
-// Deprecated: Global activity registration methods are replaced by equivalent Worker instance methods.
-// This method is kept to maintain backward compatibility and should not be used.
 // RegisterActivity - register an activity function or a pointer to a structure with the framework.
 // The public form is: activity.Register(...)
 // An activity function takes a context and input and returns a (result, error) or just error.
@@ -158,12 +156,12 @@ type (
 //
 // Serialization of all primitive types, structures is supported ... except channels, functions, variadic, unsafe pointer.
 // This method calls panic if activityFunc doesn't comply with the expected format.
+// Deprecated: Global activity registration methods are replaced by equivalent Worker instance methods.
+// This method is kept to maintain backward compatibility and should not be used.
 func RegisterActivity(activityFunc interface{}) {
 	RegisterActivityWithOptions(activityFunc, RegisterActivityOptions{})
 }
 
-// Deprecated: Global activity registration methods are replaced by equivalent Worker instance methods.
-// This method is kept to maintain backward compatibility and should not be used.
 // RegisterActivityWithOptions registers the activity function or struct pointer with options.
 // The public form is: activity.RegisterWithOptions(...)
 // The user can use options to provide an external name for the activity or leave it empty if no
@@ -181,6 +179,8 @@ func RegisterActivity(activityFunc interface{}) {
 // The other use of options is to disable duplicated activity registration check
 // which might be useful for integration tests.
 // activity.RegisterWithOptions(barActivity, RegisterActivityOptions{DisableAlreadyRegisteredCheck: true})
+// Deprecated: Global activity registration methods are replaced by equivalent Worker instance methods.
+// This method is kept to maintain backward compatibility and should not be used.
 func RegisterActivityWithOptions(activityFunc interface{}, opts RegisterActivityOptions) {
 	registry := getGlobalRegistry()
 	registry.RegisterActivityWithOptions(activityFunc, opts)
diff --git a/internal/client.go b/internal/client.go
@@ -301,6 +301,14 @@ type (
 		//  - QueryFailError
 		QueryWorkflowWithOptions(ctx context.Context, request *QueryWorkflowWithOptionsRequest) (*QueryWorkflowWithOptionsResponse, error)
 
+		// ResetWorkflow reset a given workflow execution and returns a new execution
+		// See QueryWorkflowWithOptionsRequest and QueryWorkflowWithOptionsResponse for more information.
+		// The errors it can return:
+		//  - BadRequestError
+		//  - InternalServiceError
+		//  - EntityNotExistError
+		ResetWorkflow(ctx context.Context, request *s.ResetWorkflowExecutionRequest) (*s.ResetWorkflowExecutionResponse, error)
+
 		// DescribeWorkflowExecution returns information about the specified workflow execution.
 		// The errors it can return:
 		//  - BadRequestError
diff --git a/internal/internal_workflow_client.go b/internal/internal_workflow_client.go
@@ -773,6 +773,26 @@ func (wc *workflowClient) CountWorkflow(ctx context.Context, request *s.CountWor
 	return response, nil
 }
 
+// ResetWorkflow implementation
+func (wc *workflowClient) ResetWorkflow(ctx context.Context, request *s.ResetWorkflowExecutionRequest) (*s.ResetWorkflowExecutionResponse, error) {
+	if len(request.GetDomain()) == 0 {
+		request.Domain = common.StringPtr(wc.domain)
+	}
+	var response *s.ResetWorkflowExecutionResponse
+	err := backoff.Retry(ctx,
+		func() error {
+			var err1 error
+			tchCtx, cancel, opt := newChannelContext(ctx)
+			defer cancel()
+			response, err1 = wc.workflowService.ResetWorkflowExecution(tchCtx, request, opt...)
+			return err1
+		}, createDynamicServiceRetryPolicy(ctx), isServiceTransientError)
+	if err != nil {
+		return nil, err
+	}
+	return response, nil
+}
+
 // GetSearchAttributes implementation
 func (wc *workflowClient) GetSearchAttributes(ctx context.Context) (*s.GetSearchAttributesResponse, error) {
 	var response *s.GetSearchAttributesResponse
diff --git a/internal/internal_workflow_testsuite.go b/internal/internal_workflow_testsuite.go
@@ -154,7 +154,7 @@ type (
 		onTimerFiredListener             func(timerID string)
 		onTimerCancelledListener         func(timerID string)
 
-		cronMaxIterations  int
+		cronMaxIterations int
 	}
 
 	// testWorkflowEnvironmentImpl is the environment that runs the workflow/activity unit tests.
@@ -186,10 +186,9 @@ type (
 		workerStopChannel  chan struct{}
 		sessionEnvironment *testSessionEnvironmentImpl
 
-		cronSchedule       string
-		cronIterations     int
-		workflowInput      []byte
-
+		cronSchedule   string
+		cronIterations int
+		workflowInput  []byte
 	}
 
 	testSessionEnvironmentImpl struct {
@@ -831,7 +830,7 @@ func (env *testWorkflowEnvironmentImpl) Complete(result []byte, err error) {
 	// 1. Child-Workflows
 	// 2. non-cron Workflows
 	if env.isChildWorkflow() && !env.IsCron() {
-	  close(env.doneChannel)
+		close(env.doneChannel)
 	}
 
 	if env.isChildWorkflow() {
@@ -919,7 +918,7 @@ func (h *testWorkflowHandle) rerun(asChild bool) bool {
 				workflowNow := env.Now().In(time.UTC)
 				backoff := schedule.Next(workflowNow).Sub(workflowNow)
 				if backoff > 0 {
-					env.cronIterations += 1
+					env.cronIterations++
 					delete(env.runningWorkflows, env.workflowInfo.WorkflowExecution.ID)
 					params.attempt = 0
 					params.scheduledTime = env.Now()
@@ -943,7 +942,7 @@ func (h *testWorkflowHandle) rerun(asChild bool) bool {
 				workflowNow := env.Now().In(time.UTC)
 				backoff := schedule.Next(workflowNow).Sub(workflowNow)
 				if backoff > 0 {
-					env.cronIterations += 1
+					env.cronIterations++
 					// Prepare the env for the next iteration
 					env.runningCount--
 					env.setLastCompletionResult(result)
@@ -955,7 +954,7 @@ func (h *testWorkflowHandle) rerun(asChild bool) bool {
 						// Use the existing headers and input
 						env.workflowDef.Execute(env, env.header, env.workflowInput)
 						env.startDecisionTask()
-					}, backoff - backoff)
+					}, backoff-backoff)
 					return true
 				}
 			}
diff --git a/internal/internal_workflow_testsuite_test.go b/internal/internal_workflow_testsuite_test.go
@@ -2804,7 +2804,7 @@ func (s *WorkflowTestSuiteUnitTest) Test_CronChildWorkflow() {
 		if HasLastCompletionResult(ctx) {
 			GetLastCompletionResult(ctx, &result)
 		}
-		Sleep(ctx, time.Second * 3)
+		Sleep(ctx, time.Second*3)
 		if info.Attempt == 0 {
 			failedCount++
 			return 0, errors.New("please-retry")
@@ -2831,7 +2831,7 @@ func (s *WorkflowTestSuiteUnitTest) Test_CronChildWorkflow() {
 
 		cronFuture := ExecuteChildWorkflow(ctx1, cronWorkflow) // cron never stop so this future won't return
 
-		timeoutTimer := NewTimer(ctx, time.Hour * 3)
+		timeoutTimer := NewTimer(ctx, time.Hour*3)
 		selector := NewSelector(ctx)
 		var err error
 		selector.AddFuture(cronFuture, func(f Future) {
@@ -2867,7 +2867,7 @@ func (s *WorkflowTestSuiteUnitTest) Test_CronWorkflow() {
 		if HasLastCompletionResult(ctx) {
 			GetLastCompletionResult(ctx, &result)
 		}
-		Sleep(ctx, time.Second * 3)
+		Sleep(ctx, time.Second*3)
 		result++
 		return result, nil
 	}
diff --git a/internal/worker.go b/internal/worker.go
@@ -250,11 +250,11 @@ func NewWorker(
 	return newAggregatedWorker(service, domain, taskList, options)
 }
 
-// Deprecated: Global workflow replay methods are replaced by equivalent WorkflowReplayer instance methods.
-// This method is kept to maintain backward compatibility and should not be used.
 // ReplayWorkflowExecution loads a workflow execution history from the Cadence service and executes a single decision task for it.
 // Use for testing backwards compatibility of code changes and troubleshooting workflows in a debugger.
 // The logger is the only optional parameter. Defaults to the noop logger.
+// Deprecated: Global workflow replay methods are replaced by equivalent WorkflowReplayer instance methods.
+// This method is kept to maintain backward compatibility and should not be used.
 func ReplayWorkflowExecution(
 	ctx context.Context,
 	service workflowserviceclient.Interface,
@@ -266,32 +266,32 @@ func ReplayWorkflowExecution(
 	return r.ReplayWorkflowExecution(ctx, service, logger, domain, execution)
 }
 
-// Deprecated: Global workflow replay methods are replaced by equivalent WorkflowReplayer instance methods.
-// This method is kept to maintain backward compatibility and should not be used.
 // ReplayWorkflowHistory executes a single decision task for the given history.
 // Use for testing the backwards compatibility of code changes and troubleshooting workflows in a debugger.
 // The logger is an optional parameter. Defaults to the noop logger.
+// Deprecated: Global workflow replay methods are replaced by equivalent WorkflowReplayer instance methods.
+// This method is kept to maintain backward compatibility and should not be used.
 func ReplayWorkflowHistory(logger *zap.Logger, history *shared.History) error {
 	r := NewWorkflowReplayer()
 	return r.ReplayWorkflowHistory(logger, history)
 }
 
-// Deprecated: Global workflow replay methods are replaced by equivalent WorkflowReplayer instance methods.
-// This method is kept to maintain backward compatibility and should not be used.
 // ReplayWorkflowHistoryFromJSONFile executes a single decision task for the given json history file.
 // Use for testing backwards compatibility of code changes and troubleshooting workflows in a debugger.
 // The logger is an optional parameter. Defaults to the noop logger.
+// Deprecated: Global workflow replay methods are replaced by equivalent WorkflowReplayer instance methods.
+// This method is kept to maintain backward compatibility and should not be used.
 func ReplayWorkflowHistoryFromJSONFile(logger *zap.Logger, jsonfileName string) error {
 	r := NewWorkflowReplayer()
 	return r.ReplayWorkflowHistoryFromJSONFile(logger, jsonfileName)
 }
 
-// Deprecated: Global workflow replay methods are replaced by equivalent WorkflowReplayer instance methods.
-// This method is kept to maintain backward compatibility and should not be used.
 // ReplayPartialWorkflowHistoryFromJSONFile executes a single decision task for the given json history file upto provided
 // lastEventID(inclusive).
 // Use for testing backwards compatibility of code changes and troubleshooting workflows in a debugger.
 // The logger is an optional parameter. Defaults to the noop logger.
+// Deprecated: Global workflow replay methods are replaced by equivalent WorkflowReplayer instance methods.
+// This method is kept to maintain backward compatibility and should not be used.
 func ReplayPartialWorkflowHistoryFromJSONFile(logger *zap.Logger, jsonfileName string, lastEventID int64) error {
 	r := NewWorkflowReplayer()
 	return r.ReplayPartialWorkflowHistoryFromJSONFile(logger, jsonfileName, lastEventID)
diff --git a/internal/workflow.go b/internal/workflow.go
@@ -233,8 +233,6 @@ type RegisterWorkflowOptions struct {
 	DisableAlreadyRegisteredCheck bool
 }
 
-// Deprecated: Global workflow registration methods are replaced by equivalent Worker instance methods.
-// This method is kept to maintain backward compatibility and should not be used.
 // RegisterWorkflow - registers a workflow function with the framework.
 // The public form is: workflow.Register(...)
 // A workflow takes a cadence context and input and returns a (result, error) or just error.
@@ -245,12 +243,12 @@ type RegisterWorkflowOptions struct {
 //	func sampleWorkflow(ctx workflow.Context, arg1 int) (result string, err error)
 // Serialization of all primitive types, structures is supported ... except channels, functions, variadic, unsafe pointer.
 // This method calls panic if workflowFunc doesn't comply with the expected format.
+// Deprecated: Global workflow registration methods are replaced by equivalent Worker instance methods.
+// This method is kept to maintain backward compatibility and should not be used.
 func RegisterWorkflow(workflowFunc interface{}) {
 	RegisterWorkflowWithOptions(workflowFunc, RegisterWorkflowOptions{})
 }
 
-// Deprecated: Global workflow registration methods are replaced by equivalent Worker instance methods.
-// This method is kept to maintain backward compatibility and should not be used.
 // RegisterWorkflowWithOptions registers the workflow function with options.
 // The public form is: workflow.RegisterWithOptions(...)
 // The user can use options to provide an external name for the workflow or leave it empty if no
@@ -266,6 +264,8 @@ func RegisterWorkflow(workflowFunc interface{}) {
 // Serialization of all primitive types, structures is supported ... except channels, functions, variadic, unsafe pointer.
 // This method calls panic if workflowFunc doesn't comply with the expected format or tries to register the same workflow
 // type name twice. Use workflow.RegisterOptions.DisableAlreadyRegisteredCheck to allow multiple registrations.
+// Deprecated: Global workflow registration methods are replaced by equivalent Worker instance methods.
+// This method is kept to maintain backward compatibility and should not be used.
 func RegisterWorkflowWithOptions(workflowFunc interface{}, opts RegisterWorkflowOptions) {
 	registry := getGlobalRegistry()
 	registry.RegisterWorkflowWithOptions(workflowFunc, opts)
@@ -744,6 +744,7 @@ type WorkflowInfo struct {
 	RetryPolicy                         *s.RetryPolicy
 }
 
+// GetBinaryChecksum returns the binary checksum(identifier) of this worker
 func (wInfo *WorkflowInfo) GetBinaryChecksum() string {
 	if wInfo.BinaryChecksum == nil {
 		return getBinaryChecksum()
diff --git a/internal/workflow_testsuite.go b/internal/workflow_testsuite.go
@@ -179,21 +179,25 @@ func (t *TestActivityEnvironment) SetWorkerStopChannel(c chan struct{}) {
 	t.impl.setWorkerStopChannel(c)
 }
 
+// RegisterWorkflow register workflows
 func (t *TestWorkflowEnvironment) RegisterWorkflow(w interface{}) {
 	t.impl.RegisterWorkflow(w)
 }
 
+// RegisterWorkflowWithOptions register workflows
 func (t *TestWorkflowEnvironment) RegisterWorkflowWithOptions(w interface{}, options RegisterWorkflowOptions) {
 	if len(t.ExpectedCalls) > 0 {
 		panic("RegisterWorkflow calls cannot follow mock related ones like OnWorkflow or similar")
 	}
 	t.impl.RegisterWorkflowWithOptions(w, options)
 }
 
+// RegisterActivity registers activity
 func (t *TestWorkflowEnvironment) RegisterActivity(a interface{}) {
 	t.impl.RegisterActivity(a)
 }
 
+// RegisterActivityWithOptions registers activity
 func (t *TestWorkflowEnvironment) RegisterActivityWithOptions(a interface{}, options RegisterActivityOptions) {
 	if len(t.ExpectedCalls) > 0 {
 		panic("RegisterActivity calls cannot follow mock related ones like OnActivity or similar")
@@ -460,7 +464,7 @@ func (t *TestWorkflowEnvironment) SetWorkflowCronSchedule(cron string) *TestWork
 // SetWorkflowCronMaxIterations sets the a limit on the number of Cron iterations, not including the first one
 // of the tested workflow.
 func (t *TestWorkflowEnvironment) SetWorkflowCronMaxIterations(maxIterations int) *TestWorkflowEnvironment {
-	t.impl.setCronMaxIterationas(maxIterations )
+	t.impl.setCronMaxIterationas(maxIterations)
 	return t
 }
 
diff --git a/mocks/Client.go b/mocks/Client.go
diff --git a/workflow/deterministic_wrappers.go b/workflow/deterministic_wrappers.go
@@ -49,7 +49,7 @@ type (
 	WaitGroup = internal.WaitGroup
 )
 
-// Blocks the calling thread until condition() returns true.
+// Await blocks the calling thread until condition() returns true.
 // Do not mutate values or trigger side effects inside condition.
 // Returns CanceledError if the ctx is canceled.
 // The following code is going to block until the captured count
diff --git a/workflow/workflow.go b/workflow/workflow.go
