The mediator pattern implementation

go.mod

@@ -11,7 +11,10 @@ require (
 	github.com/stretchr/testify v1.8.4
 )
 
-require github.com/corpix/uarand v0.0.0-20170723150923-031be390f409 // indirect
+require (
+	github.com/corpix/uarand v0.0.0-20170723150923-031be390f409 // indirect
+	github.com/mitchellh/hashstructure/v2 v2.0.2 // indirect
+)
 
 require (
 	github.com/davecgh/go-spew v1.1.1 // indirect

go.sum

@@ -95,6 +95,8 @@ github.com/mattn/go-colorable v0.1.6/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope
 github.com/mattn/go-isatty v0.0.5/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
 github.com/mattn/go-isatty v0.0.7/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
 github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU=
+github.com/mitchellh/hashstructure/v2 v2.0.2 h1:vGKWl0YJqUNxE8d+h8f6NJLcCJrgbhC4NcD46KavDd4=
+github.com/mitchellh/hashstructure/v2 v2.0.2/go.mod h1:MG3aRVU/N29oo/V/IhBX8GR/zz4kQkprJgF2EVszyDE=
 github.com/oklog/ulid/v2 v2.1.0 h1:+9lhoxAP56we25tyYETBBY1YLA2SaoLvUFgrP2miPJU=
 github.com/oklog/ulid/v2 v2.1.0/go.mod h1:rcEKHmBBKfef9DhnvX7y1HZBYxjXb0cP5ExxNsTT1QQ=
 github.com/pborman/getopt v0.0.0-20170112200414-7148bc3a4c30/go.mod h1:85jBQOZwpVEaDAr341tbn15RS4fCAsIst0qp7i8ex1o=

grade/internal/application/seedwork/mediator/command.go

@@ -0,0 +1,37 @@
+package mediator
+
+type CommandType struct {
+	name string
+}
+
+func NewCommandType(name string) CommandType {
+	return CommandType{name: name}
+}
+
+func (t CommandType) GetName() string {
+	return t.name
+}
+
+type TypedCommand interface {
+	GetType() CommandType
+}
+
+type Command[T any] interface {
+	TypedCommand
+	GetPayload() T
+}
+
+type CommandImpl[T any] struct {
+	CommandType CommandType
+	Payload     T
+}
+
+func (e CommandImpl[T]) GetType() CommandType {
+	return e.CommandType
+}
+
+func (e CommandImpl[T]) GetPayload() T {
+	return e.Payload
+}
+
+type CommandHandler[T any] func(command T)

grade/internal/application/seedwork/mediator/event.go

@@ -0,0 +1,72 @@
+package mediator
+
+import (
+	"github.com/emacsway/grade/grade/internal/infrastructure/seedwork/identity"
+	"github.com/mitchellh/hashstructure/v2"
+)
+
+type EventType struct {
+	name string
+}
+
+func NewEventType(name string) EventType {
+	return EventType{name: name}
+}
+
+func (t EventType) GetName() string {
+	return t.name
+}
+
+type TypedEvent interface {
+	GetType() EventType
+}
+
+type Event[T any] interface {
+	TypedEvent
+	GetPayload() T
+}
+
+type EventImpl[T any] struct {
+	EventType EventType
+	Payload   T
+}
+
+func (e EventImpl[T]) GetType() EventType {
+	return e.EventType
+}
+
+func (e EventImpl[T]) GetPayload() T {
+	return e.Payload
+}
+
+type EventHandler[T any] interface {
+	identity.Hashed[uint64]
+	Handle(event T)
+}
+
+type EventHandlerImpl[T any] struct {
+	hash     uint64
+	callback func(T)
+}
+
+func NewEventHandler[T any](callback func(T)) (EventHandler[T], error) {
+	handler := &EventHandlerImpl[T]{
+		callback: callback,
+	}
+
+	hash, err := hashstructure.Hash(handler, hashstructure.FormatV2, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	handler.hash = hash
+	return handler, err
+}
+
+func (h *EventHandlerImpl[T]) GetHash() uint64 {
+	return h.hash
+}
+
+func (h *EventHandlerImpl[T]) Handle(smth T) {
+	h.callback(smth)
+}

grade/internal/application/seedwork/mediator/interfaces.go

@@ -0,0 +1,15 @@
+package mediator
+
+import (
+	"github.com/emacsway/grade/grade/internal/domain/seedwork/disposable"
+)
+
+type Mediator[C, E any] interface {
+	Register(commandType CommandType, handler CommandHandler[C]) disposable.Disposable
+	Unregister(commandType CommandType)
+	Send(command C)
+
+	Subscribe(eventType EventType, handler EventHandler[E]) disposable.Disposable
+	Unsubscribe(eventType EventType, handler EventHandler[E])
+	Publish(event E)
+}

grade/internal/infrastructure/seedwork/identity/interfaces.go

@@ -0,0 +1,5 @@
+package identity
+
+type Hashed[T any] interface {
+	GetHash() T
+}

grade/internal/infrastructure/seedwork/mediator/mediator.go

@@ -1,4 +1,83 @@
 package mediator
 
-type MediatorImp struct {
+import (
+	"sync"
+
+	"github.com/emacsway/grade/grade/internal/application/seedwork/mediator"
+	"github.com/emacsway/grade/grade/internal/domain/seedwork/disposable"
+)
+
+type Mediator[C mediator.TypedCommand, E mediator.TypedEvent] struct {
+	hLock    sync.RWMutex
+	handlers map[string]mediator.CommandHandler[C]
+
+	sLock       sync.RWMutex
+	subscribers map[string]map[uint64]mediator.EventHandler[E]
+}
+
+func NewMediator[C mediator.TypedCommand, E mediator.TypedEvent]() mediator.Mediator[C, E] {
+	return &Mediator[C, E]{
+		hLock:    sync.RWMutex{},
+		handlers: map[string]mediator.CommandHandler[C]{},
+
+		sLock:       sync.RWMutex{},
+		subscribers: map[string]map[uint64]mediator.EventHandler[E]{},
+	}
+}
+
+func (m *Mediator[C, E]) Send(command C) {
+	m.hLock.RLock()
+	defer m.hLock.RUnlock()
+
+	if handler, found := m.handlers[command.GetType().GetName()]; found {
+		handler(command)
+	}
+}
+
+func (m *Mediator[C, E]) Register(commandType mediator.CommandType, handler mediator.CommandHandler[C]) disposable.Disposable {
+	m.hLock.Lock()
+	defer m.hLock.Unlock()
+
+	m.handlers[commandType.GetName()] = handler
+	return disposable.NewDisposable(func() {
+		m.Unregister(commandType)
+	})
+}
+
+func (m *Mediator[C, E]) Unregister(commandType mediator.CommandType) {
+	m.hLock.RLock()
+	defer m.hLock.RUnlock()
+
+	delete(m.handlers, commandType.GetName())
+}
+
+func (m *Mediator[C, E]) Subscribe(eventType mediator.EventType, handler mediator.EventHandler[E]) disposable.Disposable {
+	m.sLock.Lock()
+	defer m.sLock.Unlock()
+
+	if _, found := m.subscribers[eventType.GetName()]; !found {
+		m.subscribers[eventType.GetName()] = map[uint64]mediator.EventHandler[E]{}
+	}
+
+	m.subscribers[eventType.GetName()][handler.GetHash()] = handler
+
+	return disposable.NewDisposable(func() {
+		m.Unsubscribe(eventType, handler)
+	})
+}
+
+func (m *Mediator[C, E]) Unsubscribe(eventType mediator.EventType, handler mediator.EventHandler[E]) {
+	m.sLock.Lock()
+	defer m.sLock.Unlock()
+
+	delete(m.subscribers[eventType.GetName()], handler.GetHash())
+}
+
+func (m *Mediator[C, E]) Publish(event E) {
+	m.sLock.RLock()
+	defer m.sLock.RUnlock()
+
+	for _, handler := range m.subscribers[event.GetType().GetName()] {
+		handler.Handle(event)
+	}
 }

grade/internal/infrastructure/seedwork/mediator/mediator_test.go

@@ -0,0 +1,46 @@
+package mediator
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/emacsway/grade/grade/internal/application/seedwork/mediator"
+)
+
+type StringBasedEvent mediator.Event[string]
+type StringBasedCommand mediator.Command[string]
+
+var (
+	cType = mediator.NewCommandType("yo")
+
+	command = mediator.CommandImpl[string]{
+		CommandType: cType,
+		Payload:     "Yoo, man!",
+	}
+
+	eType = mediator.NewEventType("m")
+
+	event = mediator.EventImpl[string]{
+		EventType: eType,
+		Payload:   "WoW!",
+	}
+)
+
+func TestMediator_Publish(t *testing.T) {
+
+	m := NewMediator[StringBasedCommand, StringBasedEvent]()
+
+	h, _ := mediator.NewEventHandler[StringBasedEvent](func(e StringBasedEvent) {
+		fmt.Println(e.GetType())
+		fmt.Println(e.GetPayload())
+	})
+
+	m.Subscribe(eType, h)
+	m.Publish(event)
+
+	m.Register(cType, func(command StringBasedCommand) {
+		fmt.Println(command.GetPayload())
+	})
+
+	m.Send(command)
+}