merge v1.x into master

README.md

@@ -2,6 +2,9 @@
 [![CircleCI](https://circleci.com/gh/wework/grabbit.svg?style=svg)](https://circleci.com/gh/wework/grabbit)
 [![Go Report Card](https://goreportcard.com/badge/github.com/wework/grabbit)](https://goreportcard.com/report/github.com/wework/grabbit)
 [![Coverage Status](https://coveralls.io/repos/github/wework/grabbit/badge.svg?branch=master)](https://coveralls.io/github/wework/grabbit?branch=master)
+![GitHub release](https://img.shields.io/github/release/wework/grabbit.svg)
+
+
 
 # grabbit
 
@@ -141,5 +144,6 @@ if e != nil{
 
 ## Testing
 
+0) ensure that you have the dependencies installed: `go get -v -t -d ./...`
 1) make sure to first: `docker-compose up -V -d`
 2) then to run the tests: `go test ./...`

gbus/abstractions.go

@@ -3,6 +3,7 @@ package gbus
 import (
 	"context"
 	"database/sql"
+	"github.com/sirupsen/logrus"
 	"time"
 
 	"github.com/streadway/amqp"
@@ -17,7 +18,7 @@ const (
 
 //BusConfiguration provides configuration passed to the bus builder
 type BusConfiguration struct {
-	MaxRetryCount uint
+	MaxRetryCount     uint
 	BaseRetryDuration int
 }
 
@@ -29,6 +30,7 @@ type Bus interface {
 	Messaging
 	SagaRegister
 	Health
+	Logged
 }
 
 //Message a common interface that passes to the serializers to allow decoding and encoding of content
@@ -177,10 +179,14 @@ type Builder interface {
 
 	//Build the bus
 	Build(svcName string) Bus
+
+	//WithLogger set custom logger instance
+	WithLogger(logger logrus.FieldLogger) Builder
 }
 
 //Invocation context for a specific processed message
 type Invocation interface {
+	Logged
 	Reply(ctx context.Context, message *BusMessage) error
 	Bus() Messaging
 	Tx() *sql.Tx
@@ -209,3 +215,8 @@ type TxOutbox interface {
 	Start(amqpOut *AMQPOutbox) error
 	Stop() error
 }
+
+type Logged interface {
+	SetLogger(entry logrus.FieldLogger)
+	Log() logrus.FieldLogger
+}

gbus/builder/builder.go

@@ -2,6 +2,7 @@ package builder
 
 import (
 	"fmt"
+	"github.com/sirupsen/logrus"
 	"sync"
 	"time"
 
@@ -27,13 +28,14 @@ type defaultBuilder struct {
 	confirm          bool
 	dbPingTimeout    time.Duration
 	usingPingTimeout bool
+	logger           logrus.FieldLogger
 }
 
 func (builder *defaultBuilder) Build(svcName string) gbus.Bus {
 
 	gb := &gbus.DefaultBus{
 		AmqpConnStr:   builder.connStr,
-		PrefetchCount: 1,
+		PrefetchCount: builder.PrefetchCount,
 		Outgoing: &gbus.AMQPOutbox{
 			SvcName: svcName,
 		},
@@ -50,15 +52,21 @@ func (builder *defaultBuilder) Build(svcName string) gbus.Bus {
 		Serializer:           builder.serializer,
 		DLX:                  builder.dlx,
 		DefaultPolicies:      builder.defaultPolicies,
-		DbPingTimeout:        3}
+		DbPingTimeout:        3,
+		Confirm:              builder.confirm,
+	}
+
+	if builder.logger != nil {
+		gb.SetLogger(builder.logger)
+	} else {
+		gb.SetLogger(logrus.New())
+	}
 
-	gb.Confirm = builder.confirm
 	if builder.workerNum < 1 {
 		gb.WorkerNum = 1
 	} else {
 		gb.WorkerNum = builder.workerNum
 	}
-	gb.PrefetchCount = builder.PrefetchCount
 	var (
 		sagaStore saga.Store
 	)
@@ -175,11 +183,16 @@ func (builder *defaultBuilder) WithConfiguration(config gbus.BusConfiguration) g
 		gbus.MaxRetryCount = config.MaxRetryCount
 	}
 	if config.BaseRetryDuration > 0 {
-		gbus.BaseRetryDuration = time.Millisecond*time.Duration(config.BaseRetryDuration)
+		gbus.BaseRetryDuration = time.Millisecond * time.Duration(config.BaseRetryDuration)
 	}
 	return builder
 }
 
+func (builder *defaultBuilder) WithLogger(logger logrus.FieldLogger) gbus.Builder {
+	builder.logger = logger
+	return builder
+}
+
 //New :)
 func New() Nu {
 	return Nu{}