-
Notifications
You must be signed in to change notification settings - Fork 171
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Signed-off-by: mikeee <hey@mike.ee>
- Loading branch information
Showing
6 changed files
with
251 additions
and
0 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
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
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,13 @@ | ||
| package workflow | ||
|
|
||
| import ( | ||
| "github.com/microsoft/durabletask-go/task" | ||
| ) | ||
|
|
||
| type ActivityContext struct { | ||
| ctx task.ActivityContext | ||
| } | ||
|
|
||
| func (wfac *ActivityContext) GetInput(v interface{}) error { | ||
| return wfac.ctx.GetInput(&v) | ||
| } |
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,60 @@ | ||
| package workflow | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "log" | ||
| "time" | ||
|
|
||
| "github.com/microsoft/durabletask-go/task" | ||
| ) | ||
|
|
||
| type Context struct { | ||
| orchestrationContext *task.OrchestrationContext | ||
| } | ||
|
|
||
| func (wfc *Context) GetInput(v interface{}) error { | ||
| return wfc.orchestrationContext.GetInput(&v) | ||
| } | ||
|
|
||
| func (wfc *Context) Name() string { | ||
| return wfc.orchestrationContext.Name | ||
| } | ||
|
|
||
| func (wfc *Context) InstanceID() string { | ||
| return fmt.Sprintf("%v", wfc.orchestrationContext.ID) | ||
| } | ||
|
|
||
| func (wfc *Context) CurrentUTCDateTime() time.Time { | ||
| return wfc.orchestrationContext.CurrentTimeUtc | ||
| } | ||
|
|
||
| func (wfc *Context) IsReplaying() bool { | ||
| return wfc.orchestrationContext.IsReplaying | ||
| } | ||
|
|
||
| func (wfc *Context) CallActivity(activity interface{}) task.Task { | ||
| var inp string | ||
| if err := wfc.GetInput(&inp); err != nil { | ||
| log.Printf("unable to get activity input: %v", err) | ||
| } | ||
| // the call should continue despite being unable to obtain an input | ||
|
|
||
| return wfc.orchestrationContext.CallActivity(activity, task.WithActivityInput(inp)) | ||
| } | ||
|
|
||
| func (wfc *Context) CallChildWorkflow() { | ||
| // TODO: implement | ||
| // call suborchestrator | ||
| } | ||
|
|
||
| func (wfc *Context) CreateTimer() { | ||
| // TODO: implement | ||
| } | ||
|
|
||
| func (wfc *Context) WaitForExternalEvent() { | ||
| // TODO: implement | ||
| } | ||
|
|
||
| func (wfc *Context) ContinueAsNew() { | ||
| // TODO: implement | ||
| } |
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,131 @@ | ||
| package workflow | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "fmt" | ||
| "log" | ||
| "reflect" | ||
| "runtime" | ||
| "strings" | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/microsoft/durabletask-go/backend" | ||
| "github.com/microsoft/durabletask-go/client" | ||
| "github.com/microsoft/durabletask-go/task" | ||
| "google.golang.org/grpc" | ||
| "google.golang.org/grpc/credentials/insecure" | ||
| ) | ||
|
|
||
| type WorkflowRuntime struct { | ||
| tasks *task.TaskRegistry | ||
| client *client.TaskHubGrpcClient | ||
|
|
||
| mutex sync.Mutex // TODO: implement | ||
| quit chan bool | ||
| cancel context.CancelFunc | ||
| } | ||
|
|
||
| type Workflow func(ctx *Context) (any, error) | ||
|
|
||
| type Activity func(ctx ActivityContext) (any, error) | ||
|
|
||
| func NewRuntime(host string, port string) (*WorkflowRuntime, error) { | ||
| ctx, canc := context.WithTimeout(context.Background(), time.Second*10) | ||
| defer canc() | ||
|
|
||
| address := fmt.Sprintf("%s:%s", host, port) | ||
|
|
||
| clientConn, err := grpc.DialContext( | ||
| ctx, | ||
| address, | ||
| grpc.WithTransportCredentials(insecure.NewCredentials()), | ||
| grpc.WithBlock(), // TODO: config | ||
| ) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("failed to create runtime - grpc connection failed: %v", err) | ||
| } | ||
|
|
||
| return &WorkflowRuntime{ | ||
| tasks: task.NewTaskRegistry(), | ||
| client: client.NewTaskHubGrpcClient(clientConn, backend.DefaultLogger()), | ||
| quit: make(chan bool), | ||
| cancel: canc, | ||
| }, nil | ||
| } | ||
|
|
||
| func getDecorator(f interface{}) (string, error) { | ||
| if f == nil { | ||
| return "", errors.New("nil function name") | ||
| } | ||
|
|
||
| callSplit := strings.Split(runtime.FuncForPC(reflect.ValueOf(f).Pointer()).Name(), ".") | ||
|
|
||
| funcName := callSplit[len(callSplit)-1] | ||
|
|
||
| return funcName, nil | ||
| } | ||
|
|
||
| func (wr *WorkflowRuntime) RegisterWorkflow(w Workflow) error { | ||
| wrappedOrchestration := func(ctx *task.OrchestrationContext) (any, error) { | ||
| wfCtx := &Context{orchestrationContext: ctx} | ||
|
|
||
| return w(wfCtx) | ||
| } | ||
|
|
||
| // getdecorator for workflow | ||
| name, err := getDecorator(w) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to get workflow decorator: %v", err) | ||
| } | ||
|
|
||
| err = wr.tasks.AddOrchestratorN(name, wrappedOrchestration) | ||
| return err | ||
| } | ||
|
|
||
| func (wr *WorkflowRuntime) RegisterActivity(a Activity) error { | ||
| wrappedActivity := func(ctx task.ActivityContext) (any, error) { | ||
| ac := ActivityContext{ctx: ctx} | ||
|
|
||
| return a(ac) | ||
| } | ||
|
|
||
| // getdecorator for activity | ||
| name, err := getDecorator(a) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to get activity decorator: %v", err) | ||
| } | ||
|
|
||
| err = wr.tasks.AddActivityN(name, wrappedActivity) | ||
| return err | ||
| } | ||
|
|
||
| func (wr *WorkflowRuntime) Start() error { | ||
| // go func start | ||
| go func() { | ||
| err := wr.client.StartWorkItemListener(context.Background(), wr.tasks) | ||
| if err != nil { | ||
| log.Fatalf("failed to start work stream: %v", err) | ||
| } | ||
| for { | ||
| select { | ||
| case <-wr.quit: | ||
| return | ||
| default: | ||
| // continue serving | ||
| } | ||
| } | ||
| }() | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func (wr *WorkflowRuntime) Shutdown() error { | ||
| // cancel grpc context | ||
| wr.cancel() | ||
| // send close signal | ||
| wr.quit <- true | ||
|
|
||
| return 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,28 @@ | ||
| package workflow | ||
|
|
||
| import ( | ||
| "testing" | ||
|
|
||
| "github.com/stretchr/testify/assert" | ||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| func TestWorkflowRuntime(t *testing.T) { | ||
| // TODO: Mock grpc conn - currently requires dapr to be available | ||
| t.Run("test workflow name is correct", func(t *testing.T) { | ||
| wr, err := NewRuntime("localhost", "50001") | ||
| require.NoError(t, err) | ||
| err = wr.RegisterWorkflow(testOrchestrator) | ||
| require.NoError(t, err) | ||
| }) | ||
| } | ||
|
|
||
| func TestGetDecorator(t *testing.T) { | ||
| name, err := getDecorator(testOrchestrator) | ||
| require.NoError(t, err) | ||
| assert.Equal(t, "testOrchestrator", name) | ||
| } | ||
|
|
||
| func testOrchestrator(ctx *Context) (any, error) { | ||
| return nil, nil | ||
| } |