@@ -6,17 +6,14 @@ import (
66 "log"
77 "time"
88
9- "github.com/dapr/durabletask-go/api"
10- "github.com/dapr/durabletask-go/backend"
11- "github.com/dapr/durabletask-go/client"
12- "github.com/dapr/durabletask-go/task"
9+ "github.com/dapr/durabletask-go/workflow"
1310 dapr "github.com/dapr/go-sdk/client"
1411)
1512
1613func main () {
17- registry := task .NewTaskRegistry ()
14+ registry := workflow .NewTaskRegistry ()
1815
19- if err := registry .AddOrchestrator (BatchProcessingWorkflow ); err != nil {
16+ if err := registry .AddWorkflow (BatchProcessingWorkflow ); err != nil {
2017 log .Fatalf ("failed to register workflow: %v" , err )
2118 }
2219 if err := registry .AddActivity (GetWorkBatch ); err != nil {
@@ -37,70 +34,70 @@ func main() {
3734 log .Fatalf ("failed to create Dapr client: %v" , err )
3835 }
3936
40- client := client . NewTaskHubGrpcClient (daprClient .GrpcClientConn (), backend . DefaultLogger ())
41- if err := client . StartWorkItemListener (ctx , registry ); err != nil {
37+ wfclient := workflow . NewClient (daprClient .GrpcClientConn ())
38+ if err := wfclient . StartWorker (ctx , registry ); err != nil {
4239 log .Fatalf ("failed to start work item listener: %v" , err )
4340 }
4441
45- id , err := client . ScheduleNewOrchestration (ctx , "BatchProcessingWorkflow" , api .WithInput (10 ))
42+ id , err := wfclient . ScheduleNewWorkflow (ctx , "BatchProcessingWorkflow" , workflow .WithInput (10 ))
4643 if err != nil {
4744 log .Fatalf ("failed to schedule a new workflow: %v" , err )
4845 }
4946
50- metadata , err := client . WaitForOrchestrationCompletion (ctx , id )
47+ metadata , err := wfclient . WaitForWorkflowCompletion (ctx , id )
5148 if err != nil {
5249 log .Fatalf ("failed to get workflow: %v" , err )
5350 }
5451 fmt .Printf ("workflow status: %s\n " , metadata .RuntimeStatus .String ())
5552
56- err = client . TerminateOrchestration (ctx , id )
53+ err = wfclient . TerminateWorkflow (ctx , id )
5754 if err != nil {
5855 log .Fatalf ("failed to terminate workflow: %v" , err )
5956 }
6057 fmt .Println ("workflow terminated" )
6158
62- err = client . PurgeOrchestrationState (ctx , id )
59+ err = wfclient . PurgeWorkflowState (ctx , id )
6360 if err != nil {
6461 log .Fatalf ("failed to purge workflow: %v" , err )
6562 }
6663 fmt .Println ("workflow purged" )
6764}
6865
69- func BatchProcessingWorkflow (ctx * task. OrchestrationContext ) (any , error ) {
66+ func BatchProcessingWorkflow (ctx * workflow. WorkflowContext ) (any , error ) {
7067 var input int
7168 if err := ctx .GetInput (& input ); err != nil {
7269 return 0 , err
7370 }
7471
7572 var workBatch []int
76- if err := ctx .CallActivity (GetWorkBatch , task .WithActivityInput (input )).Await (& workBatch ); err != nil {
73+ if err := ctx .CallActivity (GetWorkBatch , workflow .WithActivityInput (input )).Await (& workBatch ); err != nil {
7774 return 0 , err
7875 }
7976
80- parallelTasks := make ([]task .Task , len (workBatch ))
77+ parallelTasks := make ([]workflow .Task , len (workBatch ))
8178 for i , workItem := range workBatch {
82- parallelTasks [i ] = ctx .CallActivity (ProcessWorkItem , task .WithActivityInput (workItem ))
79+ parallelTasks [i ] = ctx .CallActivity (ProcessWorkItem , workflow .WithActivityInput (workItem ))
8380 }
8481
8582 var outputs int
86- for _ , task := range parallelTasks {
83+ for _ , workflow := range parallelTasks {
8784 var output int
88- err := task .Await (& output )
85+ err := workflow .Await (& output )
8986 if err == nil {
9087 outputs += output
9188 } else {
9289 return 0 , err
9390 }
9491 }
9592
96- if err := ctx .CallActivity (ProcessResults , task .WithActivityInput (outputs )).Await (nil ); err != nil {
93+ if err := ctx .CallActivity (ProcessResults , workflow .WithActivityInput (outputs )).Await (nil ); err != nil {
9794 return 0 , err
9895 }
9996
10097 return 0 , nil
10198}
10299
103- func GetWorkBatch (ctx task .ActivityContext ) (any , error ) {
100+ func GetWorkBatch (ctx workflow .ActivityContext ) (any , error ) {
104101 var batchSize int
105102 if err := ctx .GetInput (& batchSize ); err != nil {
106103 return 0 , err
@@ -112,7 +109,7 @@ func GetWorkBatch(ctx task.ActivityContext) (any, error) {
112109 return batch , nil
113110}
114111
115- func ProcessWorkItem (ctx task .ActivityContext ) (any , error ) {
112+ func ProcessWorkItem (ctx workflow .ActivityContext ) (any , error ) {
116113 var workItem int
117114 if err := ctx .GetInput (& workItem ); err != nil {
118115 return 0 , err
@@ -124,7 +121,7 @@ func ProcessWorkItem(ctx task.ActivityContext) (any, error) {
124121 return result , nil
125122}
126123
127- func ProcessResults (ctx task .ActivityContext ) (any , error ) {
124+ func ProcessResults (ctx workflow .ActivityContext ) (any , error ) {
128125 var finalResult int
129126 if err := ctx .GetInput (& finalResult ); err != nil {
130127 return 0 , err
0 commit comments