Merge pull request #48 from JoshVanL/workflow-vanity-client

Create vanity `workflow` client

Author: cicoyle

api/orchestration.go

@@ -80,6 +80,7 @@ func WithInstanceID(id InstanceID) NewOrchestrationOptions {
 }
 
 // WithOrchestrationIdReusePolicy configures Orchestration ID reuse policy.
+// Deprecated.
 func WithOrchestrationIdReusePolicy(policy *protos.OrchestrationIdReusePolicy) NewOrchestrationOptions {
 	return func(req *protos.CreateInstanceRequest) error {
 		// initialize CreateInstanceOption

task/activity.go

@@ -11,7 +11,7 @@ import (
 	"github.com/dapr/durabletask-go/api/protos"
 )
 
-type callActivityOption func(*callActivityOptions) error
+type CallActivityOption func(*callActivityOptions) error
 
 type callActivityOptions struct {
 	rawInput    *wrapperspb.StringValue
@@ -59,7 +59,7 @@ func (policy *RetryPolicy) Validate() error {
 	return nil
 }
 
-func WithActivityAppID(targetAppID string) callActivityOption {
+func WithActivityAppID(targetAppID string) CallActivityOption {
 	return func(opt *callActivityOptions) error {
 		opt.targetAppID = &targetAppID
 		return nil
@@ -68,7 +68,7 @@ func WithActivityAppID(targetAppID string) callActivityOption {
 
 // WithActivityInput configures an input for an activity invocation.
 // The specified input must be JSON serializable.
-func WithActivityInput(input any) callActivityOption {
+func WithActivityInput(input any) CallActivityOption {
 	return func(opt *callActivityOptions) error {
 		data, err := marshalData(input)
 		if err != nil {
@@ -80,14 +80,14 @@ func WithActivityInput(input any) callActivityOption {
 }
 
 // WithRawActivityInput configures a raw input for an activity invocation.
-func WithRawActivityInput(input *wrapperspb.StringValue) callActivityOption {
+func WithRawActivityInput(input *wrapperspb.StringValue) CallActivityOption {
 	return func(opt *callActivityOptions) error {
 		opt.rawInput = input
 		return nil
 	}
 }
 
-func WithActivityRetryPolicy(policy *RetryPolicy) callActivityOption {
+func WithActivityRetryPolicy(policy *RetryPolicy) CallActivityOption {
 	return func(opt *callActivityOptions) error {
 		if policy == nil {
 			return nil

task/orchestrator.go

@@ -58,14 +58,14 @@ type callSubOrchestratorOptions struct {
 	retryPolicy *RetryPolicy
 }
 
-// subOrchestratorOption is a functional option type for the CallSubOrchestrator orchestrator method.
-type subOrchestratorOption func(*callSubOrchestratorOptions) error
+// SubOrchestratorOption is a functional option type for the CallSubOrchestrator orchestrator method.
+type SubOrchestratorOption func(*callSubOrchestratorOptions) error
 
 // ContinueAsNewOption is a functional option type for the ContinueAsNew orchestrator method.
 type ContinueAsNewOption func(*OrchestrationContext)
 
 // WithSubOrchestratorAppID is a functional option type for the CallSubOrchestrator orchestrator method that specifies the app ID of the target activity.
-func WithSubOrchestratorAppID(appID string) subOrchestratorOption {
+func WithSubOrchestratorAppID(appID string) SubOrchestratorOption {
 	return func(opts *callSubOrchestratorOptions) error {
 		opts.targetAppID = &appID
 		return nil
@@ -82,7 +82,7 @@ func WithKeepUnprocessedEvents() ContinueAsNewOption {
 
 // WithSubOrchestratorInput is a functional option type for the CallSubOrchestrator
 // orchestrator method that takes an input value and marshals it to JSON.
-func WithSubOrchestratorInput(input any) subOrchestratorOption {
+func WithSubOrchestratorInput(input any) SubOrchestratorOption {
 	return func(opts *callSubOrchestratorOptions) error {
 		bytes, err := marshalData(input)
 		if err != nil {
@@ -95,7 +95,7 @@ func WithSubOrchestratorInput(input any) subOrchestratorOption {
 
 // WithRawSubOrchestratorInput is a functional option type for the CallSubOrchestrator
 // orchestrator method that takes a raw input value.
-func WithRawSubOrchestratorInput(input *wrapperspb.StringValue) subOrchestratorOption {
+func WithRawSubOrchestratorInput(input *wrapperspb.StringValue) SubOrchestratorOption {
 	return func(opts *callSubOrchestratorOptions) error {
 		opts.rawInput = input
 		return nil
@@ -104,14 +104,14 @@ func WithRawSubOrchestratorInput(input *wrapperspb.StringValue) subOrchestratorO
 
 // WithSubOrchestrationInstanceID is a functional option type for the CallSubOrchestrator
 // orchestrator method that specifies the instance ID of the sub-orchestration.
-func WithSubOrchestrationInstanceID(instanceID string) subOrchestratorOption {
+func WithSubOrchestrationInstanceID(instanceID string) SubOrchestratorOption {
 	return func(opts *callSubOrchestratorOptions) error {
 		opts.instanceID = instanceID
 		return nil
 	}
 }
 
-func WithSubOrchestrationRetryPolicy(policy *RetryPolicy) subOrchestratorOption {
+func WithSubOrchestrationRetryPolicy(policy *RetryPolicy) SubOrchestratorOption {
 	return func(opt *callSubOrchestratorOptions) error {
 		if policy == nil {
 			return nil
@@ -267,7 +267,7 @@ func (octx *OrchestrationContext) GetInput(v any) error {
 // CallActivity schedules an asynchronous invocation of an activity function. The [activity]
 // parameter can be either the name of an activity as a string or can be a pointer to the function
 // that implements the activity, in which case the name is obtained via reflection.
-func (ctx *OrchestrationContext) CallActivity(activity interface{}, opts ...callActivityOption) Task {
+func (ctx *OrchestrationContext) CallActivity(activity interface{}, opts ...CallActivityOption) Task {
 	options := new(callActivityOptions)
 	for _, configure := range opts {
 		if err := configure(options); err != nil {
@@ -317,8 +317,7 @@ func (ctx *OrchestrationContext) internalScheduleActivity(activityName, taskExec
 	return task
 }
 
-// TODO: cassie wire appID into suborchestration options too for cross app wf
-func (ctx *OrchestrationContext) CallSubOrchestrator(orchestrator interface{}, opts ...subOrchestratorOption) Task {
+func (ctx *OrchestrationContext) CallSubOrchestrator(orchestrator interface{}, opts ...SubOrchestratorOption) Task {
 	options := new(callSubOrchestratorOptions)
 	for _, configure := range opts {
 		if err := configure(options); err != nil {
@@ -421,7 +420,7 @@ func computeNextDelay(currentTimeUtc time.Time, policy RetryPolicy, attempt int,
 }
 
 // CreateTimer schedules a durable timer that expires after the specified delay.
-func (ctx *OrchestrationContext) CreateTimer(delay time.Duration, opts ...createTimerOption) Task {
+func (ctx *OrchestrationContext) CreateTimer(delay time.Duration, opts ...CreateTimerOption) Task {
 	options := new(createTimerOptions)
 	for _, configure := range opts {
 		if err := configure(options); err != nil {

task/timer.go

@@ -1,12 +1,12 @@
 package task
 
-type createTimerOption func(*createTimerOptions) error
+type CreateTimerOption func(*createTimerOptions) error
 
 type createTimerOptions struct {
 	name *string
 }
 
-func WithTimerName(name string) createTimerOption {
+func WithTimerName(name string) CreateTimerOption {
 	return func(opt *createTimerOptions) error {
 		opt.name = &name
 		return nil

workflow/activity.go

@@ -0,0 +1,50 @@
+package workflow
+
+import (
+	"time"
+
+	"github.com/dapr/durabletask-go/task"
+	"google.golang.org/protobuf/types/known/wrapperspb"
+)
+
+type CallActivityOption task.CallActivityOption
+
+type RetryPolicy struct {
+	// Max number of attempts to try the activity call, first execution inclusive
+	MaxAttempts int
+	// Timespan to wait for the first retry
+	InitialRetryInterval time.Duration
+	// Used to determine rate of increase of back-off
+	BackoffCoefficient float64
+	// Max timespan to wait for a retry
+	MaxRetryInterval time.Duration
+	// Total timeout across all the retries performed
+	RetryTimeout time.Duration
+	// Optional function to control if retries should proceed
+	Handle func(error) bool
+}
+
+func WithActivityAppID(targetAppID string) CallActivityOption {
+	return CallActivityOption(task.WithActivityAppID(targetAppID))
+}
+
+// WithActivityInput configures an input for an activity invocation. The
+// specified input must be JSON serializable.
+func WithActivityInput(input any) CallActivityOption {
+	return CallActivityOption(task.WithActivityInput(input))
+}
+
+// WithRawActivityInput configures a raw input for an activity invocation.
+func WithRawActivityInput(input *wrapperspb.StringValue) CallActivityOption {
+	return CallActivityOption(task.WithRawActivityInput(input))
+}
+
+func WithActivityRetryPolicy(policy *RetryPolicy) CallActivityOption {
+	return CallActivityOption(task.WithActivityRetryPolicy((*task.RetryPolicy)(policy)))
+}
+
+// ActivityContext is the context parameter type for activity implementations.
+type ActivityContext task.ActivityContext
+
+// Activity is the functional interface for activity implementations.
+type Activity func(ActivityContext) (any, error)

workflow/api.go

@@ -0,0 +1,101 @@
+package workflow
+
+import (
+	"time"
+
+	"github.com/dapr/durabletask-go/api"
+	"github.com/dapr/durabletask-go/api/protos"
+	"github.com/dapr/kit/ptr"
+	"google.golang.org/protobuf/types/known/wrapperspb"
+)
+
+type NewWorkflowOptions api.NewOrchestrationOptions
+type FetchWorkflowMetadataOptions api.FetchOrchestrationMetadataOptions
+type RaiseEventOptions api.RaiseEventOptions
+type TerminateOptions api.TerminateOptions
+type PurgeOptions api.PurgeOptions
+type RerunOptions api.RerunOptions
+
+// WithInstanceID configures an explicit workflow instance ID. If not
+// specified, a random UUID value will be used for the workflow instance ID.
+func WithInstanceID(id string) NewWorkflowOptions {
+	return NewWorkflowOptions(api.WithInstanceID(api.InstanceID(id)))
+}
+
+// WithInput configures an input for the workflow. The specified input must be
+// serializable.
+func WithInput(input any) NewWorkflowOptions {
+	return NewWorkflowOptions(api.WithInput(input))
+}
+
+// WithRawInput configures an input for the workflow. The specified input must
+// be a string.
+func WithRawInput(rawInput *wrapperspb.StringValue) NewWorkflowOptions {
+	return NewWorkflowOptions(api.WithRawInput(rawInput))
+}
+
+// WithStartTime configures a start time at which the workflow should start
+// running. Note that the actual start time could be later than the specified
+// start time if the task hub is under load or if the app is not running at the
+// specified start time.
+func WithStartTime(startTime time.Time) NewWorkflowOptions {
+	return NewWorkflowOptions(api.WithStartTime(startTime))
+}
+
+// WithFetchPayloads configures whether to load workflow inputs, outputs, and
+// custom status values, which could be large.
+func WithFetchPayloads(fetchPayloads bool) FetchWorkflowMetadataOptions {
+	return FetchWorkflowMetadataOptions(api.WithFetchPayloads(fetchPayloads))
+}
+
+// WithEventPayload configures an event payload. The specified payload must be
+// serializable.
+func WithEventPayload(data any) RaiseEventOptions {
+	return RaiseEventOptions(api.WithEventPayload(data))
+}
+
+// WithRawEventData configures an event payload that is a raw, unprocessed
+// string (e.g. JSON data).
+func WithRawEventData(data *wrapperspb.StringValue) RaiseEventOptions {
+	return RaiseEventOptions(api.WithRawEventData(data))
+}
+
+// WithOutput configures an output for the terminated workflow. The specified
+// output must be serializable.
+func WithOutput(data any) TerminateOptions {
+	return TerminateOptions(api.WithOutput(data))
+}
+
+// WithRawOutput configures a raw, unprocessed output (i.e. pre-serialized) for
+// the terminated workflow.
+func WithRawOutput(data *wrapperspb.StringValue) TerminateOptions {
+	return TerminateOptions(api.WithRawOutput(data))
+}
+
+// WithRecursiveTerminate configures whether to terminate all child-workflows
+// created by the target workflow.
+func WithRecursiveTerminate(recursive bool) TerminateOptions {
+	return TerminateOptions(api.WithRecursiveTerminate(recursive))
+}
+
+// WithRecursivePurge configures whether to purge all child-workflows created
+// by the target workflow.
+func WithRecursivePurge(recursive bool) PurgeOptions {
+	return PurgeOptions(api.WithRecursivePurge(recursive))
+}
+
+func WorkflowMetadataIsRunning(o *WorkflowMetadata) bool {
+	return api.OrchestrationMetadataIsComplete(ptr.Of(protos.OrchestrationMetadata(*o)))
+}
+
+func WorkflowMetadataIsComplete(o *WorkflowMetadata) bool {
+	return api.OrchestrationMetadataIsComplete(ptr.Of(protos.OrchestrationMetadata(*o)))
+}
+
+func WithRerunInput(input any) RerunOptions {
+	return RerunOptions(api.WithRerunInput(input))
+}
+
+func WithRerunNewInstanceID(id string) RerunOptions {
+	return RerunOptions(api.WithRerunNewInstanceID(api.InstanceID(id)))
+}