Merge pull request #7 from hellofresh/feature/rm-logrus-dep

Removed logrus dependency from the project

Author: vgarvardt

README.md

@@ -26,7 +26,7 @@ Engine is divided in a few small independent components.
 ## Install
 
 ```sh
-go get github.com/hellofresh/goengine
+go get -u github.com/hellofresh/goengine
 ```
 
 ## Usage
@@ -35,6 +35,42 @@ Here you can check a small tutorial of how to use this component in an orders sc
 
 [Tutorial](docs/how_to.md)
 
+## Logging
+
+GoEngine uses default `log` package for debug logging. If you want to use your own logger - `goengine.SetLogHandler()`
+is available. Here is how you can use, e.g. `github.com/sirupsen/logrus` for logging:
+
+```go
+package main
+
+import (
+    "github.com/hellofresh/goengine"
+    log "github.com/sirupsen/logrus"
+)
+
+func main() {
+    goengine.SetLogHandler(func(msg string, fields map[string]interface{}, err error) {
+        if nil == fields && nil == err {
+            log.Debug(msg)
+        } else {
+            var entry *log.Entry
+            if fields != nil {
+                entry = log.WithFields(log.Fields(fields))
+                if nil != err {
+                    entry = entry.WithError(err)
+                }
+            } else {
+                entry = log.WithError(err)
+            }
+
+            entry.Debug(msg)
+        }
+    })
+
+    // do your application stuff
+}
+```
+
 ## Contributing
 
 Please see [CONTRIBUTING](CONTRIBUTING.md) for details.

aggregate_repository.go

