Add CQRS marshalers

components/cqrs/marshaler_name.go

@@ -0,0 +1,55 @@
+package cqrs
+
+import (
+	"github.com/ThreeDotsLabs/watermill/message"
+)
+
+// NameMarshaler retrieves the name from a message implementing the following interface:
+// 		type namedMessage interface {
+// 			Name() string
+// 		}
+//
+// It wraps another marshaler which takes care of the message serialization/deserialization.
+// If none provided, it falls back to JSON marshaler.
+type NameMarshaler struct {
+	Marshaler CommandEventMarshaler
+}
+
+func (m *NameMarshaler) marshaler() CommandEventMarshaler {
+	if m.Marshaler == nil {
+		return JSONMarshaler{}
+	}
+
+	return m.Marshaler
+}
+
+type namedMessage interface {
+	Name() string
+}
+
+func (m *NameMarshaler) Marshal(v interface{}) (*message.Message, error) {
+	msg, err := m.marshaler().Marshal(v)
+	if err != nil {
+		return nil, err
+	}
+
+	msg.Metadata.Set("name", m.Name(v))
+
+	return msg, nil
+}
+
+func (m *NameMarshaler) Unmarshal(msg *message.Message, v interface{}) error {
+	return m.marshaler().Unmarshal(msg, v)
+}
+
+func (m *NameMarshaler) Name(v interface{}) string {
+	if v, ok := v.(namedMessage); ok {
+		return v.Name()
+	}
+
+	return m.marshaler().Name(v)
+}
+
+func (m *NameMarshaler) NameFromMessage(msg *message.Message) string {
+	return msg.Metadata.Get("name")
+}

components/cqrs/marshaler_name_test.go

@@ -0,0 +1,61 @@
+package cqrs_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/ThreeDotsLabs/watermill"
+	"github.com/ThreeDotsLabs/watermill/components/cqrs"
+)
+
+type NamedEvent struct {
+	TestEvent
+}
+
+func (e NamedEvent) Name() string {
+	return "testEvent"
+}
+
+func TestNameMarshaler(t *testing.T) {
+	eventToMarshal := NamedEvent{
+		TestEvent: TestEvent{
+			ID:   watermill.NewULID(),
+			When: time.Date(2016, time.August, 15, 14, 13, 12, 0, time.UTC),
+		},
+	}
+
+	marshaler := &cqrs.NameMarshaler{}
+
+	msg, err := marshaler.Marshal(eventToMarshal)
+	require.NoError(t, err)
+
+	assert.Equal(t, "testEvent", marshaler.NameFromMessage(msg))
+
+	eventToUnmarshal := NamedEvent{}
+	err = marshaler.Unmarshal(msg, &eventToUnmarshal)
+	require.NoError(t, err)
+
+	assert.EqualValues(t, eventToMarshal, eventToUnmarshal)
+	assert.Equal(t, "testEvent", marshaler.Name(eventToUnmarshal))
+}
+
+func TestNameMarshaler_NotNamedMessage(t *testing.T) {
+	eventToMarshal := TestEvent{
+		ID:   watermill.NewULID(),
+		When: time.Date(2016, time.August, 15, 14, 13, 12, 0, time.UTC),
+	}
+
+	marshaler := &cqrs.NameMarshaler{}
+
+	msg, err := marshaler.Marshal(eventToMarshal)
+	require.NoError(t, err)
+
+	eventToUnmarshal := TestEvent{}
+	err = marshaler.Unmarshal(msg, &eventToUnmarshal)
+	require.NoError(t, err)
+
+	assert.EqualValues(t, eventToMarshal, eventToUnmarshal)
+}

components/cqrs/marshaler_structname.go

@@ -0,0 +1,49 @@
+package cqrs
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/ThreeDotsLabs/watermill/message"
+)
+
+// StructNameMarshaler uses the struct name (without the package) as the message name.
+//
+// It wraps another marshaler which takes care of the message serialization/deserialization.
+// If none provided, it falls back to JSON marshaler.
+type StructNameMarshaler struct {
+	Marshaler CommandEventMarshaler
+}
+
+func (m *StructNameMarshaler) marshaler() CommandEventMarshaler {
+	if m.Marshaler == nil {
+		return JSONMarshaler{}
+	}
+
+	return m.Marshaler
+}
+
+func (m *StructNameMarshaler) Marshal(v interface{}) (*message.Message, error) {
+	msg, err := m.marshaler().Marshal(v)
+	if err != nil {
+		return nil, err
+	}
+
+	msg.Metadata.Set("name", m.Name(v))
+
+	return msg, nil
+}
+
+func (m *StructNameMarshaler) Unmarshal(msg *message.Message, v interface{}) error {
+	return m.marshaler().Unmarshal(msg, v)
+}
+
+func (m *StructNameMarshaler) Name(v interface{}) string {
+	segments := strings.Split(fmt.Sprintf("%T", v), ".")
+
+	return segments[len(segments)-1]
+}
+
+func (m *StructNameMarshaler) NameFromMessage(msg *message.Message) string {
+	return msg.Metadata.Get("name")
+}

components/cqrs/marshaler_structname_test.go

@@ -0,0 +1,33 @@
+package cqrs_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/ThreeDotsLabs/watermill"
+	"github.com/ThreeDotsLabs/watermill/components/cqrs"
+)
+
+func TestStructNameMarshaler(t *testing.T) {
+	eventToMarshal := TestEvent{
+		ID:   watermill.NewULID(),
+		When: time.Date(2016, time.August, 15, 14, 13, 12, 0, time.UTC),
+	}
+
+	marshaler := &cqrs.StructNameMarshaler{}
+
+	msg, err := marshaler.Marshal(eventToMarshal)
+	require.NoError(t, err)
+
+	assert.Equal(t, "TestEvent", marshaler.NameFromMessage(msg))
+
+	eventToUnmarshal := TestEvent{}
+	err = marshaler.Unmarshal(msg, &eventToUnmarshal)
+	require.NoError(t, err)
+
+	assert.EqualValues(t, eventToMarshal, eventToUnmarshal)
+	assert.Equal(t, "TestEvent", marshaler.Name(eventToUnmarshal))
+}