refactoring

Author: andrewwebber

commands.go

@@ -1,11 +1,8 @@
 package cqrs
 
 import (
-	"log"
 	"reflect"
 	"time"
-
-	"github.com/pborman/uuid"
 )
 
 // Command represents an actor intention to alter the state of the system
@@ -20,39 +17,60 @@ type Command struct {
 // CreateCommand is a helper for creating a new command object with populated default properties
 func CreateCommand(body interface{}) Command {
 	commandType := reflect.TypeOf(body)
-	return Command{uuid.New(), uuid.New(), commandType.String(), time.Now(), body}
+	return Command{MessageID: "mid:" + NewUUIDString(),
+		CorrelationID: "cid:" + NewUUIDString(),
+		CommandType:   commandType.String(),
+		Created:       time.Now(),
+
+		Body: body}
 }
 
 // CreateCommandWithCorrelationID is a helper for creating a new command object with populated default properties
 func CreateCommandWithCorrelationID(body interface{}, correlationID string) Command {
 	commandType := reflect.TypeOf(body)
-	return Command{uuid.New(), correlationID, commandType.String(), time.Now(), body}
+	return Command{MessageID: "mid:" + NewUUIDString(),
+		CorrelationID: correlationID,
+		CommandType:   commandType.String(),
+		Created:       time.Now(),
+		Body:          body}
 }
 
 // CommandPublisher is responsilbe for publishing commands
 type CommandPublisher interface {
-	Publish(Command) error
+	PublishCommands([]Command) error
 }
 
 // CommandReceiver is responsible for receiving commands
 type CommandReceiver interface {
 	ReceiveCommands(CommandReceiverOptions) error
 }
 
+// CommandBus ...
+type CommandBus interface {
+	CommandReceiver
+	CommandPublisher
+}
+
 // CommandDispatchManager is responsible for coordinating receiving messages from command receivers and dispatching them to the command dispatcher.
 type CommandDispatchManager struct {
 	commandDispatcher *MapBasedCommandDispatcher
 	typeRegistry      TypeRegistry
 	receiver          CommandReceiver
 }
 
+// CommandDispatcher the internal command dispatcher
+func (m *CommandDispatchManager) CommandDispatcher() CommandDispatcher {
+	return m.commandDispatcher
+}
+
 // CommandReceiverOptions is an initalization structure to communicate to and from a command receiver go routine
 type CommandReceiverOptions struct {
 	TypeRegistry   TypeRegistry
 	Close          chan chan error
 	Error          chan error
-	ReceiveCommand chan CommandTransactedAccept
+	ReceiveCommand CommandHandler
 	Exclusive      bool
+	ListenerCount  int
 }
 
 // CommandTransactedAccept is the message routed from a command receiver to the command manager.
@@ -106,17 +124,21 @@ func (m *MapBasedCommandDispatcher) DispatchCommand(command Command) error {
 	if handlers, ok := m.registry[bodyType]; ok {
 		for _, handler := range handlers {
 			if err := handler(command); err != nil {
+				metricsCommandsFailed.WithLabelValues(command.CommandType).Inc()
 				return err
 			}
 		}
 	}
 
 	for _, handler := range m.globalHandlers {
 		if err := handler(command); err != nil {
+			metricsCommandsFailed.WithLabelValues(command.CommandType).Inc()
 			return err
 		}
 	}
 
+	metricsCommandsDispatched.WithLabelValues(command.CommandType).Inc()
+
 	return nil
 }
 
@@ -137,46 +159,48 @@ func (m *CommandDispatchManager) RegisterGlobalHandler(handler CommandHandler) {
 }
 
 // Listen starts a listen loop processing channels related to new incoming events, errors and stop listening requests
-func (m *CommandDispatchManager) Listen(stop <-chan bool, exclusive bool) error {
+func (m *CommandDispatchManager) Listen(stop <-chan bool, exclusive bool, listenerCount int) error {
 	// Create communication channels
 	//
 	// for closing the queue listener,
 	closeChannel := make(chan chan error)
 	// receiving errors from the listener thread (go routine)
 	errorChannel := make(chan error)
-	// and receiving commands from the queue
-	receiveCommandChannel := make(chan CommandTransactedAccept)
+
+	// Command received channel receives a result with a channel to respond to, signifying successful processing of the message.
+	// This should eventually call a command handler. See cqrs.NewVersionedCommandDispatcher()
+	receiveCommandHandler := func(command Command) error {
+		PackageLogger().Debugf("CommandDispatchManager.DispatchCommand: %v", command.CorrelationID)
+		err := m.commandDispatcher.DispatchCommand(command)
+		if err != nil {
+			PackageLogger().Debugf("Error dispatching command: %v", err)
+		}
+
+		return err
+	}
 
 	// Start receiving commands by passing these channels to the worker thread (go routine)
-	options := CommandReceiverOptions{m.typeRegistry, closeChannel, errorChannel, receiveCommandChannel, exclusive}
+	options := CommandReceiverOptions{m.typeRegistry, closeChannel, errorChannel, receiveCommandHandler, exclusive, listenerCount}
 	if err := m.receiver.ReceiveCommands(options); err != nil {
 		return err
 	}
+	go func() {
+		for {
+			// Wait on multiple channels using the select control flow.
+			select {
+			case <-stop:
+				PackageLogger().Debugf("CommandDispatchManager.Stopping")
+				closeSignal := make(chan error)
+				closeChannel <- closeSignal
+				PackageLogger().Debugf("CommandDispatchManager.Stopped")
+				<-closeSignal
+			// Receiving on this channel signifys an error has occured worker processor side
+			case err := <-errorChannel:
+				PackageLogger().Debugf("CommandDispatchManager.ErrorReceived: %s", err)
 
-	for {
-		// Wait on multiple channels using the select control flow.
-		select {
-		// Command received channel receives a result with a channel to respond to, signifying successful processing of the message.
-		// This should eventually call a command handler. See cqrs.NewVersionedCommandDispatcher()
-		case command := <-receiveCommandChannel:
-			log.Println("CommandDispatchManager.DispatchCommand: ", command.Command)
-			if err := m.commandDispatcher.DispatchCommand(command.Command); err != nil {
-				log.Println("Error dispatching command: ", err)
-				command.ProcessedSuccessfully <- false
-			} else {
-				command.ProcessedSuccessfully <- true
-				log.Println("CommandDispatchManager.DispatchSuccessful")
 			}
-		case <-stop:
-			log.Println("CommandDispatchManager.Stopping")
-			closeSignal := make(chan error)
-			closeChannel <- closeSignal
-			defer log.Println("CommandDispatchManager.Stopped")
-			return <-closeSignal
-		// Receiving on this channel signifys an error has occured worker processor side
-		case err := <-errorChannel:
-			log.Println("CommandDispatchManager.ErrorReceived: ", err)
-			return err
 		}
-	}
+	}()
+
+	return nil
 }

commands_test.go

@@ -1,7 +1,6 @@
 package cqrs_test
 
 import (
-	"log"
 	"testing"
 
 	"github.com/andrewwebber/cqrs"
@@ -15,12 +14,15 @@ func TestCommandDispatcher(t *testing.T) {
 	dispatcher := cqrs.NewMapBasedCommandDispatcher()
 	success := false
 	dispatcher.RegisterCommandHandler(SampleMessageCommand{}, func(command cqrs.Command) error {
-		log.Println("Received Command : ", command.Body.(SampleMessageCommand).Message)
+		cqrs.PackageLogger().Debugf("Received Command : ", command.Body.(SampleMessageCommand).Message)
 		success = true
 		return nil
 	})
 
-	dispatcher.DispatchCommand(cqrs.Command{Body: SampleMessageCommand{"Hello world"}})
+	err := dispatcher.DispatchCommand(cqrs.Command{Body: SampleMessageCommand{"Hello world"}})
+	if err != nil {
+		t.Fatal(err)
+	}
 	if !success {
 		t.Fatal("Expected success")
 	}

correlation.go

@@ -2,24 +2,29 @@ package cqrs
 
 import (
 	"time"
-
-	"github.com/pborman/uuid"
 )
 
-const CQRSErrorEventType = "cqrs.CQRSErrorEvent"
+// CQRSErrorEventType ...
+const CQRSErrorEventType = "cqrs.ErrorEvent"
 
-// CQRSErrorEvent is a generic event raised within the CQRS framework
-type CQRSErrorEvent struct {
+// ErrorEvent is a generic event raised within the CQRS framework
+type ErrorEvent struct {
 	Message string
 }
 
+// DeliverCQRSError will deliver a CQRS error
 func DeliverCQRSError(correlationID string, err error, repo EventSourcingRepository) {
-	repo.GetEventStreamRepository().SaveIntegrationEvent(VersionedEvent{
-		ID:            uuid.New(),
+	err = repo.GetEventStreamRepository().SaveIntegrationEvent(VersionedEvent{
+		ID:            "ve:" + NewUUIDString(),
 		CorrelationID: correlationID,
 		SourceID:      "",
 		Version:       0,
 		EventType:     CQRSErrorEventType,
 		Created:       time.Now(),
-		Event:         CQRSErrorEvent{Message: err.Error()}})
+
+		Event: ErrorEvent{Message: err.Error()}})
+
+	if err != nil {
+		PackageLogger().Debugf("ERROR saving integration event: %v\n", err)
+	}
 }

cqrs-testentities_test.go

@@ -3,9 +3,10 @@ package cqrs_test
 import (
 	"errors"
 	"fmt"
-	"log"
+	"reflect"
 
 	"github.com/andrewwebber/cqrs"
+
 	"golang.org/x/crypto/bcrypt"
 )
 
@@ -72,6 +73,28 @@ type Account struct {
 	Balance      float64
 }
 
+func (account *Account) CopyFrom(source interface{}) {
+
+	account.FirstName = reflect.Indirect(reflect.ValueOf(source)).FieldByName("FirstName").Interface().(string)
+	account.LastName = reflect.Indirect(reflect.ValueOf(source)).FieldByName("LastName").Interface().(string)
+	account.EmailAddress = reflect.Indirect(reflect.ValueOf(source)).FieldByName("EmailAddress").Interface().(string)
+	account.PasswordHash = reflect.Indirect(reflect.ValueOf(source)).FieldByName("PasswordHash").Interface().([]byte)
+	account.Balance = reflect.Indirect(reflect.ValueOf(source)).FieldByName("Balance").Interface().(float64)
+
+	/*cqrs.PackageLogger().Debugf("valueOfSource", valueOfSource)
+	fieldValue := valueOfSource
+	cqrs.PackageLogger().Debugf("fieldValue", fieldValue)
+	fieldValueInterface := fieldValue
+	cqrs.PackageLogger().Debugf("fieldValueInterface", fieldValueInterface)
+	firstName := fieldValueInterface.(string)
+	cqrs.PackageLogger().Debugf("firstName", firstName)*/
+
+	/*account.LastName = reflect.Indirect(reflect.ValueOf(source).FieldByName("LastName")).Interface().(string)
+	account.EmailAddress = reflect.Indirect(reflect.ValueOf(source).FieldByName("EmailAddress")).Interface().(string)
+	account.PasswordHash = reflect.Indirect(reflect.ValueOf(source).FieldByName("PasswordHash")).Interface().([]byte)
+	account.Balance = reflect.Indirect(reflect.ValueOf(source).FieldByName("FirstName")).Interface().(float64)*/
+}
+
 func (account *Account) String() string {
 	return fmt.Sprintf("Account %s with Email Address %s has balance %f", account.ID(), account.EmailAddress, account.Balance)
 }
@@ -89,10 +112,20 @@ func NewAccountFromHistory(id string, repository cqrs.EventSourcingRepository) (
 	account := new(Account)
 	account.EventSourceBased = cqrs.NewEventSourceBasedWithID(account, id)
 
-	if error := repository.Get(id, account); error != nil {
-		return account, error
+	snapshot, err := repository.GetSnapshot(id)
+	if err == nil {
+		cqrs.PackageLogger().Debugf("Loaded snapshot: %+v", snapshot)
+		account.SetVersion(snapshot.Version())
+		account.CopyFrom(snapshot)
+		cqrs.PackageLogger().Debugf("Updated account: %+v ", account)
 	}
 
+	if err := repository.Get(id, account); err != nil {
+		return nil, err
+	}
+
+	cqrs.PackageLogger().Debugf("Loaded account: %+v", account)
+
 	return account, nil
 }
 
@@ -134,7 +167,7 @@ func (account *Account) ChangePassword(newPassword string) error {
 
 	hashedPassword, err := GetHashForPassword(newPassword)
 	if err != nil {
-		panic(err)
+		return (err)
 	}
 
 	account.Update(PasswordChangedEvent{hashedPassword})
@@ -147,7 +180,7 @@ func GetHashForPassword(password string) ([]byte, error) {
 	// Hashing the password with the cost of 10
 	hashedPassword, err := bcrypt.GenerateFromPassword(passwordBytes, 10)
 	if err != nil {
-		log.Println("Error getting password hash: ", err)
+		cqrs.PackageLogger().Debugf("Error getting password hash: ", err)
 		return nil, err
 	}