Merge pull request #27 from wework/cleaner-code

Cleaner code

Author: Guy Baron

gbus/abstractions.go

@@ -8,6 +8,13 @@ import (
 	"github.com/streadway/amqp"
 )
 
+type Semantics string
+
+const (
+	CMD Semantics = "cmd"
+	EVT Semantics = "evt"
+)
+
 //Bus interface provides the majority of functionality to Send, Reply and Publish messages to the Bus
 type Bus interface {
 	HandlerRegister

gbus/bus.go

@@ -66,7 +66,8 @@ var (
 
 	//MaxRetryCount defines the max times a retry can run
 	MaxRetryCount uint = 3
-	rpcHeaderName      = "x-grabbit-msg-rpc-id"
+	//RpcHeaderName used to define the header in grabbit for RPC
+	RpcHeaderName = "x-grabbit-msg-rpc-id"
 )
 
 func (b *DefaultBus) createRPCQueue() (amqp.Queue, error) {
@@ -149,13 +150,13 @@ func (b *DefaultBus) bindServiceQueue() error {
 		if e != nil {
 			b.log().WithError(e).WithField("exchange", exchange).Error("failed to declare exchange")
 			return e
-		} else {
-			e = b.bindQueue(topic, exchange)
-			if e != nil {
-				b.log().WithError(e).WithFields(log.Fields{"topic": topic, "exchange": exchange}).Error("failed to bind topic to exchange")
-				return e
-			}
 		}
+		e = b.bindQueue(topic, exchange)
+		if e != nil {
+			b.log().WithError(e).WithFields(log.Fields{"topic": topic, "exchange": exchange}).Error("failed to bind topic to exchange")
+			return e
+		}
+
 	}
 	return nil
 }
@@ -173,7 +174,7 @@ func (b *DefaultBus) Start() error {
 
 	var e error
 	//create amqo connection and channel
-	if b.amqpConn, e = b.connect(int(MaxRetryCount)); e != nil {
+	if b.amqpConn, e = b.connect(MaxRetryCount); e != nil {
 		return e
 	}
 
@@ -436,7 +437,7 @@ func (b *DefaultBus) RPC(ctx context.Context, service string, request, reply *Bu
 	b.RPCHandlers[rpcID] = handler
 	//we do not defer this as we do not want b.RPCHandlers to be locked until a reply returns
 	b.RPCLock.Unlock()
-	request.Semantics = "cmd"
+	request.Semantics = CMD
 	rpc := rpcPolicy{
 		rpcID: rpcID}
 
@@ -467,7 +468,7 @@ func (b *DefaultBus) publishWithTx(ctx context.Context, ambientTx *sql.Tx, excha
 	if !b.started {
 		return errors.New("bus not strated or already shutdown, make sure you call bus.Start() before sending messages")
 	}
-	message.Semantics = "evt"
+	message.Semantics = EVT
 	publish := func(tx *sql.Tx) error {
 		return b.sendImpl(ctx, tx, "", b.SvcName, exchange, topic, message, policies...)
 	}
@@ -478,7 +479,7 @@ func (b *DefaultBus) sendWithTx(ctx context.Context, ambientTx *sql.Tx, toServic
 	if !b.started {
 		return errors.New("bus not strated or already shutdown, make sure you call bus.Start() before sending messages")
 	}
-	message.Semantics = "cmd"
+	message.Semantics = CMD
 	send := func(tx *sql.Tx) error {
 		return b.sendImpl(ctx, tx, toService, b.SvcName, "", "", message, policies...)
 	}
@@ -543,20 +544,15 @@ func (b *DefaultBus) RegisterSaga(saga Saga, conf ...SagaConfFn) error {
 
 }
 
-func (b *DefaultBus) connect(retryCount int) (*amqp.Connection, error) {
+func (b *DefaultBus) connect(retryCount uint) (*amqp.Connection, error) {
+	var conn *amqp.Connection
+	err := b.SafeWithRetries(func() error {
+		var err error
+		conn, err = amqp.Dial(b.AmqpConnStr)
+		return err
+	}, retryCount)
+	return conn, err
 
-	connected := false
-	attempts := uint(0)
-	var lastErr error
-	for !connected && attempts < MaxRetryCount {
-		conn, e := amqp.Dial(b.AmqpConnStr)
-		if e == nil {
-			return conn, e
-		}
-		lastErr = e
-		attempts++
-	}
-	return nil, lastErr
 }
 
 //
@@ -633,7 +629,7 @@ func (b *DefaultBus) sendImpl(sctx context.Context, tx *sql.Tx, toService, reply
 
 	key := ""
 
-	if message.Semantics == "cmd" {
+	if message.Semantics == CMD {
 		key = toService
 	} else {
 		key = topic
@@ -695,5 +691,5 @@ type rpcPolicy struct {
 }
 
 func (p rpcPolicy) Apply(publishing *amqp.Publishing) {
-	publishing.Headers[rpcHeaderName] = p.rpcID
+	publishing.Headers[RpcHeaderName] = p.rpcID
 }

gbus/messages.go

@@ -12,7 +12,7 @@ type BusMessage struct {
 	CorrelationID     string
 	SagaID            string
 	SagaCorrelationID string
-	Semantics         string /*cmd or evt*/
+	Semantics         Semantics /*cmd or evt*/
 	Payload           Message
 	PayloadFQN        string
 	RPCID             string
@@ -68,7 +68,7 @@ func (bm *BusMessage) GetTraceLog() (fields []log.Field) {
 		log.String("SagaID", bm.SagaID),
 		log.String("CorrelationID", bm.CorrelationID),
 		log.String("SagaCorrelationID", bm.SagaCorrelationID),
-		log.String("Semantics", bm.Semantics),
+		log.String("Semantics", string(bm.Semantics)),
 		log.String("RPCID", bm.RPCID),
 	}
 }

gbus/saga/glue.go

@@ -163,7 +163,7 @@ func (imsm *Glue) handler(invocation gbus.Invocation, message *gbus.BusMessage)
 
 			return imsm.completeOrUpdateSaga(invocation.Tx(), instance, message)
 
-		} else if message.Semantics == "cmd" {
+		} else if message.Semantics == gbus.CMD {
 			e := fmt.Errorf("Warning:Command or Reply message with no saga reference received. message will be dropped.\nmessage as of type:%v", reflect.TypeOf(message).Name())
 			return e
 		} else {
@@ -207,7 +207,7 @@ func (imsm *Glue) completeOrUpdateSaga(tx *sql.Tx, instance *Instance, lastMessa
 }
 
 func (imsm *Glue) registerMessage(message gbus.Message) error {
-	//only register once on each message so we will not duplicate invokations
+	//only register once on each message so we will not duplicate invocations
 	if _, exists := imsm.alreadyRegistred[message.SchemaName()]; exists {
 		return nil
 	}

gbus/saga/invocation.go

@@ -49,17 +49,20 @@ func (si *sagaInvocation) Ctx() context.Context {
 	return si.ctx
 }
 
-func (si *sagaInvocation) Send(ctx context.Context, toService string, command *gbus.BusMessage, policies ...gbus.MessagePolicy) error {
+func (si *sagaInvocation) Send(ctx context.Context, toService string,
+	command *gbus.BusMessage, policies ...gbus.MessagePolicy) error {
 	si.setCorrelationIDs(command, false)
 	return si.decoratedBus.Send(ctx, toService, command, policies...)
 }
 
-func (si *sagaInvocation) Publish(ctx context.Context, exchange, topic string, event *gbus.BusMessage, policies ...gbus.MessagePolicy) error {
+func (si *sagaInvocation) Publish(ctx context.Context, exchange, topic string,
+	event *gbus.BusMessage, policies ...gbus.MessagePolicy) error {
 	si.setCorrelationIDs(event, true)
 	return si.decoratedBus.Publish(ctx, exchange, topic, event, policies...)
 }
 
-func (si *sagaInvocation) RPC(ctx context.Context, service string, request, reply *gbus.BusMessage, timeout time.Duration) (*gbus.BusMessage, error) {
+func (si *sagaInvocation) RPC(ctx context.Context, service string, request,
+	reply *gbus.BusMessage, timeout time.Duration) (*gbus.BusMessage, error) {
 	return si.decoratedBus.RPC(ctx, service, request, reply, timeout)
 }
 

gbus/serialization/avro.go

@@ -167,7 +167,7 @@ func (as *Avro) RegisterAvroMessageFromFile(schemaName, schemaPath, namespace st
 }
 
 //RegisterAvroMessage registers a schema to a topic and binds it to an object (obj)
-func (as *Avro) RegisterAvroMessage(schemaName, namespace, schema string, obj AvroMessageGenerated, deserializer avroDeserializer) (err error) {
+func (as *Avro) RegisterAvroMessage(schemaName, namespace, schema string, obj gbus.Message, deserializer avroDeserializer) (err error) {
 	as.lock.Lock()
 	defer as.lock.Unlock()
 	if _, ok := as.registeredSchemas[obj.SchemaName()]; !ok {

gbus/tx/mysql/txoutbox.go

@@ -119,7 +119,7 @@ func (outbox *TxOutbox) Save(tx *sql.Tx, exchange, routingKey string, amqpMessag
 
 func (outbox *TxOutbox) purge(tx *sql.Tx) error {
 
-	purgeSQL := `DELETE FROM  ` + getOutboxName(outbox.svcName)
+	purgeSQL := fmt.Sprintf("DELETE FROM %s", getOutboxName(outbox.svcName))
 	_, err := tx.Exec(purgeSQL)
 	return err
 }
@@ -224,8 +224,8 @@ func (outbox *TxOutbox) getMessageRecords(tx *sql.Tx) (*sql.Rows, error) {
 }
 
 func (outbox *TxOutbox) scavengeOrphanedRecords(tx *sql.Tx) (*sql.Rows, error) {
-	selectSQL := "SELECT rec_id, exchange, routing_key, publishing FROM " + getOutboxName(outbox.svcName) + " WHERE status = 1  ORDER BY rec_id ASC LIMIT " + strconv.Itoa(maxPageSize) + " FOR UPDATE SKIP LOCKED"
-	return tx.Query(selectSQL)
+	selectSQL := "SELECT rec_id, exchange, routing_key, publishing FROM " + getOutboxName(outbox.svcName) + " WHERE status = 1  ORDER BY rec_id ASC LIMIT ? FOR UPDATE SKIP LOCKED"
+	return tx.Query(selectSQL, strconv.Itoa(maxPageSize))
 }
 
 func (outbox *TxOutbox) sendMessages(recordSelector func(tx *sql.Tx) (*sql.Rows, error)) error {
@@ -322,7 +322,7 @@ func (outbox *TxOutbox) ensureSchema(tx *sql.Tx, svcName string) error {
 
 	if schemaExists {
 		/*
-			The follwoing  performs an alter schema to accommodate for breaking change introduced in commit 6a9f5df
+			The following  performs an alter schema to accommodate for breaking change introduced in commit 6a9f5df
 			so that earlier consumers of grabbit will not break once the upgrade to the 1.0.0 release.
 			Once a proper DB migration stratagy will be in place and implemented (post 1.0.0) the following code
 			will be deleted.

gbus/tx/sagastore.go

@@ -115,7 +115,7 @@ func (store *SagaStore) RegisterSagaType(saga gbus.Saga) {
 //DeleteSaga implements interface method store.DeleteSaga
 func (store *SagaStore) DeleteSaga(tx *sql.Tx, instance *saga.Instance) error {
 	tblName := store.GetSagatableName()
-	deleteSQL := `DELETE FROM ` + tblName + ` WHERE saga_id=` + store.ParamsMarkers[0] + ``
+	deleteSQL := `DELETE FROM ` + tblName + ` WHERE saga_id= ?`
 	_, err := tx.Exec(deleteSQL, instance.ID)
 	return err
 }
@@ -154,8 +154,7 @@ func (store *SagaStore) GetSagaByID(tx *sql.Tx, sagaID string) (*saga.Instance,
 func (store *SagaStore) SaveNewSaga(tx *sql.Tx, sagaType reflect.Type, newInstance *saga.Instance) (err error) {
 	store.RegisterSagaType(newInstance.UnderlyingInstance)
 	tblName := store.GetSagatableName()
-	insertSQL := `INSERT INTO ` + tblName + ` (saga_id, saga_type, saga_data, version)
-	VALUES (` + store.ParamsMarkers[0] + `, ` + store.ParamsMarkers[1] + `, ` + store.ParamsMarkers[2] + `, ` + store.ParamsMarkers[3] + `)`
+	insertSQL := `INSERT INTO ` + tblName + ` (saga_id, saga_type, saga_data, version) VALUES (?, ?, ?, ?)`
 
 	var buf []byte
 	if buf, err = store.serilizeSaga(newInstance); err != nil {
@@ -174,7 +173,8 @@ func (store *SagaStore) SaveNewSaga(tx *sql.Tx, sagaType reflect.Type, newInstan
 func (store *SagaStore) Purge() error {
 	tx := store.NewTx()
 	store.log().WithField("saga_table", store.GetSagatableName()).Info("Purging saga table")
-	results, err := tx.Exec("DELETE FROM  " + store.GetSagatableName())
+	deleteSQL := fmt.Sprintf("DELETE FROM %s", store.GetSagatableName())
+	results, err := tx.Exec(deleteSQL)
 	if err != nil {
 		store.log().WithError(err).Error("failed to purge saga table")
 		return err

gbus/worker.go

@@ -137,9 +137,9 @@ func (worker *worker) extractBusMessage(delivery amqp.Delivery) (*BusMessage, er
 	bm.ID = delivery.MessageId
 	bm.CorrelationID = delivery.CorrelationId
 	if delivery.Exchange != "" {
-		bm.Semantics = "evt"
+		bm.Semantics = EVT
 	} else {
-		bm.Semantics = "cmd"
+		bm.Semantics = CMD
 	}
 	if bm.PayloadFQN == "" || bm.Semantics == "" {
 		//TODO: Log poision pill message
@@ -160,7 +160,7 @@ func (worker *worker) extractBusMessage(delivery amqp.Delivery) (*BusMessage, er
 func (worker *worker) resolveHandlers(isRPCreply bool, bm *BusMessage, delivery amqp.Delivery) []MessageHandler {
 	handlers := make([]MessageHandler, 0)
 	if isRPCreply {
-		rpcID, rpcHeaderFound := delivery.Headers[rpcHeaderName].(string)
+		rpcID, rpcHeaderFound := delivery.Headers[RpcHeaderName].(string)
 		if !rpcHeaderFound {
 			worker.log().Warn("rpc message received but no rpc header found...rejecting message")
 			return handlers