feat: adding destination id (#9)

rudderlabs · Apr 22, 2024 · ee57769 · ee57769
1 parent aa5a8e7
commit ee57769
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 0 deletions.
diff --git a/go/stream/message.go b/go/stream/message.go
@@ -13,6 +13,7 @@ const (
 	mapKeyRoutingKey      = "routingKey"
 	mapKeyWorkspaceID     = "workspaceID"
 	mapKeySourceID        = "sourceID"
+	mapKeyDestinationID   = "destinationID"
 	mapKeyRequestIP       = "requestIP"
 	mapKeyReceivedAt      = "receivedAt"
 	mapKeyUserID          = "userID"
@@ -33,6 +34,7 @@ type MessageProperties struct {
 	SourceID        string    `json:"sourceID" validate:"required"`
 	ReceivedAt      time.Time `json:"receivedAt" validate:"required"`
 	RequestIP       string    `json:"requestIP" validate:"required"`
+	DestinationID   string    `json:"destinationID"`             // optional
 	UserID          string    `json:"userID,omitempty"`          // optional
 	SourceJobRunID  string    `json:"sourceJobRunID,omitempty"`  // optional
 	SourceTaskRunID string    `json:"sourceTaskRunID,omitempty"` // optional
@@ -53,6 +55,7 @@ func FromMapProperties(properties map[string]string) (MessageProperties, error)
 		RequestIP:       properties[mapKeyRequestIP],
 		UserID:          properties[mapKeyUserID],
 		SourceID:        properties[mapKeySourceID],
+		DestinationID:   properties[mapKeyDestinationID],
 		ReceivedAt:      receivedAt,
 		SourceJobRunID:  properties[mapKeySourceJobRunID],
 		SourceTaskRunID: properties[mapKeySourceTaskRunID],
@@ -68,6 +71,7 @@ func ToMapProperties(properties MessageProperties) map[string]string {
 		mapKeyWorkspaceID:     properties.WorkspaceID,
 		mapKeyUserID:          properties.UserID,
 		mapKeySourceID:        properties.SourceID,
+		mapKeyDestinationID:   properties.DestinationID,
 		mapKeyRequestIP:       properties.RequestIP,
 		mapKeyReceivedAt:      properties.ReceivedAt.Format(time.RFC3339Nano),
 		mapKeySourceJobRunID:  properties.SourceJobRunID,

diff --git a/go/stream/message_test.go b/go/stream/message_test.go
@@ -18,6 +18,7 @@ func TestMessage(t *testing.T) {
 			"workspaceID":     "workspaceID",
 			"userID":          "userID",
 			"sourceID":        "sourceID",
+			"destinationID":   "destinationID",
 			"requestIP":       "10.29.13.20",
 			"receivedAt":      time.Date(2024, 8, 1, 0o2, 30, 50, 200, time.UTC).Format(time.RFC3339Nano),
 			"sourceJobRunID":  "sourceJobRunID",
@@ -34,6 +35,7 @@ func TestMessage(t *testing.T) {
 			WorkspaceID:     "workspaceID",
 			UserID:          "userID",
 			SourceID:        "sourceID",
+			DestinationID:   "destinationID",
 			RequestIP:       "10.29.13.20",
 			ReceivedAt:      time.Date(2024, 8, 1, 0o2, 30, 50, 200, time.UTC),
 			SourceJobRunID:  "sourceJobRunID",
@@ -62,6 +64,7 @@ func TestMessage(t *testing.T) {
 				"workspaceID": "workspaceID",
 				"userID": "userID",
 				"sourceID": "sourceID",
+				"destinationID": "destinationID",
 				"receivedAt": "2024-08-01T02:30:50.0000002Z",
 				"requestIP": "10.29.13.20",
 				"sourceJobRunID": "sourceJobRunID",
@@ -87,6 +90,7 @@ func TestMessage(t *testing.T) {
 				WorkspaceID:     "workspaceID",
 				UserID:          "userID",
 				SourceID:        "sourceID",
+				DestinationID:   "destinationID",
 				RequestIP:       "10.29.13.20",
 				ReceivedAt:      time.Date(2024, 8, 1, 0o2, 30, 50, 200, time.UTC),
 				SourceJobRunID:  "sourceJobRunID",