func (b *DefaultBus) HandleEvent(exchange, topic string, event Message, handler MessageHandler) error {
	if !b.started {
		subscription := make([]string, 0)
		subscription = append(subscription, topic, exchange)
		b.DelayedSubscriptions = append(b.DelayedSubscriptions, subscription)
	} else {
		err := b.bindQueue(topic, exchange)
		if err != nil {
			return err
		}
	}
	return b.registerHandlerImpl(exchange, topic, event, handler)
}
HandleEvent implements GBus.HandleEvent
package gbus
import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"runtime/debug"
	"sync"
	"time"
	"github.com/wework/grabbit/gbus/metrics"
	"github.com/opentracing-contrib/go-amqp/amqptracer"
	"github.com/opentracing/opentracing-go"
	slog "github.com/opentracing/opentracing-go/log"
	"github.com/rs/xid"
	"github.com/sirupsen/logrus"
	"github.com/streadway/amqp"
)
var _ SagaRegister = &DefaultBus{}
var _ Bus = &DefaultBus{}
type DefaultBus struct {
	*Safety
	*Glogged
	Outbox         TxOutbox
	PrefetchCount  uint
	AmqpConnStr    string
	ingressConn    *amqp.Connection
	egressConn     *amqp.Connection
	workers        []*worker
	ingressChannel *amqp.Channel
	egressChannel  *amqp.Channel
	serviceQueue   amqp.Queue
	rpcQueue       amqp.Queue
	SvcName        string
	amqpErrors     chan *amqp.Error
	amqpBlocks     chan amqp.Blocking
	Registrations  []*Registration
	amqpOutbox     *AMQPOutbox
	RPCHandlers          map[string]MessageHandler
	deadletterHandler    RawMessageHandler
	globalRawHandler     RawMessageHandler
	HandlersLock         *sync.Mutex
	RPCLock              *sync.Mutex
	SenderLock           *sync.Mutex
	ConsumerLock         *sync.Mutex
	RegisteredSchemas    map[string]bool
	DelayedSubscriptions [][]string
	PurgeOnStartup       bool
	started              bool
	Glue                 SagaGlue
	TxProvider           TxProvider
	WorkerNum       uint
	Serializer      Serializer
	DLX             string
	DefaultPolicies []MessagePolicy
	Confirm         bool
	healthChan      chan error
	backpressure    bool
	DbPingTimeout   time.Duration
	amqpConnected   bool
}
var (
	MaxRetryCount uint = 3
	BaseRetryDuration = 10 * time.Millisecond
	RPCHeaderName                  = "x-grabbit-msg-rpc-id"
	ResurrectedHeaderName          = "x-resurrected-from-death"
	FirstDeathRoutingKeyHeaderName = "x-first-death-routing-key"
)
func (b *DefaultBus) createRPCQueue() (amqp.Queue, error) {
	uid := xid.New().String()
	qName := b.SvcName + "_rpc_" + uid
	q, e := b.ingressChannel.QueueDeclare(qName,
		false,  
		true,   
		false,  
		false,  
		nil  )
	return q, e
}
func (b *DefaultBus) createServiceQueue() (amqp.Queue, error) {
	qName := b.SvcName
	var q amqp.Queue
	if b.PurgeOnStartup {
		msgsPurged, purgeError := b.ingressChannel.QueueDelete(qName, false  , false  , false  )
		if purgeError != nil {
			b.Log().WithError(purgeError).WithField("deleted_messages", msgsPurged).Error("failed to purge queue")
			return q, purgeError
		}
	}
	args := amqp.Table{}
	if b.DLX != "" {
		args["x-dead-letter-exchange"] = b.DLX
	}
	q, e := b.ingressChannel.QueueDeclare(qName,
		true,   
		false,  
		false,  
		false,  
		args  )
	if e != nil {
		b.Log().WithError(e).Error("failed to declare queue")
	}
	b.serviceQueue = q
	return q, e
}
func (b *DefaultBus) bindServiceQueue() error {
	if b.deadletterHandler != nil && b.DLX != "" {
		err := b.ingressChannel.ExchangeDeclare(b.DLX,  
			"fanout",  
			true,      
			false,     
			false,     
			false,     
			nil  )
		if err != nil {
			b.Log().WithError(err).Error("could not declare exchange")
			return err
		}
		err = b.bindQueue("", b.DLX)
		if err != nil {
			b.Log().WithError(err).Error("could not bind exchange")
			return err
		}
	}
	for _, subscription := range b.DelayedSubscriptions {
		topic := subscription[0]
		exchange := subscription[1]
		e := b.ingressChannel.ExchangeDeclare(exchange,  
			"topic",  
			true,     
			false,    
			false,    
			false,    
			nil  )
		if e != nil {
			b.Log().WithError(e).WithField("exchange", exchange).Error("failed to declare exchange")
			return e
		}
		e = b.bindQueue(topic, exchange)
		if e != nil {
			b.Log().WithError(e).WithFields(logrus.Fields{"topic": topic, "exchange": exchange}).Error("failed to bind topic to exchange")
			return e
		}
	}
	return nil
}
func (b *DefaultBus) Start() error {
	var e error
	if b.ingressConn, e = b.connect(MaxRetryCount); e != nil {
		return e
	}
	if b.egressConn, e = b.connect(MaxRetryCount); e != nil {
		return e
	}
	if b.ingressChannel, e = b.ingressConn.Channel(); e != nil {
		return e
	}
	if b.egressChannel, e = b.egressConn.Channel(); e != nil {
		return e
	}
	b.amqpErrors = make(chan *amqp.Error)
	b.amqpBlocks = make(chan amqp.Blocking)
	b.ingressConn.NotifyClose(b.amqpErrors)
	b.ingressConn.NotifyBlocked(b.amqpBlocks)
	b.egressConn.NotifyClose(b.amqpErrors)
	b.egressConn.NotifyBlocked(b.amqpBlocks)
	b.egressChannel.NotifyClose(b.amqpErrors)
	var amqpChan *amqp.Channel
	if amqpChan, e = b.egressConn.Channel(); e != nil {
		b.Log().WithError(e).Error("failed to create amqp channel for transactional outbox")
		return e
	}
	amqpChan.NotifyClose(b.amqpErrors)
	b.amqpOutbox = &AMQPOutbox{
		SvcName: b.SvcName,
	}
	err := b.amqpOutbox.init(amqpChan, b.Confirm, false)
	if err != nil {
		b.Log().WithError(err).Error("failed initializing amqpOutbox")
		return err
	}
	if startErr := b.Outbox.Start(b.amqpOutbox); startErr != nil {
		b.Log().WithError(startErr).Error("failed to start transactional outbox")
		return startErr
	}
	var q amqp.Queue
	if q, e = b.createServiceQueue(); e != nil {
		return e
	}
	b.serviceQueue = q
	bindErr := b.bindServiceQueue()
	if bindErr != nil {
		b.Log().WithError(err).Error("could not bind service to queue")
		return bindErr
	}
	if b.rpcQueue, e = b.createRPCQueue(); e != nil {
		return e
	}
	b.Log().WithField("number_of_workers", b.WorkerNum).Info("initiating workers")
	workers, createWorkersErr := b.createBusWorkers(b.WorkerNum)
	if createWorkersErr != nil {
		b.Log().WithError(createWorkersErr).Error("error creating channel for worker")
		return createWorkersErr
	}
	if err := b.Glue.Start(); err != nil {
		return err
	}
	b.workers = workers
	b.started = true
	go b.monitorAMQPErrors()
	b.amqpConnected = true
	return nil
}
func (b *DefaultBus) createBusWorkers(workerNum uint) ([]*worker, error) {
	workers := make([]*worker, 0)
	for i := uint(0); i < workerNum; i++ {
		amqpChan, createChanErr := b.ingressConn.Channel()
		if createChanErr != nil {
			return nil, createChanErr
		}
		qosErr := amqpChan.Qos(int(b.PrefetchCount), 0, false)
		if qosErr != nil {
			b.Log().Printf("failed to set worker qos\n %v", qosErr)
		}
		tag := fmt.Sprintf("%s_worker_%d", b.SvcName, i)
		w := &worker{
			consumerTag:       tag,
			channel:           amqpChan,
			q:                 b.serviceQueue,
			rpcq:              b.rpcQueue,
			svcName:           b.SvcName,
			txProvider:        b.TxProvider,
			rpcLock:           b.RPCLock,
			rpcHandlers:       b.RPCHandlers,
			deadletterHandler: b.deadletterHandler,
			globalRawHandler:  b.globalRawHandler,
			handlersLock:      &sync.Mutex{},
			registrations:     b.Registrations,
			serializer:        b.Serializer,
			b:                 b,
			amqpErrors:        b.amqpErrors}
		err := w.Start()
		if err != nil {
			b.Log().WithError(err).Error("failed to start worker")
		}
		workers = append(workers, w)
	}
	return workers, nil
}
func (b *DefaultBus) Shutdown() (shutdwonErr error) {
	b.Log().Info("Bus shuting down")
	defer func() {
		if p := recover(); p != nil {
			pncMsg := fmt.Sprintf("%v\n%s", p, debug.Stack())
			shutdwonErr = errors.New(pncMsg)
			b.Log().WithError(shutdwonErr).Error("error when shutting down bus")
		}
	}()
	for _, worker := range b.workers {
		err := worker.Stop()
		if err != nil {
			b.Log().WithError(err).Error("could not stop worker")
			return err
		}
	}
	if err := b.Glue.Stop(); err != nil {
		return err
	}
	b.started = false
	err := b.Outbox.Stop()
	if err != nil {
		b.Log().WithError(err).Error("could not shutdown outbox")
		return err
	}
	b.amqpOutbox.Shutdown()
	b.TxProvider.Dispose()
	return nil
}
func (b *DefaultBus) NotifyHealth(health chan error) {
	if health == nil {
		panic("can't pass nil as health channel")
	}
	b.healthChan = health
}
func (b *DefaultBus) GetHealth() HealthCard {
	dbConnected := b.TxProvider.Ping(b.DbPingTimeout)
	return HealthCard{
		DbConnected:        dbConnected,
		RabbitBackPressure: b.backpressure,
		RabbitConnected:    b.amqpConnected,
	}
}
func (b *DefaultBus) withTx(action func(tx *sql.Tx) error, ambientTx *sql.Tx) error {
	var shouldCommitTx bool
	var activeTx *sql.Tx
	if ambientTx == nil {
		shouldCommitTx = true
		newTx, newTxErr := b.TxProvider.New()
		if newTxErr != nil {
			b.Log().WithError(newTxErr).Error("failed to create transaction when sending a transactional message")
			return newTxErr
		}
		activeTx = newTx
	} else {
		activeTx = ambientTx
	}
	retryAction := func() error {
		return action(activeTx)
	}
	actionErr := b.SafeWithRetries(retryAction, MaxRetryCount)
	if shouldCommitTx {
		if actionErr != nil {
			err := activeTx.Rollback()
			if err != nil {
				b.Log().WithError(err).Error("could not rollback transaction")
			}
		} else {
			commitErr := activeTx.Commit()
			if commitErr != nil {
				b.Log().WithError(commitErr).Error("could not commit transaction")
				return commitErr
			}
		}
	}
	return actionErr
}
func (b *DefaultBus) Send(ctx context.Context, toService string, message *BusMessage, policies ...MessagePolicy) error {
	return b.sendWithTx(ctx, nil, toService, message, policies...)
}
func (b *DefaultBus) RPC(ctx context.Context, service string, request, reply *BusMessage, timeout time.Duration) (*BusMessage, error) {
	if !b.started {
		return nil, errors.New("bus not strated or already shutdown, make sure you call bus.Start() before sending messages")
	}
	b.RPCLock.Lock()
	rpcID := xid.New().String()
	request.RPCID = rpcID
	replyChan := make(chan *BusMessage)
	handler := func(invocation Invocation, message *BusMessage) error {
		replyChan <- message
		return nil
	}
	b.RPCHandlers[rpcID] = handler
	b.RPCLock.Unlock()
	request.Semantics = CMD
	rpc := rpcPolicy{
		rpcID: rpcID}
	b.Serializer.Register(reply.Payload)
	sendRPC := func(tx *sql.Tx) error {
		return b.sendImpl(ctx, tx, service, b.rpcQueue.Name, "", "", request, rpc)
	}
	err := b.withTx(sendRPC, nil)
	if err != nil {
		b.Log().WithError(err).Error("could not send message")
		return nil, err
	}
	select {
	case reply := <-replyChan:
		b.RPCLock.Lock()
		delete(b.RPCHandlers, rpcID)
		b.RPCLock.Unlock()
		return reply, nil
	case <-time.After(timeout):
		b.RPCLock.Lock()
		delete(b.RPCHandlers, rpcID)
		b.RPCLock.Unlock()
		return nil, errors.New("rpc call timed out")
	}
}
func (b *DefaultBus) publishWithTx(ctx context.Context, ambientTx *sql.Tx, exchange, topic string, message *BusMessage, policies ...MessagePolicy) error {
	if !b.started {
		return errors.New("bus not strated or already shutdown, make sure you call bus.Start() before sending messages")
	}
	message.Semantics = EVT
	publish := func(tx *sql.Tx) error {
		return b.sendImpl(ctx, tx, "", b.SvcName, exchange, topic, message, policies...)
	}
	return b.withTx(publish, ambientTx)
}
func (b *DefaultBus) sendWithTx(ctx context.Context, ambientTx *sql.Tx, toService string, message *BusMessage, policies ...MessagePolicy) error {
	if !b.started {
		return errors.New("bus not strated or already shutdown, make sure you call bus.Start() before sending messages")
	}
	message.Semantics = CMD
	send := func(tx *sql.Tx) error {
		return b.sendImpl(ctx, tx, toService, b.SvcName, "", "", message, policies...)
	}
	return b.withTx(send, ambientTx)
}
func (b *DefaultBus) returnDeadToQueue(ctx context.Context, ambientTx *sql.Tx, publishing *amqp.Publishing) error {
	if !b.started {
		return errors.New("bus not started or already shutdown, make sure you call bus.Start() before sending messages")
	}
	targetQueue, ok := publishing.Headers["x-first-death-queue"].(string)
	if !ok {
		return fmt.Errorf("bad x-first-death-queue field - %v", publishing.Headers["x-first-death-queue"])
	}
	exchange, ok := publishing.Headers["x-first-death-exchange"].(string)
	if !ok {
		return fmt.Errorf("bad x-first-death-exchange field - %v", publishing.Headers["x-first-death-exchange"])
	}
	routingKey, err := extractFirstDeathRoutingKey(publishing.Headers)
	if err != nil {
		return err
	}
	publishing.Headers[FirstDeathRoutingKeyHeaderName] = routingKey  
	publishing.Headers[ResurrectedHeaderName] = true                 
	delete(publishing.Headers, "x-death")
	delete(publishing.Headers, "x-first-death-queue")
	delete(publishing.Headers, "x-first-death-reason")
	b.Log().
		WithField("message_id", publishing.MessageId).
		WithField("target_queue", targetQueue).
		WithField("first_death_routing_key", routingKey).
		WithField("first_death_exchange", exchange).
		Info("returning dead message to queue...")
	send := func(tx *sql.Tx) error {
		return b.publish(tx, "", targetQueue, publishing)
	}
	return b.withTx(send, ambientTx)
}
func extractFirstDeathRoutingKey(headers amqp.Table) (result string, err error) {
	xDeathList, ok := headers["x-death"].([]interface{})
	if !ok {
		return "", fmt.Errorf("failed extracting routing-key from headers, bad 'x-death' field - %v", headers["x-death"])
	}
	xDeath, ok := xDeathList[0].(amqp.Table)
	if !ok {
		return "", fmt.Errorf("failed extracting routing-key from headers, bad 'x-death' field - %v", headers["x-death"])
	}
	routingKeys, ok := xDeath["routing-keys"].([]interface{})
	if !ok {
		return "", fmt.Errorf("failed extracting routing-key from headers, bad 'routing-keys' field - %v", xDeath["routing-keys"])
	}
	routingKey, ok := routingKeys[len(routingKeys)-1].(string)
	if !ok {
		return "", fmt.Errorf("failed extracting routing-key from headers, bad 'routing-keys' field - %v", xDeath["routing-keys"])
	}
	return routingKey, nil
}
func (b *DefaultBus) Publish(ctx context.Context, exchange, topic string, message *BusMessage, policies ...MessagePolicy) error {
	return b.publishWithTx(ctx, nil, exchange, topic, message, policies...)
}
func (b *DefaultBus) HandleMessage(message Message, handler MessageHandler) error {
	return b.registerHandlerImpl("", b.SvcName, message, handler)
}