create a copy of ServerComMessage for using in a cluster

server/datamodel.go

@@ -439,6 +439,15 @@ type MsgServerCtrl struct {
 	Timestamp time.Time `json:"ts"`
 }
 
+// Deep-shallow copy.
+func (src *MsgServerCtrl) copy() *MsgServerCtrl {
+	if src == nil {
+		return nil
+	}
+	dst := *src
+	return &dst
+}
+
 // MsgServerData is a server {data} message.
 type MsgServerData struct {
 	Topic string `json:"topic"`
@@ -451,6 +460,15 @@ type MsgServerData struct {
 	Content   interface{}            `json:"content"`
 }
 
+// Deep-shallow copy.
+func (src *MsgServerData) copy() *MsgServerData {
+	if src == nil {
+		return nil
+	}
+	dst := *src
+	return &dst
+}
+
 // MsgServerPres is presence notification {pres} (authoritative update).
 type MsgServerPres struct {
 	Topic     string        `json:"topic"`
@@ -488,6 +506,15 @@ type MsgServerPres struct {
 	ExcludeUser string `json:"-"`
 }
 
+// Deep-shallow copy.
+func (src *MsgServerPres) copy() *MsgServerPres {
+	if src == nil {
+		return nil
+	}
+	dst := *src
+	return &dst
+}
+
 // MsgServerMeta is a topic metadata {meta} update.
 type MsgServerMeta struct {
 	Id    string `json:"id,omitempty"`
@@ -507,6 +534,15 @@ type MsgServerMeta struct {
 	Cred []*MsgCredServer `json:"cred,omitempty"`
 }
 
+// Deep-shallow copy of meta message. Deep copy of Id and Topic fields, shallow copy of payload.
+func (src *MsgServerMeta) copy() *MsgServerMeta {
+	if src == nil {
+		return nil
+	}
+	dst := *src
+	return &dst
+}
+
 // MsgServerInfo is the server-side copy of MsgClientNote with From added (non-authoritative).
 type MsgServerInfo struct {
 	Topic string `json:"topic"`
@@ -518,6 +554,15 @@ type MsgServerInfo struct {
 	SeqId int `json:"seq,omitempty"`
 }
 
+// Deep copy
+func (src *MsgServerInfo) copy() *MsgServerInfo {
+	if src == nil {
+		return nil
+	}
+	dst := *src
+	return &dst
+}
+
 // ServerComMessage is a wrapper for server-side messages.
 type ServerComMessage struct {
 	Ctrl *MsgServerCtrl `json:"ctrl,omitempty"`
@@ -540,6 +585,30 @@ type ServerComMessage struct {
 	skipSid string
 }
 
+// Deep-shallow copy of ServerComMessage. Deep copy of service fields,
+// shallow copy of session and payload.
+func (src *ServerComMessage) copy() *ServerComMessage {
+	if src == nil {
+		return nil
+	}
+	dst := &ServerComMessage{
+		id:        src.id,
+		rcptto:    src.rcptto,
+		timestamp: src.timestamp,
+		from:      src.from,
+		sess:      src.sess,
+		skipSid:   src.skipSid,
+	}
+
+	dst.Ctrl = src.Ctrl.copy()
+	dst.Data = src.Data.copy()
+	dst.Meta = src.Meta.copy()
+	dst.Pres = src.Pres.copy()
+	dst.Info = src.Info.copy()
+
+	return dst
+}
+
 // Generators of server-side error messages {ctrl}.
 
 // NoErr indicates successful completion (200)

server/session.go

@@ -1064,8 +1064,9 @@ func (s *Session) serialize(msg *ServerComMessage) interface{} {
 	}
 
 	if s.proto == CLUSTER {
-		// No need to serialize the message to bytes within the cluster.
-		return msg
+		// No need to serialize the message to bytes within the cluster,
+		// but we have to create a copy because the original msg can be mutated.
+		return msg.copy()
 	}
 
 	out, _ := json.Marshal(msg)