Add type to the message sent by the sender subsystem (#535)

resonatehq · Jan 22, 2025 · fb84e20 · fb84e20
1 parent de5339e
commit fb84e20
Show file tree

Hide file tree

Showing 2 changed files with 28 additions and 12 deletions.
diff --git a/api/openapi-poll.yml b/api/openapi-poll.yml
@@ -40,9 +40,10 @@ paths:
               schema:
                 type: string
                 example: |
-                  data: {"href":{"claim":"http://localhost:8001/tasks/claim/1/1","complete":"http://localhost:8001/tasks/complete/1/1","heartbeat":"http://localhost:8001/tasks/heartbeat/1/1"},"task":{"id":"1","counter":1}}
-                  data: {"href":{"claim":"http://localhost:8001/tasks/claim/1/2","complete":"http://localhost:8001/tasks/complete/1/2","heartbeat":"http://localhost:8001/tasks/heartbeat/1/2"},"task":{"id":"1","counter":2}}
-                  data: {"href":{"claim":"http://localhost:8001/tasks/claim/1/3","complete":"http://localhost:8001/tasks/complete/1/3","heartbeat":"http://localhost:8001/tasks/heartbeat/1/3"},"task":{"id":"1","counter":3}}
+                  data: {"type": "invoke", "href":{"claim":"http://localhost:8001/tasks/claim/1/1","complete":"http://localhost:8001/tasks/complete/1/1","heartbeat":"http://localhost:8001/tasks/heartbeat/1/1"},"task":{"id":"1","counter":1}}
+                  data: {"type": "resume", "href":{"claim":"http://localhost:8001/tasks/claim/1/2","complete":"http://localhost:8001/tasks/complete/1/2","heartbeat":"http://localhost:8001/tasks/heartbeat/1/2"},"task":{"id":"1","counter":2}}
+                  data: {"type": "invoke", "href":{"claim":"http://localhost:8001/tasks/claim/1/3","complete":"http://localhost:8001/tasks/complete/1/3","heartbeat":"http://localhost:8001/tasks/heartbeat/1/3"},"task":{"id":"1","counter":3}}
+                  data: {"type": "notify", "promise":{"id":"bah","state":"RESOLVED","param":{},"value":{},"timeout":1737661069533,"createdOn":1737574669536,"completedOn":1737574722732}}
         400:
           description: Invalid request
         405:

diff --git a/internal/app/subsystems/aio/sender/sender.go b/internal/app/subsystems/aio/sender/sender.go
@@ -14,6 +14,7 @@ import (
 	"github.com/resonatehq/resonate/internal/kernel/t_aio"
 	"github.com/resonatehq/resonate/internal/metrics"
 	"github.com/resonatehq/resonate/internal/util"
+	"github.com/resonatehq/resonate/pkg/message"
 	"github.com/resonatehq/resonate/pkg/receiver"
 )
 
@@ -244,15 +245,29 @@ func (w *SenderWorker) Process(sqe *bus.SQE[t_aio.Submission, t_aio.Completion])
 		return
 	}
 
-	body, err := json.Marshal(map[string]interface{}{
-		"task":    sqe.Submission.Sender.Task,
-		"promise": sqe.Submission.Sender.Promise,
-		"href": map[string]string{
-			"claim":     sqe.Submission.Sender.ClaimHref,
-			"complete":  sqe.Submission.Sender.CompleteHref,
-			"heartbeat": sqe.Submission.Sender.HeartbeatHref,
-		},
-	})
+	var body []byte
+	var err error
+
+	messgType := sqe.Submission.Sender.Task.Mesg.Type
+
+	if messgType == message.Notify {
+		util.Assert(sqe.Submission.Sender.Promise != nil, "promise must not be nil for a notify message")
+		body, err = json.Marshal(map[string]interface{}{
+			"type":    messgType,
+			"promise": sqe.Submission.Sender.Promise,
+		})
+	} else {
+		body, err = json.Marshal(map[string]interface{}{
+			"type": messgType,
+			"task": sqe.Submission.Sender.Task,
+			"href": map[string]string{
+				"claim":     sqe.Submission.Sender.ClaimHref,
+				"complete":  sqe.Submission.Sender.CompleteHref,
+				"heartbeat": sqe.Submission.Sender.HeartbeatHref,
+			},
+		})
+	}
+
 	if err != nil {
 		cqe.Error = err
 		w.aio.EnqueueCQE(cqe)