Merge pull request #62 from wework/v1.x

Merge v1.x -> master

Author: Guy Baron

gbus/bus.go

@@ -323,18 +323,20 @@ func (b *DefaultBus) Shutdown() (shutdwonErr error) {
 			b.log().WithError(err).Error("could not stop worker")
 		}
 	}
-
 	b.Outgoing.shutdown()
 	b.started = false
-
 	if b.IsTxnl {
+
 		err := b.Outbox.Stop()
+
 		if err != nil {
 			b.log().WithError(err).Error("could not shutdown outbox")
 			return err
 		}
-    b.TxProvider.Dispose()
+		b.TxProvider.Dispose()
+
 	}
+
 	return nil
 }
 
@@ -639,7 +641,7 @@ func (b *DefaultBus) sendImpl(sctx context.Context, tx *sql.Tx, toService, reply
 		//send to the transactional outbox if the bus is transactional
 		//otherwise send directly to amqp
 		if b.IsTxnl && tx != nil {
-			b.log().WithField("message_id", msg.MessageId).Info("sending message to outbox")
+			b.log().WithField("message_id", msg.MessageId).Debug("sending message to outbox")
 			saveErr := b.Outbox.Save(tx, exchange, key, msg)
 			if saveErr != nil {
 				log.WithError(saveErr).Error("failed to save to transactional outbox")

gbus/tx/mysql/txoutbox.go

@@ -88,7 +88,7 @@ func (outbox *TxOutbox) Start(amqpOut *gbus.AMQPOutbox) error {
 
 //Stop forcess the transactional outbox to stop processing additional messages
 func (outbox *TxOutbox) Stop() error {
-	outbox.exit <- true
+	close(outbox.exit)
 	return nil
 }
 
@@ -189,7 +189,7 @@ func (outbox *TxOutbox) updateAckedRecord(deliveryTag uint64) error {
 		outbox.log().WithError(txErr).WithField("delivery_tag", deliveryTag).Error("failed to create transaction for updating acked delivery tag")
 		return txErr
 	}
-	outbox.log().WithField("delivery_tag", deliveryTag).Info("ack received for delivery tag")
+	outbox.log().WithField("delivery_tag", deliveryTag).Debug("ack received for delivery tag")
 
 	outbox.gl.Lock()
 	recID := outbox.recordsPendingConfirms[deliveryTag]
@@ -268,24 +268,26 @@ func (outbox *TxOutbox) sendMessages(recordSelector func(tx *sql.Tx) (*sql.Rows,
 			outbox.log().WithError(decErr).Error("failed to decode amqp message from outbox record")
 			continue
 		}
-		outbox.log().WithField("message_id", publishing.MessageId).Info("relay message")
 
 		//send the amqp message to rabbitmq
 		if deliveryTag, postErr := outbox.amqpOutbox.Post(exchange, routingKey, publishing); postErr != nil {
-
 			outbox.log().WithError(postErr).
 				WithFields(log.Fields{"message_name": publishing.Headers["x-msg-name"], "message_id": publishing.MessageId}).
 				Error("failed to send amqp message")
 			failedDeliveries = append(failedDeliveries, recID)
 		} else {
+			outbox.log().WithFields(log.Fields{"message_id": publishing.MessageId, "delivery_tag": deliveryTag}).Debug("relay message")
 			successfulDeliveries[deliveryTag] = recID
 		}
+
 	}
 	err := rows.Close()
 	if err != nil {
 		outbox.log().WithError(err).Error("could not close Rows")
 	}
-
+	if messagesSent := len(successfulDeliveries); messagesSent > 0 {
+		outbox.log().WithField("messages_sent", len(successfulDeliveries)).Info("outbox relayed messages")
+	}
 	for deliveryTag, id := range successfulDeliveries {
 		_, updateErr := tx.Exec("UPDATE "+getOutboxName(outbox.svcName)+" SET status=1, delivery_tag=?, relay_id=? WHERE rec_id=?", deliveryTag, outbox.ID, id)
 		if updateErr != nil {

gbus/worker.go

@@ -179,28 +179,29 @@ func (worker *worker) resolveHandlers(isRPCreply bool, bm *BusMessage, delivery
 	} else {
 		worker.handlersLock.Lock()
 		defer worker.handlersLock.Unlock()
-		worker.log().WithFields(log.Fields{"number_of_handlers": len(worker.registrations)}).Info("found message handlers")
+
 		for _, registration := range worker.registrations {
 			if registration.Matches(delivery.Exchange, delivery.RoutingKey, bm.PayloadFQN) {
 				handlers = append(handlers, registration.Handler)
 			}
 		}
 	}
 
+	worker.log().WithFields(log.Fields{"number_of_handlers": len(handlers)}).Info("found message handlers")
 	return handlers
 }
 
 func (worker *worker) ack(delivery amqp.Delivery) error {
 	ack := func(attempts uint) error { return delivery.Ack(false /*multiple*/) }
-	worker.log().WithField("message_id", delivery.MessageId).Info("acking message")
+	worker.log().WithField("message_id", delivery.MessageId).Debug("acking message")
 	err := retry.Retry(ack,
 		strategy.Wait(100*time.Millisecond))
 
 	if err != nil {
 		worker.log().WithError(err).Error("could not ack the message")
 		worker.span.LogFields(slog.Error(err))
 	} else {
-		worker.log().WithField("message_id", delivery.MessageId).Info("message acked")
+		worker.log().WithField("message_id", delivery.MessageId).Debug("message acked")
 	}
 
 	return err