Handle a PublishSuccess message to support batched events

Author: exanubes

internal/app/protocol/protocol.go

@@ -35,3 +35,14 @@ type DataMessage struct {
 	SubId   string
 	Payload app.Payload
 }
+
+type PublishResultMessage struct {
+	ID       string
+	Failures []FailedEvent
+}
+
+type FailedEvent struct {
+	Success    bool
+	Index      int
+	RawMessage []byte
+}

internal/app/router/router.go

@@ -34,6 +34,10 @@ func (router *MessageHandler) Handle(ctx context.Context, msg app.Message) error
 	case protocol.SuccessMessage:
 		return router.pending.Fulfill(ctx, msg.ID, nil)
 
+	case protocol.PublishResultMessage:
+		// TODO: When publishing a batch of events, need to be able to return an error per failed event or nil
+		return router.pending.Fulfill(ctx, msg.ID, nil)
+
 	case protocol.DataMessage:
 		return router.receive_use_case.Execute(ctx, msg)
 	}

internal/infrastructure/codec/codec.go

@@ -40,7 +40,11 @@ func (codec Codec) Decode(payload app.Payload) (app.Message, error) {
 	case protocol.TypeKeepAlive:
 		return protocol.KeepAliveMessage{}, nil
 
-	case protocol.TypePublishSuccess, protocol.TypeSubscribeSuccess, protocol.TypeUnsubscribeSuccess:
+	case protocol.TypePublishSuccess:
+		event := events.PublishResult{}
+		err := json.Unmarshal(payload, &event)
+		return event.ToProtocol(), err
+	case /*protocol.TypePublishSuccess,*/ protocol.TypeSubscribeSuccess, protocol.TypeUnsubscribeSuccess:
 		event := events.SuccessEvent{}
 		err := json.Unmarshal(payload, &event)
 		return event.ToProtocol(), err

internal/infrastructure/codec/codec_test.go

@@ -40,7 +40,7 @@ func TestDecode(t *testing.T) {
 		{
 			name:       "publish_success message",
 			payload:    []byte(`{"type":"publish_success","id":"sub-1"}`),
-			expect_msg: protocol.SuccessMessage{ID: "sub-1"},
+			expect_msg: protocol.PublishResultMessage{ID: "sub-1", Failures: []protocol.FailedEvent{}},
 		},
 		{
 			name:       "unsubscribe_success message",

internal/infrastructure/events/events.go

@@ -66,3 +66,42 @@ func (event DataEvent) ToProtocol() protocol.DataMessage {
 		Payload: app.Payload(event.Data),
 	}
 }
+
+type PublishResult struct {
+	ID         string    `json:"id"`
+	Type       string    `json:"type"`
+	Successful []Success `json:"successful"`
+	Failed     []Failure `json:"failed"`
+}
+
+type Success struct {
+	ID    string `json:"identifier"`
+	Index int    `json:"index"`
+}
+
+// Best guess, could not find a schema for a failed event item.
+// Raw is used to store the object in its entirety to be able to see the original object
+type Failure struct {
+	ID    string `json:"identifier"`
+	Index int    `json:"index"`
+	Raw   []byte
+}
+
+func (event PublishResult) ToProtocol() protocol.PublishResultMessage {
+	result := make([]protocol.FailedEvent, len(event.Failed))
+	index := 0
+
+	for _, item := range event.Failed {
+		result[index] = protocol.FailedEvent{
+			Success:    false,
+			Index:      item.Index,
+			RawMessage: item.Raw,
+		}
+		index += 1
+	}
+
+	return protocol.PublishResultMessage{
+		ID:       event.ID,
+		Failures: result,
+	}
+}