@@ -2,8 +2,6 @@ package goengine
 
 import (
 	"fmt"
-
-	log "github.com/sirupsen/logrus"
 )
 
 type AggregateRepository interface {
@@ -22,7 +20,7 @@ func NewPublisherRepository(eventStore EventStore, eventBus VersionedEventPublis
 }
 
 func (r *PublisherRepository) Load(id string, streamName StreamName) (*EventStream, error) {
-	log.WithFields(log.Fields{"stream": streamName, "id": id}).Debug("Loading events from stream for aggregate")
+	Log("Loading events from stream for aggregate", map[string]interface{}{"stream": streamName, "id": id}, nil)
 	stream, err := r.EventStore.GetEventsFor(streamName, id)
 	if nil != err {
 		return nil, err
@@ -34,14 +32,15 @@ func (r *PublisherRepository) Load(id string, streamName StreamName) (*EventStre
 func (r *PublisherRepository) Save(aggregateRoot AggregateRoot, streamName StreamName) error {
 	events := aggregateRoot.GetUncommittedEvents()
 	eventStream := NewEventStream(streamName, events)
-	log.WithFields(log.Fields{"count": len(events), "stream": streamName}).Debug("Saving events to stream")
+	Log("Saving events to stream", map[string]interface{}{"count": len(events), "stream": streamName}, nil)
+
 	err := r.EventStore.Append(eventStream)
 	if nil != err {
 		return err
 	}
 
 	if nil == r.EventBus {
-		log.Debug("Event bus not detected, skipping publishing events")
+		Log("Event bus not detected, skipping publishing events", nil, nil)
 		return nil
 	}
 
@@ -53,7 +52,8 @@ func (r *PublisherRepository) Save(aggregateRoot AggregateRoot, streamName Strea
 }
 
 func (r *PublisherRepository) Reconstitute(id string, source AggregateRoot, streamName StreamName) error {
-	log.WithFields(log.Fields{"stream": streamName, "id": id}).Debug("Reconstituting aggregate from stream")
+	Log("Reconstituting aggregate from stream", map[string]interface{}{"stream": streamName, "id": id}, nil)
+
 	stream, err := r.Load(id, streamName)
 	if nil != err {
 		return err
@@ -69,6 +69,6 @@ func (r *PublisherRepository) Reconstitute(id string, source AggregateRoot, stre
 	}
 
 	source.SetVersion(events[len(events)-1].Version)
-	log.WithField("id", id).Debug("Aggregate reconstituted")
+	Log("Aggregate reconstituted", map[string]interface{}{"id": id}, nil)
 	return nil
 }

aggregate_root.go

@@ -5,7 +5,6 @@ import (
 
 	"github.com/hellofresh/goengine/reflection"
 	"github.com/pborman/uuid"
-	log "github.com/sirupsen/logrus"
 )
 
 type AggregateRoot interface {
@@ -48,7 +47,7 @@ func (r *AggregateRootBased) SetVersion(version int) {
 func (r *AggregateRootBased) GetUncommittedEvents() []*DomainMessage {
 	stream := r.uncommittedEvents
 	r.uncommittedEvents = nil
-	log.WithField("count", len(stream)).Debug("Uncommitted events cleaned")
+	Log("Uncommitted events cleaned", map[string]interface{}{"count": len(stream)}, nil)
 
 	return stream
 }
@@ -57,10 +56,10 @@ func (r *AggregateRootBased) Apply(event DomainEvent) {
 	t := reflection.TypeOf(event)
 	methodName := fmt.Sprintf("When%s", t.Name())
 
-	entry := log.WithFields(log.Fields{"source": fmt.Sprintf("%+v", r.source), "method": methodName, "event": fmt.Sprintf("%+v", event)})
-	entry.Debug("Applying event")
+	fields := map[string]interface{}{"source": fmt.Sprintf("%+v", r.source), "method": methodName, "event": fmt.Sprintf("%+v", event)}
+	Log("Applying event", fields, nil)
 	reflection.CallMethod(r.source, methodName, event)
-	entry.Debug("Event applied")
+	Log("Event applied", fields, nil)
 }
 
 func (r *AggregateRootBased) RecordThat(event DomainEvent) {
@@ -72,5 +71,5 @@ func (r *AggregateRootBased) RecordThat(event DomainEvent) {
 func (r *AggregateRootBased) Record(event DomainEvent) {
 	message := RecordNow(r.ID, r.version, event)
 	r.uncommittedEvents = append(r.uncommittedEvents, message)
-	log.Debug("Event recorded")
+	Log("Event recorded", nil, nil)
 }

cmd/goengine/main.go

@@ -6,58 +6,59 @@ import (
 	"github.com/hellofresh/goengine"
 	"github.com/hellofresh/goengine/mongodb"
 	"github.com/hellofresh/goengine/rabbit"
-	log "github.com/sirupsen/logrus"
 	"gopkg.in/mgo.v2"
 )
 
 func main() {
-	log.SetLevel(log.DebugLevel)
 	var streamName goengine.StreamName = "test"
 
 	mongoDSN := os.Getenv("STORAGE_DSN")
-	log.WithField("dsn", mongoDSN).Debug("Connecting to the database")
+	goengine.Log("Connecting to the database", map[string]interface{}{"dsn": mongoDSN}, nil)
 	session, err := mgo.Dial(mongoDSN)
 	if err != nil {
-		log.Panic(err)
+		goengine.Log("Failed to connect to Mongo", nil, err)
+		panic(err)
 	}
 	defer session.Close()
 
 	// Optional. Switch the session to a monotonic behavior.
 	session.SetMode(mgo.Monotonic, true)
 
-	log.Info("Setting up the registry")
+	goengine.Log("Setting up the registry", nil, nil)
 	registry := goengine.NewInMemoryTypeRegistry()
 	registry.RegisterType(&RecipeCreated{})
 	registry.RegisterType(&RecipeRated{})
 
-	log.Info("Setting up the event bus")
 	// bus := inmemory.NewInMemoryEventBus()
-	bus := rabbit.NewEventBus(os.Getenv("BROKER_DSN"), "events", "events")
+	brokerDSN := os.Getenv("BROKER_DSN")
+	goengine.Log("Setting up the event bus", map[string]interface{}{"dsn": brokerDSN}, nil)
+	bus := rabbit.NewEventBus(brokerDSN, "events", "events")
 
-	log.Info("Setting up the event store")
+	goengine.Log("Setting up the event store", nil, nil)
 	es := mongodb.NewEventStore(session, registry)
 
 	eventDispatcher := goengine.NewVersionedEventDispatchManager(bus, registry)
 	eventDispatcher.RegisterEventHandler(&RecipeCreated{}, func(event *goengine.DomainMessage) error {
-		log.Debug("Event received")
+		goengine.Log("Event received", nil, nil)
 		return nil
 	})
 
 	stopChannel := make(chan bool)
 	go eventDispatcher.Listen(stopChannel, false)
 
-	log.Info("Creating a recipe")
+	goengine.Log("Creating a recipe", nil, nil)
 	aggregateRoot := CreateScenario(streamName)
 
 	repository := goengine.NewPublisherRepository(es, bus)
 	repository.Save(aggregateRoot, streamName)
 
 	_, err = NewRecipeFromHisotry(aggregateRoot.ID, streamName, repository)
 	if err != nil {
-		log.Panic(err)
+		goengine.Log("Failed to connect to Mongo", nil, err)
+		panic(err)
 	}
 
-	log.Println("Stop channel")
+	goengine.Log("Stopping channel", nil, err)
 	stopChannel <- true
 }
 

event_dispatcher_manager.go

@@ -1,9 +1,5 @@
 package goengine
 
-import (
-	log "github.com/sirupsen/logrus"
-)
-
 // VersionedEventDispatchManager is responsible for coordinating receiving messages
 // from event receivers and dispatching them to the event dispatcher.
 type VersionedEventDispatchManager struct {
@@ -52,25 +48,25 @@ func (m *VersionedEventDispatchManager) Listen(stop <-chan bool, exclusive bool)
 		// signifying successful processing of the message.
 		// This should eventually call an event handler. See NewVersionedEventDispatcher()
 		case event := <-receiveEventChannel:
-			entry := log.WithField("event", event.Event)
-			entry.Debugf("EventDispatchManager.DispatchEvent")
+			fields := map[string]interface{}{"event": event.Event}
+			Log("EventDispatchManager.DispatchEvent", fields, nil)
 			if err := m.versionedEventDispatcher.DispatchEvent(event.Event); err != nil {
-				entry.WithError(err).Error("Error dispatching event")
+				Log("Error dispatching event", fields, err)
 			}
 
 			event.ProcessedSuccessfully <- true
-			entry.Debug("EventDispatchManager.DispatchSuccessful")
+			Log("EventDispatchManager.DispatchSuccessful", fields, nil)
 
 		case <-stop:
-			log.Debug("EventDispatchManager.Stopping")
+			Log("EventDispatchManager.Stopping", nil, nil)
 			closeSignal := make(chan error)
 			closeChannel <- closeSignal
-			log.Debug("EventDispatchManager.Stopped")
+			Log("EventDispatchManager.Stopped", nil, nil)
 			return <-closeSignal
 
 		// Receiving on this channel signifys an error has occurred worker processor side
 		case err := <-errorChannel:
-			log.WithError(err).Debugf("EventDispatchManager.ErrorReceived")
+			Log("EventDispatchManager.ErrorReceived", nil, err)
 			return err
 		}
 	}

logging.go

@@ -0,0 +1,44 @@
+package goengine
+
+import (
+	"fmt"
+	"log"
+	"os"
+	"strings"
+)
+
+type LogHandler func(msg string, fields map[string]interface{}, err error)
+
+var logHandler LogHandler
+
+func init() {
+	l := log.New(os.Stderr, "[GoEngine] ", log.LstdFlags)
+
+	SetLogHandler(func(msg string, fields map[string]interface{}, err error) {
+		if nil != err {
+			if nil == fields {
+				fields = make(map[string]interface{})
+			}
+			fields["error"] = err.Error()
+		}
+
+		msgParts := make([]string, len(fields)+1)
+
+		msgParts[0] = msg
+		idx := 1
+		for k, v := range fields {
+			msgParts[idx] = fmt.Sprintf("%s=%v", k, v)
+			idx++
+		}
+
+		l.Println(strings.Join(msgParts, "\t"))
+	})
+}
+
+func SetLogHandler(handler LogHandler) {
+	logHandler = handler
+}
+
+func Log(msg string, fields map[string]interface{}, err error) {
+	logHandler(msg, fields, err)
+}

rabbit/eventbus.go

@@ -8,7 +8,6 @@ import (
 
 	"github.com/hellofresh/goengine"
 	"github.com/hellofresh/goengine/reflection"
-	log "github.com/sirupsen/logrus"
 	"github.com/streadway/amqp"
 )
 
@@ -115,29 +114,28 @@ func (bus *EventBus) ReceiveEvents(options goengine.VersionedEventReceiverOption
 					} else {
 						domainEvent, err := bus.fromRawEvent(options.TypeRegistry, &raw)
 						if nil != err {
-							log.Debugf("EventBus.Cannot find event type %s", raw.Type)
+							goengine.Log("EventBus.Cannot find event type", map[string]interface{}{"type": raw.Type}, nil)
 							options.Error <- errors.New("Cannot find event type " + raw.Type)
 							message.Ack(true)
 						} else {
 							ackCh := make(chan bool)
 
-							log.Debug("EventBus.Dispatching Message")
+							goengine.Log("EventBus.Dispatching Message", nil, nil)
 							options.ReceiveEvent <- goengine.VersionedEventTransactedAccept{Event: domainEvent, ProcessedSuccessfully: ackCh}
 							result := <-ackCh
 							message.Ack(result)
 						}
 					}
 				} else {
-					log.Debug("RabbitMQ: Could have been disconnected")
+					goengine.Log("RabbitMQ: Could have been disconnected", nil, nil)
 					for {
 						retryError := exponential(func() error {
 							connR, cR, eventsR, errR := bus.consumeEventsQueue(options.Exclusive)
 							if errR == nil {
 								conn, c, events, err = connR, cR, eventsR, errR
 							}
 
-							log.WithError(err).Debug("Failed to reconnect to RabbitMQ")
-
+							goengine.Log("Failed to reconnect to RabbitMQ", nil, err)
 							return errR
 						}, 5)
 

rabbit/retry.go

@@ -4,7 +4,7 @@ import (
 	"math"
 	"time"
 
-	log "github.com/sirupsen/logrus"
+	"github.com/hellofresh/goengine"
 )
 
 type function func() error
@@ -26,7 +26,7 @@ func exponential(operation function, maxRetries int) error {
 
 		sleepTime := time.Duration(sleepMs) * time.Millisecond
 		time.Sleep(sleepTime)
-		log.WithFields(log.Fields{"attempt": i, "sleep": sleepTime.String()}).Debug("Retry exponential")
+		goengine.Log("Retry exponential", map[string]interface{}{"attempt": i, "sleep": sleepTime.String()}, err)
 	}
 
 	return err