Merge pull request wework#59 from wework/shutdown_fix

Shutdown fix
whugoh · May 1, 2019 · eea9d56 · eea9d56
2 parents 96ebe02 + 7caa110
commit eea9d56
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 7 deletions.
diff --git a/gbus/bus.go b/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")

diff --git a/gbus/tx/mysql/txoutbox.go b/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
 }
 

diff --git a/gbus/worker.go b/gbus/worker.go
@@ -192,15 +192,15 @@ func (worker *worker) resolveHandlers(isRPCreply bool, bm *BusMessage, delivery
 
 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