Create GetWorkflow method allowing to wait on workflow completion (#706)

Author: shreyassrivatsan

client/client.go

@@ -78,7 +78,6 @@ type (
 		// The errors it can return:
 		//	- EntityNotExistsError, if domain does not exists
 		//	- BadRequestError
-		//	- WorkflowExecutionAlreadyStartedError
 		//	- InternalServiceError
 		//
 		// WorkflowRun has 2 methods:
@@ -95,6 +94,23 @@ type (
 		// NOTE: DO NOT USE THIS API INSIDE A WORKFLOW, USE workflow.ExecuteChildWorkflow instead
 		ExecuteWorkflow(ctx context.Context, options StartWorkflowOptions, workflow interface{}, args ...interface{}) (WorkflowRun, error)
 
+		// GetWorkfow retrieves a workflow execution and return a WorkflowRun instance (described above)
+		// - workflow ID of the workflow.
+		// - runID can be default(empty string). if empty string then it will pick the last running execution of that workflow ID.
+		//
+		// WorkflowRun has 2 methods:
+		//  - GetRunID() string: which return the first started workflow run ID (please see below)
+		//  - Get(ctx context.Context, valuePtr interface{}) error: which will fill the workflow
+		//    execution result to valuePtr, if workflow execution is a success, or return corresponding
+		//    error. This is a blocking API.
+		// NOTE: if the started workflow return ContinueAsNewError during the workflow execution, the
+		// return result of GetRunID() will be the started workflow run ID, not the new run ID caused by ContinueAsNewError,
+		// however, Get(ctx context.Context, valuePtr interface{}) will return result from the run which did not return ContinueAsNewError.
+		// Say ExecuteWorkflow started a workflow, in its first run, has run ID "run ID 1", and returned ContinueAsNewError,
+		// the second run has run ID "run ID 2" and return some result other than ContinueAsNewError:
+		// GetRunID() will always return "run ID 1" and  Get(ctx context.Context, valuePtr interface{}) will return the result of second run.
+		GetWorkflow(ctx context.Context, workflowID string, runID string) WorkflowRun
+
 		// SignalWorkflow sends a signals to a workflow in execution
 		// - workflow ID of the workflow.
 		// - runID can be default(empty string). if empty string then it will pick the running execution of that workflow ID.

internal/client.go

@@ -65,7 +65,6 @@ type (
 		// The errors it can return:
 		//	- EntityNotExistsError, if domain does not exists
 		//	- BadRequestError
-		//	- WorkflowExecutionAlreadyStartedError
 		//	- InternalServiceError
 		//
 		// The current timeout resolution implementation is in seconds and uses math.Ceil(d.Seconds()) as the duration. But is
@@ -86,6 +85,21 @@ type (
 		// NOTE: DO NOT USE THIS API INSIDE A WORKFLOW, USE workflow.ExecuteChildWorkflow instead
 		ExecuteWorkflow(ctx context.Context, options StartWorkflowOptions, workflow interface{}, args ...interface{}) (WorkflowRun, error)
 
+		// GetWorkfow retrieves a workflow execution and return a WorkflowRun instance
+		// - workflow ID of the workflow.
+		// - runID can be default(empty string). if empty string then it will pick the last running execution of that workflow ID.
+		//
+		// WorkflowRun has three methods:
+		//  - GetID() string: which return workflow ID (which is same as StartWorkflowOptions.ID if provided)
+		//  - GetRunID() string: which return the first started workflow run ID (please see below)
+		//  - Get(ctx context.Context, valuePtr interface{}) error: which will fill the workflow
+		//    execution result to valuePtr, if workflow execution is a success, or return corresponding
+		//    error. This is a blocking API.
+		// NOTE: if the retrieved workflow returned ContinueAsNewError during the workflow execution, the
+		// return result of GetRunID() will be the retrieved workflow run ID, not the new run ID caused by ContinueAsNewError,
+		// however, Get(ctx context.Context, valuePtr interface{}) will return result from the run which did not return ContinueAsNewError.
+		GetWorkflow(ctx context.Context, workflowID string, runID string) WorkflowRun
+
 		// SignalWorkflow sends a signals to a workflow in execution
 		// - workflow ID of the workflow.
 		// - runID can be default(empty string). if empty string then it will pick the running execution of that workflow ID.

internal/internal_workflow_client.go

@@ -216,13 +216,13 @@ func (wc *workflowClient) StartWorkflow(
 	return executionInfo, nil
 }
 
-// ExecuteWorkflow starts a workflow execution and wait until this workflow reaches the end state, such as
-// workflow finished successfully or timeout.
+// ExecuteWorkflow starts a workflow execution and returns a WorkflowRun that will allow you to wait until this workflow
+// reaches the end state, such as workflow finished successfully or timeout.
 // The user can use this to start using a functor like below and get the workflow execution result, as encoded.Value
 // Either by
-//     RunWorkflow(options, "workflowTypeName", arg1, arg2, arg3)
+//     ExecuteWorkflow(options, "workflowTypeName", arg1, arg2, arg3)
 //     or
-//     RunWorkflow(options, workflowExecuteFn, arg1, arg2, arg3)
+//     ExecuteWorkflow(options, workflowExecuteFn, arg1, arg2, arg3)
 // The current timeout resolution implementation is in seconds and uses math.Ceil(d.Seconds()) as the duration. But is
 // subjected to change in the future.
 // NOTE: the context.Context should have a fairly large timeout, since workflow execution may take a while to be finished
@@ -260,6 +260,25 @@ func (wc *workflowClient) ExecuteWorkflow(ctx context.Context, options StartWork
 	}, nil
 }
 
+// GetWorkflow gets a workflow execution and returns a WorkflowRun that will allow you to wait until this workflow
+// reaches the end state, such as workflow finished successfully or timeout.
+// The current timeout resolution implementation is in seconds and uses math.Ceil(d.Seconds()) as the duration. But is
+// subjected to change in the future.
+func (wc *workflowClient) GetWorkflow(ctx context.Context, workflowID string, runID string) WorkflowRun {
+
+	iterFn := func(fnCtx context.Context, fnRunID string) HistoryEventIterator {
+		return wc.GetWorkflowHistory(fnCtx, workflowID, fnRunID, true, s.HistoryEventFilterTypeCloseEvent)
+	}
+
+	return &workflowRunImpl{
+		workflowID:    workflowID,
+		firstRunID:    runID,
+		currentRunID:  runID,
+		iterFn:        iterFn,
+		dataConverter: wc.dataConverter,
+	}
+}
+
 // SignalWorkflow signals a workflow in execution.
 func (wc *workflowClient) SignalWorkflow(ctx context.Context, workflowID string, runID string, signalName string, arg interface{}) error {
 	input, err := encodeArg(wc.dataConverter, arg)

internal/internal_workflow_client_test.go

@@ -625,6 +625,43 @@ func (s *workflowRunSuite) TestExecuteWorkflow_NoDup_ContinueAsNew() {
 	s.Equal(workflowResult, decodedResult)
 }
 
+func (s *workflowRunSuite) TestGetWorkflow() {
+	filterType := shared.HistoryEventFilterTypeCloseEvent
+	eventType := shared.EventTypeWorkflowExecutionCompleted
+	workflowResult := time.Hour * 59
+	encodedResult, _ := encodeArg(getDefaultDataConverter(), workflowResult)
+	getRequest := getGetWorkflowExecutionHistoryRequest(filterType)
+	getResponse := &shared.GetWorkflowExecutionHistoryResponse{
+		History: &shared.History{
+			Events: []*shared.HistoryEvent{
+				&shared.HistoryEvent{
+					EventType: &eventType,
+					WorkflowExecutionCompletedEventAttributes: &shared.WorkflowExecutionCompletedEventAttributes{
+						Result: encodedResult,
+					},
+				},
+			},
+		},
+		NextPageToken: nil,
+	}
+	s.workflowServiceClient.EXPECT().GetWorkflowExecutionHistory(gomock.Any(), getRequest, gomock.Any(), gomock.Any(), gomock.Any()).Return(getResponse, nil).Times(1)
+
+	workflowID := workflowID
+	runID := runID
+
+	workflowRun := s.workflowClient.GetWorkflow(
+		context.Background(),
+		workflowID,
+		runID,
+	)
+	s.Equal(workflowRun.GetID(), workflowID)
+	s.Equal(workflowRun.GetRunID(), runID)
+	decodedResult := time.Minute
+	err := workflowRun.Get(context.Background(), &decodedResult)
+	s.Nil(err)
+	s.Equal(workflowResult, decodedResult)
+}
+
 func getGetWorkflowExecutionHistoryRequest(filterType shared.HistoryEventFilterType) *shared.GetWorkflowExecutionHistoryRequest {
 	isLongPoll := true