func (imsm *Glue) SagaHandler(invocation gbus.Invocation, message *gbus.BusMessage) error {
	imsm.lock.Lock()
	defer imsm.lock.Unlock()
	msgName := message.PayloadFQN
	defs := imsm.msgToDefMap[strings.ToLower(msgName)]
	for _, def := range defs {
		logInContext := invocation.Log().WithFields(
			logrus.Fields{"saga_def": def.String(),
				"saga_type": def.sagaType})
		startNew := def.shouldStartNewSaga(message)
		if startNew {
			return imsm.handleNewSaga(def, invocation, message)
		} else if message.SagaCorrelationID != "" {
			instance, getErr := imsm.sagaStore.GetSagaByID(invocation.Tx(), message.SagaCorrelationID)
			logInContext = logInContext.WithField("saga_correlation_id", message.SagaCorrelationID)
			if getErr != nil {
				logInContext.Error("failed to fetch saga by id")
				return getErr
			}
			if instance == nil {
				logInContext.Warn("message routed with SagaCorrelationID but no saga instance with the same id found")
				return nil
			}
			logInContext = logInContext.WithField("saga_id", instance.ID)
			def.configureSaga(instance)
			if invkErr := imsm.invokeSagaInstance(def, instance, invocation, message); invkErr != nil {
				logInContext.WithError(invkErr).Error("failed to invoke saga")
				return invkErr
			}
			return imsm.completeOrUpdateSaga(invocation.Tx(), instance)
		} else if message.Semantics == gbus.CMD {
			logInContext.Warn("command or reply message with no saga reference received")
			return errors.New("can not resolve saga instance for message")
		} else {
			logInContext.Info("fetching saga instances by type")
			instances, e := imsm.sagaStore.GetSagasByType(invocation.Tx(), def.sagaType)
			if e != nil {
				return e
			}
			logInContext.WithFields(logrus.Fields{"instances_fetched": len(instances)}).Info("fetched saga instances")
			for _, instance := range instances {
				def.configureSaga(instance)
				if invkErr := imsm.invokeSagaInstance(def, instance, invocation, message); invkErr != nil {
					logInContext.WithError(invkErr).Error("failed to invoke saga")
					return invkErr
				}
				e = imsm.completeOrUpdateSaga(invocation.Tx(), instance)
				if e != nil {
					return e
				}
			}
		}
	}
	return nil
}
SagaHandler is the generic handler invoking saga instances
package saga
import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"reflect"
	"strings"
	"sync"
	"time"
	"github.com/opentracing/opentracing-go"
	slog "github.com/opentracing/opentracing-go/log"
	"github.com/sirupsen/logrus"
	"github.com/wework/grabbit/gbus"
	"github.com/wework/grabbit/gbus/metrics"
)
func fqnsFromMessages(objs []gbus.Message) []string {
	fqns := make([]string, 0)
	for _, obj := range objs {
		fqn := obj.SchemaName()
		fqns = append(fqns, fqn)
	}
	return fqns
}
var ErrInstanceNotFound = errors.New("saga not be found")
var _ gbus.SagaGlue = &Glue{}
type Glue struct {
	*gbus.Glogged
	svcName          string
	bus              gbus.Bus
	sagaDefs         []*Def
	lock             *sync.Mutex
	alreadyRegistred map[string]bool
	msgToDefMap      map[string][]*Def
	sagaStore        Store
	timeoutManager   gbus.TimeoutManager
}
func (imsm *Glue) isSagaAlreadyRegistered(sagaType reflect.Type) bool {
	for _, def := range imsm.sagaDefs {
		if def.sagaType == sagaType {
			return true
		}
	}
	return false
}
func (imsm *Glue) RegisterSaga(saga gbus.Saga, conf ...gbus.SagaConfFn) error {
	sagaType := reflect.TypeOf(saga)
	if imsm.isSagaAlreadyRegistered(sagaType) {
		return fmt.Errorf("saga of type %v already registered", sagaType)
	}
	imsm.sagaStore.RegisterSagaType(saga)
	def := &Def{
		glue:        imsm,
		sagaType:    sagaType,
		sagaConfFns: conf,
		startedBy:   fqnsFromMessages(saga.StartedBy()),
		msgToFunc:   make([]*MsgToFuncPair, 0),
		lock:        &sync.Mutex{}}
	saga.RegisterAllHandlers(def)
	imsm.sagaDefs = append(imsm.sagaDefs, def)
	msgNames := def.getHandledMessages()
	for _, msgName := range msgNames {
		imsm.addMsgNameToDef(msgName, def)
	}
	imsm.Log().
		WithFields(logrus.Fields{"saga_type": def.sagaType.String(), "handles_messages": len(msgNames)}).
		Info("registered saga with messages")
	return nil
}
func (imsm *Glue) addMsgNameToDef(msgName string, def *Def) {
	defs := imsm.getDefsForMsgName(msgName)
	defs = append(defs, def)
	imsm.msgToDefMap[msgName] = defs
}
func (imsm *Glue) getDefsForMsgName(msgName string) []*Def {
	defs := imsm.msgToDefMap[msgName]
	if defs == nil {
		defs = make([]*Def, 0)
	}
	return defs
}
func (imsm *Glue) handleNewSaga(def *Def, invocation gbus.Invocation, message *gbus.BusMessage) error {
	newInstance := def.newInstance()
	newInstance.StartedBy = invocation.InvokingSvc()
	newInstance.StartedBySaga = message.SagaID
	newInstance.StartedByRPCID = message.RPCID
	newInstance.StartedByMessageID = message.ID
	logInContext := invocation.Log().WithFields(logrus.Fields{"saga_def": def.String(), "saga_id": newInstance.ID})
	logInContext.
		Info("created new saga")
	if invkErr := imsm.invokeSagaInstance(def, newInstance, invocation, message); invkErr != nil {
		logInContext.Error("failed to invoke saga")
		return invkErr
	}
	if !newInstance.isComplete() {
		logInContext.Info("saving new saga")
		if e := imsm.sagaStore.SaveNewSaga(invocation.Tx(), def.sagaType, newInstance); e != nil {
			logInContext.Error("saving new saga failed")
			return e
		}
		if requestsTimeout, duration := newInstance.requestsTimeout(); requestsTimeout {
			logInContext.WithField("timeout_duration", duration).Info("new saga requested timeout")
			if tme := imsm.timeoutManager.RegisterTimeout(invocation.Tx(), newInstance.ID, duration); tme != nil {
				return tme
			}
		}
	}
	return nil
}