-
Notifications
You must be signed in to change notification settings - Fork 158
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: manual triggers and give clients a hook into step run events (#141
) * feat: pubsub for clients, more qol stuff * fix: generate sqlc files * chore: linting and comments
- Loading branch information
1 parent
aed11c3
commit 82d7995
Showing
52 changed files
with
2,056 additions
and
425 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,74 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "time" | ||
|
|
||
| "github.com/joho/godotenv" | ||
|
|
||
| "github.com/hatchet-dev/hatchet/pkg/client" | ||
| "github.com/hatchet-dev/hatchet/pkg/cmdutils" | ||
| ) | ||
|
|
||
| type userCreateEvent struct { | ||
| Username string `json:"username"` | ||
| UserID string `json:"user_id"` | ||
| Data map[string]string `json:"data"` | ||
| } | ||
|
|
||
| type stepOutput struct { | ||
| Message string `json:"message"` | ||
| } | ||
|
|
||
| func main() { | ||
| err := godotenv.Load() | ||
| if err != nil { | ||
| panic(err) | ||
| } | ||
|
|
||
| events := make(chan string, 50) | ||
| if err := run(cmdutils.InterruptChan(), events); err != nil { | ||
| panic(err) | ||
| } | ||
| } | ||
|
|
||
| func run(ch <-chan interface{}, events chan<- string) error { | ||
| c, err := client.New() | ||
|
|
||
| if err != nil { | ||
| return fmt.Errorf("error creating client: %w", err) | ||
| } | ||
|
|
||
| time.Sleep(1 * time.Second) | ||
|
|
||
| // trigger workflow | ||
| workflowRunId, err := c.Admin().RunWorkflow( | ||
| "post-user-update", | ||
| &userCreateEvent{ | ||
| Username: "echo-test", | ||
| UserID: "1234", | ||
| Data: map[string]string{ | ||
| "test": "test", | ||
| }, | ||
| }, | ||
| ) | ||
|
|
||
| if err != nil { | ||
| return fmt.Errorf("error running workflow: %w", err) | ||
| } | ||
|
|
||
| fmt.Println("workflow run id:", workflowRunId) | ||
|
|
||
| interruptCtx, cancel := cmdutils.InterruptContextFromChan(ch) | ||
| defer cancel() | ||
|
|
||
| err = c.Run().On(interruptCtx, workflowRunId, func(event *client.StepRunEvent) error { | ||
| if event.Type == client.StepRunEventTypeCompleted { | ||
| fmt.Println(string(event.Payload)) | ||
| } | ||
|
|
||
| return nil | ||
| }) | ||
|
|
||
| return 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,136 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "time" | ||
|
|
||
| "github.com/joho/godotenv" | ||
|
|
||
| "github.com/hatchet-dev/hatchet/pkg/client" | ||
| "github.com/hatchet-dev/hatchet/pkg/cmdutils" | ||
| "github.com/hatchet-dev/hatchet/pkg/worker" | ||
| ) | ||
|
|
||
| type userCreateEvent struct { | ||
| Username string `json:"username"` | ||
| UserID string `json:"user_id"` | ||
| Data map[string]string `json:"data"` | ||
| } | ||
|
|
||
| type stepOutput struct { | ||
| Message string `json:"message"` | ||
| } | ||
|
|
||
| func main() { | ||
| err := godotenv.Load() | ||
| if err != nil { | ||
| panic(err) | ||
| } | ||
|
|
||
| events := make(chan string, 50) | ||
| if err := run(cmdutils.InterruptChan(), events); err != nil { | ||
| panic(err) | ||
| } | ||
| } | ||
|
|
||
| func run(ch <-chan interface{}, events chan<- string) error { | ||
| c, err := client.New() | ||
|
|
||
| if err != nil { | ||
| return fmt.Errorf("error creating client: %w", err) | ||
| } | ||
|
|
||
| // Create a worker. This automatically reads in a TemporalClient from .env and workflow files from the .hatchet | ||
| // directory, but this can be customized with the `worker.WithTemporalClient` and `worker.WithWorkflowFiles` options. | ||
| w, err := worker.NewWorker( | ||
| worker.WithClient( | ||
| c, | ||
| ), | ||
| ) | ||
| if err != nil { | ||
| return fmt.Errorf("error creating worker: %w", err) | ||
| } | ||
|
|
||
| testSvc := w.NewService("test") | ||
|
|
||
| err = testSvc.On( | ||
| worker.Events("user:create:simple"), | ||
| &worker.WorkflowJob{ | ||
| Name: "post-user-update", | ||
| Description: "This runs after an update to the user model.", | ||
| Steps: []*worker.WorkflowStep{ | ||
| worker.Fn(func(ctx worker.HatchetContext) (result *stepOutput, err error) { | ||
| input := &userCreateEvent{} | ||
| ctx.WorkflowInput(input) | ||
|
|
||
| time.Sleep(1 * time.Second) | ||
|
|
||
| return &stepOutput{ | ||
| Message: "Step 1 got username: " + input.Username, | ||
| }, nil | ||
| }, | ||
| ).SetName("step-one"), | ||
| worker.Fn(func(ctx worker.HatchetContext) (result *stepOutput, err error) { | ||
| input := &userCreateEvent{} | ||
| ctx.WorkflowInput(input) | ||
|
|
||
| time.Sleep(2 * time.Second) | ||
|
|
||
| return &stepOutput{ | ||
| Message: "Step 2 got username: " + input.Username, | ||
| }, nil | ||
| }).SetName("step-two"), | ||
| worker.Fn(func(ctx worker.HatchetContext) (result *stepOutput, err error) { | ||
| step1Out := &stepOutput{} | ||
| ctx.StepOutput("step-one", step1Out) | ||
|
|
||
| step2Out := &stepOutput{} | ||
| ctx.StepOutput("step-two", step2Out) | ||
|
|
||
| time.Sleep(3 * time.Second) | ||
|
|
||
| return &stepOutput{ | ||
| Message: "Step 3: has parents 1 and 2:" + step1Out.Message + ", " + step2Out.Message, | ||
| }, nil | ||
| }).SetName("step-three").AddParents("step-one", "step-two"), | ||
| worker.Fn(func(ctx worker.HatchetContext) (result *stepOutput, err error) { | ||
| step1Out := &stepOutput{} | ||
| ctx.StepOutput("step-one", step1Out) | ||
|
|
||
| step3Out := &stepOutput{} | ||
| ctx.StepOutput("step-three", step3Out) | ||
|
|
||
| time.Sleep(4 * time.Second) | ||
|
|
||
| return &stepOutput{ | ||
| Message: "Step 4: has parents 1 and 3" + step1Out.Message + ", " + step3Out.Message, | ||
| }, nil | ||
| }).SetName("step-four").AddParents("step-one", "step-three"), | ||
| worker.Fn(func(ctx worker.HatchetContext) (result *stepOutput, err error) { | ||
| step4Out := &stepOutput{} | ||
| ctx.StepOutput("step-four", step4Out) | ||
|
|
||
| time.Sleep(5 * time.Second) | ||
|
|
||
| return &stepOutput{ | ||
| Message: "Step 5: has parent 4" + step4Out.Message, | ||
| }, nil | ||
| }).SetName("step-five").AddParents("step-four"), | ||
| }, | ||
| }, | ||
| ) | ||
| if err != nil { | ||
| return fmt.Errorf("error registering workflow: %w", err) | ||
| } | ||
|
|
||
| interruptCtx, cancel := cmdutils.InterruptContextFromChan(ch) | ||
| defer cancel() | ||
|
|
||
| err = w.Start(interruptCtx) | ||
|
|
||
| if err != nil { | ||
| panic(err) | ||
| } | ||
|
|
||
| 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
Oops, something went wrong.