-
Notifications
You must be signed in to change notification settings - Fork 37
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* Add notifications api * Add basic api tests * Add notify command to resonate cli * Add notify cmd * Add promise to task message * Send Notify at most once + DST * Apply suggestions from code review Co-authored-by: David Farr <[email protected]> --------- Co-authored-by: David Farr <[email protected]>
- Loading branch information
Showing
29 changed files
with
1,480 additions
and
13 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,96 @@ | ||
| package notify | ||
|
|
||
| import ( | ||
| "context" | ||
| "encoding/json" | ||
| "errors" | ||
| "strings" | ||
| "time" | ||
|
|
||
| "github.com/resonatehq/resonate/pkg/client" | ||
| v1 "github.com/resonatehq/resonate/pkg/client/v1" | ||
| "github.com/spf13/cobra" | ||
| ) | ||
|
|
||
| var notifyExample = ` | ||
| # Create a notify | ||
| resonate notify create foo --promise-id bar --timeout 1h --recv default | ||
| # Create a notify with url | ||
| resonate notify create foo --promise-id bar --timeout 1h --recv poll://default/6fa89b7e-4a56-40e8-ba4e-78864caa3278 | ||
| # Create a notify with object | ||
| resonate notify create foo --promise-id bar --timeout 1h --recv {"type": "poll", "data": {"group": "default", "id": "6fa89b7e-4a56-40e8-ba4e-78864caa3278"}} | ||
| ` | ||
|
|
||
| func CreateNotifyCmd(c client.Client) *cobra.Command { | ||
| var ( | ||
| promiseId string | ||
| timeout time.Duration | ||
| recvStr string | ||
| ) | ||
| cmd := &cobra.Command{ | ||
| Use: "create <id>", | ||
| Short: "Create notify", | ||
| Example: notifyExample, | ||
| RunE: func(cmd *cobra.Command, args []string) error { | ||
| if len(args) != 1 { | ||
| return errors.New("must specify an id") | ||
| } | ||
|
|
||
| id := args[0] | ||
|
|
||
| var recv v1.Recv | ||
|
|
||
| if json.Valid([]byte(recvStr)) { | ||
| var recv0 v1.Recv0 | ||
|
|
||
| if err := json.Unmarshal([]byte(recvStr), &recv0); err != nil { | ||
| return err | ||
| } | ||
| if err := recv.FromRecv0(recv0); err != nil { | ||
| return err | ||
| } | ||
| } else { | ||
| if err := recv.FromRecv1(recvStr); err != nil { | ||
| return err | ||
| } | ||
| } | ||
|
|
||
| body := v1.CreateNotifyJSONRequestBody{ | ||
| Id: id, | ||
| PromiseId: promiseId, | ||
| Timeout: time.Now().Add(timeout).UnixMilli(), | ||
| Recv: recv, | ||
| } | ||
|
|
||
| res, err := c.V1().CreateNotifyWithResponse(context.TODO(), nil, body) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| if res.StatusCode() == 201 { | ||
| cmd.Printf("Created notification: %s\n", id) | ||
| } else if res.StatusCode() == 200 { | ||
| if res.JSON200.Promise != nil && res.JSON200.Promise.State != v1.PromiseStatePENDING { | ||
| cmd.Printf("Promise %s already %s\n", promiseId, strings.ToLower(string(res.JSON200.Promise.State))) | ||
| } else { | ||
| cmd.Printf("Created notification: %s (deduplicated)\n", id) | ||
| } | ||
| } else { | ||
| cmd.PrintErrln(res.Status(), string(res.Body)) | ||
| } | ||
|
|
||
| return nil | ||
| }, | ||
| } | ||
|
|
||
| cmd.Flags().StringVar(&promiseId, "promise-id", "", "promise id") | ||
| cmd.Flags().DurationVar(&timeout, "timeout", 0, "task timeout") | ||
| cmd.Flags().StringVar(&recvStr, "recv", "default", "task receiver") | ||
|
|
||
| _ = cmd.MarkFlagRequired("promise-id") | ||
| _ = cmd.MarkFlagRequired("timeout") | ||
|
|
||
| return cmd | ||
| } |
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,41 @@ | ||
| package notify | ||
|
|
||
| import ( | ||
| "github.com/resonatehq/resonate/pkg/client" | ||
| "github.com/spf13/cobra" | ||
| ) | ||
|
|
||
| func NewCmd() *cobra.Command { | ||
| var ( | ||
| c = client.New() | ||
| server string | ||
| username string | ||
| password string | ||
| ) | ||
|
|
||
| cmd := &cobra.Command{ | ||
| Use: "notifications", | ||
| Aliases: []string{"notifications"}, | ||
| Short: "Resonate notifications", | ||
| Run: func(cmd *cobra.Command, args []string) { | ||
| _ = cmd.Help() | ||
| }, | ||
| PersistentPreRunE: func(cmd *cobra.Command, args []string) error { | ||
| if username != "" || password != "" { | ||
| c.SetBasicAuth(username, password) | ||
| } | ||
|
|
||
| return c.Setup(server) | ||
| }, | ||
| } | ||
|
|
||
| // Add subcommands | ||
| cmd.AddCommand(CreateNotifyCmd(c)) | ||
|
|
||
| // Flags | ||
| cmd.PersistentFlags().StringVarP(&server, "server", "", "http://localhost:8001", "resonate url") | ||
| cmd.PersistentFlags().StringVarP(&username, "username", "U", "", "basic auth username") | ||
| cmd.PersistentFlags().StringVarP(&password, "password", "P", "", "basic auth password") | ||
|
|
||
| return cmd | ||
| } |
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,140 @@ | ||
| package coroutines | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "log/slog" | ||
|
|
||
| "github.com/resonatehq/gocoro" | ||
| "github.com/resonatehq/resonate/internal/kernel/t_aio" | ||
| "github.com/resonatehq/resonate/internal/kernel/t_api" | ||
| "github.com/resonatehq/resonate/internal/util" | ||
| "github.com/resonatehq/resonate/pkg/callback" | ||
| "github.com/resonatehq/resonate/pkg/message" | ||
| "github.com/resonatehq/resonate/pkg/promise" | ||
| ) | ||
|
|
||
| func CreateNotify(c gocoro.Coroutine[*t_aio.Submission, *t_aio.Completion, any], r *t_api.Request) (*t_api.Response, error) { | ||
| util.Assert(r.Kind == t_api.CreateNotify, "Request kind must be CreateNotify") | ||
| var res *t_api.Response | ||
|
|
||
| // read the promise to see if it exists | ||
| completion, err := gocoro.YieldAndAwait(c, &t_aio.Submission{ | ||
| Kind: t_aio.Store, | ||
| Tags: r.Tags, | ||
| Store: &t_aio.StoreSubmission{ | ||
| Transaction: &t_aio.Transaction{ | ||
| Commands: []*t_aio.Command{ | ||
| { | ||
| Kind: t_aio.ReadPromise, | ||
| ReadPromise: &t_aio.ReadPromiseCommand{ | ||
| Id: r.CreateNotify.PromiseId, | ||
| }, | ||
| }, | ||
| }, | ||
| }, | ||
| }, | ||
| }) | ||
|
|
||
| if err != nil { | ||
| slog.Error("failed to read promise", "req", r, "err", err) | ||
| return nil, t_api.NewError(t_api.StatusAIOStoreError, err) | ||
| } | ||
|
|
||
| util.Assert(completion.Store != nil, "completion must not be nil") | ||
| util.Assert(len(completion.Store.Results) == 1, "completion must have one result") | ||
|
|
||
| result := completion.Store.Results[0].ReadPromise | ||
| util.Assert(result != nil, "result must not be nil") | ||
| util.Assert(result.RowsReturned == 0 || result.RowsReturned == 1, "result must return 0 or 1 rows") | ||
|
|
||
| if result.RowsReturned == 1 { | ||
| p, err := result.Records[0].Promise() | ||
| if err != nil { | ||
| slog.Error("failed to parse promise record", "record", result.Records[0], "err", err) | ||
| return nil, t_api.NewError(t_api.StatusAIOStoreError, err) | ||
| } | ||
|
|
||
| // If the notify is already created return 200 and an empty notify | ||
| var cb *callback.Callback | ||
| status := t_api.StatusOK | ||
|
|
||
| if p.State == promise.Pending { | ||
| mesg := &message.Mesg{ | ||
| Type: message.Notify, | ||
| Root: r.CreateNotify.PromiseId, | ||
| } | ||
|
|
||
| createdOn := c.Time() | ||
|
|
||
| callbackId := fmt.Sprintf("%s.%s", r.CreateNotify.PromiseId, r.CreateNotify.Id) | ||
| completion, err := gocoro.YieldAndAwait(c, &t_aio.Submission{ | ||
| Kind: t_aio.Store, | ||
| Tags: r.Tags, | ||
| Store: &t_aio.StoreSubmission{ | ||
| Transaction: &t_aio.Transaction{ | ||
| Commands: []*t_aio.Command{ | ||
| { | ||
| Kind: t_aio.CreateCallback, | ||
| CreateCallback: &t_aio.CreateCallbackCommand{ | ||
| Id: callbackId, | ||
| PromiseId: r.CreateNotify.PromiseId, | ||
| Recv: r.CreateNotify.Recv, | ||
| Mesg: mesg, | ||
| Timeout: r.CreateNotify.Timeout, | ||
| CreatedOn: createdOn, | ||
| }, | ||
| }, | ||
| }, | ||
| }, | ||
| }, | ||
| }) | ||
|
|
||
| if err != nil { | ||
| slog.Error("failed to create notify", "req", r, "err", err) | ||
| return nil, t_api.NewError(t_api.StatusAIOStoreError, err) | ||
| } | ||
|
|
||
| util.Assert(completion.Store != nil, "completion must not be nil") | ||
| util.Assert(len(completion.Store.Results) == 1, "completion must have one result") | ||
|
|
||
| result := completion.Store.Results[0].CreateCallback | ||
| util.Assert(result != nil, "result must not be nil") | ||
| util.Assert(result.RowsAffected == 0 || result.RowsAffected == 1, "result must return 0 or 1 rows") | ||
|
|
||
| if result.RowsAffected == 1 { | ||
| status = t_api.StatusCreated | ||
| cb = &callback.Callback{ | ||
| Id: callbackId, | ||
| PromiseId: r.CreateNotify.PromiseId, | ||
| Recv: r.CreateNotify.Recv, | ||
| Mesg: mesg, | ||
| Timeout: r.CreateNotify.Timeout, | ||
| CreatedOn: createdOn, | ||
| } | ||
| } | ||
| } | ||
|
|
||
| res = &t_api.Response{ | ||
| Kind: t_api.CreateNotify, | ||
| Tags: r.Tags, | ||
| CreateNotify: &t_api.CreateNotifyResponse{ | ||
| // Status could be StatusOk or StatusCreated if the Callback Id was already present | ||
| Status: status, | ||
| Callback: cb, | ||
| Promise: p, | ||
| }, | ||
| } | ||
|
|
||
| } else { | ||
| res = &t_api.Response{ | ||
| Kind: t_api.CreateNotify, | ||
| Tags: r.Tags, | ||
| CreateNotify: &t_api.CreateNotifyResponse{ | ||
| Status: t_api.StatusPromiseNotFound, | ||
| }, | ||
| } | ||
| } | ||
|
|
||
| util.Assert(res != nil, "response must not be nil") | ||
| return res, nil | ||
| } |
Oops, something went wrong.