EmissionQueue.Add: fix adding a []byte message

xbus/emissionqueue/emission_queue.go

@@ -77,7 +77,8 @@
 	if eq == nil {
 		return nil
 	}
-	if bArr, ok := payload.([][]byte); ok {
-		msg := NewChunkedMessage(msgType, bArr)
+	switch v := payload.(type) {
+	case [][]byte:
+		msg := NewChunkedMessage(msgType, v)
 
 		return eq.AddMessage(db, msg)
@@ -82,3 +83,17 @@
 
 		return eq.AddMessage(db, msg)
+	case []byte:
+		return eq.AddMessage(db, NewMessage(relatedTo, msgType, v))
+	default:
+		var (
+			b   []byte
+			err error
+		)
+		b, err = json.Marshal(payload)
+		if err != nil {
+			return fmt.Errorf(
+				"xbus.EmissionQueue: failed to marshal. err was: %w. payload is %#v", err, payload)
+		}
+
+		return eq.AddMessage(db, NewMessage(relatedTo, msgType, b))
 	}
@@ -84,15 +99,4 @@
 	}
-	var (
-		b   []byte
-		err error
-	)
-	b, err = json.Marshal(payload)
-	if err != nil {
-		return fmt.Errorf(
-			"xbus.EmissionQueue: failed to marshal. err was: %w. payload is %#v", err, payload)
-	}
-
-	return eq.AddMessage(db, NewMessage(relatedTo, msgType, b))
 }
 
 type syncMessageBuffer struct {

xbus/emissionqueue/emission_queue_test.go

@@ -26,6 +26,7 @@
 
 	require.NoError(t, queue.AddMessage(&sqlHelper, emissionqueue.NewMessage("AddMessage", "test", []byte("abcde"))))
 	require.NoError(t, queue.Add(&sqlHelper, "Add", "test", "fghij"))
+	require.NoError(t, queue.Add(&sqlHelper, "Add", "test", []byte("{}")))
 
 	var messages []emissionqueue.Message
 	require.NoError(t,
@@ -37,7 +38,7 @@
 			&log))
 
 	var relatedTo []string
-	require.Len(t, messages, 2)
+	require.Len(t, messages, 3)
 	if assert.NoError(t, messages[0].RelatedTo.AssignTo(&relatedTo)) {
 		assert.Equal(t, []string{"AddMessage"}, relatedTo)
 	}
@@ -48,4 +49,7 @@
 	}
 	assert.Equal(t, "test", messages[1].MsgType)
 	assert.Equal(t, `"fghij"`, string(messages[1].Content))
+
+	assert.Equal(t, "test", messages[2].MsgType)
+	assert.Equal(t, `{}`, string(messages[2].Content))
 }