-
Notifications
You must be signed in to change notification settings - Fork 201
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
f25b9cb
commit d20529b
Showing
7 changed files
with
339 additions
and
42 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,16 @@ | ||
| ### Async Update Sample | ||
|
|
||
| Here we show an example of a workflow representing a parallel job processor. The workflow accepts | ||
| jobs through update requests, allowing up to five parallel jobs, and uses the update validator to reject any | ||
| jobs over the limit. The workflow also demonstrates how to properly drain updates so all updates are processed before completing a workflow. | ||
|
|
||
| ### Steps to run this sample: | ||
| 1) Run a [Temporal service](https://github.com/temporalio/samples-go/tree/main/#how-to-use). | ||
| 2) Run the following command to start the worker | ||
| ``` | ||
| go run async-update/worker/main.go | ||
| ``` | ||
| 3) Run the following command to start the example | ||
| ``` | ||
| go run async-update/starter/main.go | ||
| ``` |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,94 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "context" | ||
| "log" | ||
|
|
||
| "github.com/google/uuid" | ||
| async_update "github.com/temporalio/samples-go/async-update" | ||
| enumspb "go.temporal.io/api/enums/v1" | ||
| updatepb "go.temporal.io/api/update/v1" | ||
|
|
||
| "github.com/temporalio/samples-go/update" | ||
| "go.temporal.io/sdk/client" | ||
| ) | ||
|
|
||
| func main() { | ||
| c, err := client.Dial(client.Options{}) | ||
| if err != nil { | ||
| log.Fatalln("Unable to create client", err) | ||
| } | ||
| defer c.Close() | ||
|
|
||
| workflowOptions := client.StartWorkflowOptions{ | ||
| ID: "async-update-workflow-ID", | ||
| TaskQueue: "async-update", | ||
| } | ||
|
|
||
| we, err := c.ExecuteWorkflow(context.Background(), workflowOptions, async_update.ProcessWorkflow) | ||
| if err != nil { | ||
| log.Fatalln("Unable to execute workflow", err) | ||
| } | ||
|
|
||
| log.Println("Started workflow", "WorkflowID", we.GetID(), "RunID", we.GetRunID()) | ||
|
|
||
| // Send multiple updates to the workflow | ||
| var updates []client.WorkflowUpdateHandle | ||
| // ProcessWorkflow only allows 5 in progress jobs at a time, so we send 6 updates to test the validator. | ||
| for i := 0; i < 6; i++ { | ||
| updateID := uuid.New().String() | ||
| log.Println("Sending workflow update", "WorkflowID", we.GetID(), "RunID", we.GetRunID(), "UpdateID", updateID) | ||
| handle, err := c.UpdateWorkflowWithOptions(context.Background(), &client.UpdateWorkflowWithOptionsRequest{ | ||
| WorkflowID: we.GetID(), | ||
| RunID: we.GetRunID(), | ||
| UpdateID: updateID, | ||
| UpdateName: async_update.ProcessUpdateName, | ||
| Args: []interface{}{"world"}, | ||
| // WaitPolicy is a hint to return early if the update reaches a certain stage. | ||
| // By default the SDK will wait until the update is processed or the server sends back | ||
| // an empty response then the SDK can poll the update result later. | ||
| // Useful for short updates that can be completed with a single RPC. | ||
| WaitPolicy: &updatepb.WaitPolicy{ | ||
| // LifecycleStage UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_ACCEPTED means to wait until the update is accepted | ||
| // or the Temporal server returns an empty response. | ||
| LifecycleStage: enumspb.UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_ACCEPTED, | ||
| }, | ||
| }) | ||
| if err != nil { | ||
| log.Fatalln("Unable to send update request", err) | ||
| } | ||
| updates = append(updates, handle) | ||
| } | ||
| for _, handle := range updates { | ||
| var updateOutput string | ||
| err = handle.Get(context.Background(), &updateOutput) | ||
| if err != nil { | ||
| log.Println("Update failed with error", err) | ||
| } else { | ||
| log.Println("Update result", "WorkflowID", we.GetID(), "RunID", we.GetRunID(), "UpdateID", handle.UpdateID(), "Result", updateOutput) | ||
| } | ||
| } | ||
| // You can also create a handle for a previously sent update using the update's ID. | ||
| newHandle := c.GetWorkflowUpdateHandle(client.GetWorkflowUpdateHandleOptions{ | ||
| WorkflowID: we.GetID(), | ||
| RunID: we.GetRunID(), | ||
| UpdateID: updates[0].UpdateID(), | ||
| }) | ||
| var updateOutput string | ||
| err = newHandle.Get(context.Background(), &updateOutput) | ||
| if err != nil { | ||
| log.Println("Get update result failed with error", err) | ||
| } else { | ||
| log.Println("Get update result", "WorkflowID", we.GetID(), "RunID", we.GetRunID(), "UpdateID", newHandle.UpdateID(), "Result", updateOutput) | ||
| } | ||
| // Signal the workflow to stop accepting new work. | ||
| if err = c.SignalWorkflow(context.Background(), we.GetID(), we.GetRunID(), update.Done, nil); err != nil { | ||
| log.Fatalf("failed to send %q signal to workflow: %v", update.Done, err) | ||
| } | ||
| // Get the result of the workflow this will block until all the updates are processed. | ||
| var wfResult int | ||
| if err = we.Get(context.Background(), &wfResult); err != nil { | ||
| log.Fatalf("unable get workflow result: %v", err) | ||
| } | ||
| log.Println("Updates processed:", wfResult) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,28 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "log" | ||
|
|
||
| async_update "github.com/temporalio/samples-go/async-update" | ||
| "go.temporal.io/sdk/client" | ||
| "go.temporal.io/sdk/worker" | ||
| ) | ||
|
|
||
| func main() { | ||
| // The client and worker are heavyweight objects that should be created once per process. | ||
| c, err := client.Dial(client.Options{}) | ||
| if err != nil { | ||
| log.Fatalln("Unable to create client", err) | ||
| } | ||
| defer c.Close() | ||
|
|
||
| w := worker.New(c, "async-update", worker.Options{}) | ||
|
|
||
| w.RegisterWorkflow(async_update.ProcessWorkflow) | ||
| w.RegisterActivity(async_update.Activity) | ||
|
|
||
| err = w.Run(worker.InterruptCh()) | ||
| if err != nil { | ||
| log.Fatalln("Unable to start worker", err) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,75 @@ | ||
| package async_update | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "time" | ||
|
|
||
| "go.temporal.io/sdk/workflow" | ||
| ) | ||
|
|
||
| const ( | ||
| ProcessUpdateName = "process" | ||
| Done = "done" | ||
| ) | ||
|
|
||
| func ProcessWorkflow(ctx workflow.Context) (int, error) { | ||
| logger := workflow.GetLogger(ctx) | ||
| // inProgressJobs is used to keep track of the number of jobs currently being processed. | ||
| inProgressJobs := 0 | ||
| // processedJobs is used to keep track of the number of jobs processed so far. | ||
| processedJobs := 0 | ||
| // closing is used to keep track of whether the workflow is closing. | ||
| closing := false | ||
|
|
||
| if err := workflow.SetUpdateHandlerWithOptions( | ||
| ctx, | ||
| ProcessUpdateName, | ||
| func(ctx workflow.Context, s string) (string, error) { | ||
| inProgressJobs++ | ||
| processedJobs++ | ||
| defer func() { | ||
| inProgressJobs-- | ||
| }() | ||
| logger.Debug("Processing job", "job", s) | ||
| ao := workflow.ActivityOptions{ | ||
| StartToCloseTimeout: 10 * time.Second, | ||
| } | ||
| ctx = workflow.WithActivityOptions(ctx, ao) | ||
| var result string | ||
| err := workflow.ExecuteActivity(ctx, Activity, s).Get(ctx, &result) | ||
| logger.Debug("Processed job", "job", s) | ||
| return result, err | ||
| }, | ||
| workflow.UpdateHandlerOptions{ | ||
| Validator: func(s string) error { | ||
| logger.Debug("Validating job", "job", s, "inProgressJobs", inProgressJobs) | ||
| if inProgressJobs >= 5 { | ||
| return fmt.Errorf("too many in progress jobs: %d", inProgressJobs) | ||
| } else if closing { | ||
| return fmt.Errorf("workflow is closing") | ||
| } | ||
| return nil | ||
| }, | ||
| }, | ||
| ); err != nil { | ||
| return 0, err | ||
| } | ||
|
|
||
| _ = workflow.GetSignalChannel(ctx, Done).Receive(ctx, nil) | ||
| logger.Debug("Closing workflow, draining in progress jobs", "inProgressJobs", inProgressJobs) | ||
| // set closing to true to indicate that the workflow is closing. | ||
| // no more new jobs are allowed, but the existing jobs will be processed. | ||
| closing = true | ||
| workflow.Await(ctx, func() bool { | ||
| return inProgressJobs == 0 | ||
| }) | ||
| logger.Debug("All jobs processed, workflow can now close") | ||
| return processedJobs, ctx.Err() | ||
| } | ||
|
|
||
| func Activity(ctx context.Context, name string) (string, error) { | ||
| // Simulate a long running activity | ||
| time.Sleep(5 * time.Second) | ||
| return "Hello " + name + "!", nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,87 @@ | ||
| package async_update_test | ||
|
|
||
| import ( | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/stretchr/testify/mock" | ||
| "github.com/stretchr/testify/require" | ||
| async_update "github.com/temporalio/samples-go/async-update" | ||
| "go.temporal.io/sdk/testsuite" | ||
| ) | ||
|
|
||
| type updateCallback struct { | ||
| accept func() | ||
| reject func(error) | ||
| complete func(interface{}, error) | ||
| } | ||
|
|
||
| func (uc *updateCallback) Accept() { | ||
| uc.accept() | ||
| } | ||
|
|
||
| func (uc *updateCallback) Reject(err error) { | ||
| uc.reject(err) | ||
| } | ||
|
|
||
| func (uc *updateCallback) Complete(success interface{}, err error) { | ||
| uc.complete(success, err) | ||
| } | ||
|
|
||
| func TestWorkflow(t *testing.T) { | ||
| // Create env | ||
| var suite testsuite.WorkflowTestSuite | ||
| env := suite.NewTestWorkflowEnvironment() | ||
| env.RegisterWorkflow(async_update.ProcessWorkflow) | ||
| env.OnActivity(async_update.Activity, mock.Anything, "world").After(5*time.Second).Return("hello world", nil) | ||
|
|
||
| // Use delayed callbacks to send multiple updates at the same time | ||
| for i := 0; i < 10; i++ { | ||
| i := i | ||
| env.RegisterDelayedCallback(func() { | ||
| env.UpdateWorkflow(async_update.ProcessUpdateName, "test id", &updateCallback{ | ||
| accept: func() { | ||
| if i >= 5 { | ||
| require.Fail(t, "update should fail since we should exceed our max update limit") | ||
| } | ||
| }, | ||
| reject: func(err error) { | ||
| if i < 5 { | ||
| require.Fail(t, "this update should not fail") | ||
| } | ||
| require.Error(t, err) | ||
| }, | ||
| complete: func(response interface{}, err error) { | ||
| require.NoError(t, err) | ||
| require.Equal(t, "hello world", response) | ||
| }, | ||
| }, "world") | ||
| }, 0) | ||
| } | ||
| // Use delayed callback to send signal | ||
| env.RegisterDelayedCallback(func() { | ||
| env.SignalWorkflow(async_update.Done, nil) | ||
| }, time.Second) | ||
|
|
||
| // Send an update after the workflow is signaled to close, expect the update to be rejected | ||
| env.RegisterDelayedCallback(func() { | ||
| env.UpdateWorkflow(async_update.ProcessUpdateName, "test id", &updateCallback{ | ||
| accept: func() { | ||
| require.Fail(t, "update should fail since the workflow is closing") | ||
| }, | ||
| reject: func(err error) { | ||
| require.Error(t, err) | ||
| }, | ||
| complete: func(response interface{}, err error) { | ||
| }, | ||
| }, "world") | ||
| }, 2*time.Second) | ||
|
|
||
| // Run workflow | ||
| env.ExecuteWorkflow(async_update.ProcessWorkflow) | ||
| require.True(t, env.IsWorkflowCompleted()) | ||
| require.NoError(t, env.GetWorkflowError()) | ||
| var result int | ||
| require.NoError(t, env.GetWorkflowResult(&result)) | ||
| require.Equal(t, 5, result) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.