func (store *SagaStore) RegisterSagaType(saga gbus.Saga) {
	gob.Register(saga)
}
RegisterSagaType implements interface method store.RegisterSagaType
package tx
import (
	"bytes"
	"database/sql"
	"encoding/gob"
	"errors"
	"fmt"
	"reflect"
	"regexp"
	"strings"
	"time"
	log "github.com/sirupsen/logrus"
	"github.com/go-sql-driver/mysql"
	"github.com/wework/grabbit/gbus"
	"github.com/wework/grabbit/gbus/saga"
)
type SagaStore struct {
	*gbus.Glogged
	Tx            gbus.TxProvider
	SvcName       string
	ParamsMarkers []string
}
func (store *SagaStore) scanInstances(rows *sql.Rows) ([]*saga.Instance, error) {
	instances := make([]*saga.Instance, 0)
	for rows.Next() {
		var sagaID, sagaType string
		var version int
		var sagaData []byte
		var startedBy sql.NullString
		var startedBySaga sql.NullString
		var startedByMsgID sql.NullString
		var startedByRPCID sql.NullString
		var createdAt mysql.NullTime
		error := rows.Scan(&sagaID, &sagaType, &sagaData, &startedBy, &startedByMsgID, &startedByRPCID, &startedBySaga, &version, &createdAt)
		if error == sql.ErrNoRows {
			return nil, error
		} else if error != nil {
			store.Log().WithError(error).Error("failed to scan saga row")
			return nil, error
		}
		reader := bytes.NewReader(sagaData)
		dec := gob.NewDecoder(reader)
		var instance saga.Instance
		instance.ConcurrencyCtrl = version
		decErr := dec.Decode(&instance)
		if startedBy.Valid {
			instance.StartedBy = startedBy.String
		}
		if startedBySaga.Valid {
			instance.StartedBySaga = startedBySaga.String
		}
		if startedByMsgID.Valid {
			instance.StartedByMessageID = startedByMsgID.String
		}
		if startedByRPCID.Valid {
			instance.StartedByRPCID = startedByRPCID.String
		}
		if createdAt.Valid {
			value, _ := createdAt.Value()
			instance.CreatedAt = value.(time.Time)
		}
		if decErr != nil {
			store.Log().WithError(decErr).Error("failed to decode saga instance")
			return nil, decErr
		}
		instances = append(instances, &instance)
	}
	return instances, nil
}
func (store *SagaStore) GetSagasByType(tx *sql.Tx, sagaType reflect.Type) (instances []*saga.Instance, err error) {
	tblName := GetSagatableName(store.SvcName)
	selectSQL := "SELECT saga_id, saga_type, saga_data, started_by_request_of_svc, started_by_msg_id, started_by_rpcid, started_by_request_of_saga, version, created_at FROM " + tblName + " WHERE saga_type=" + store.ParamsMarkers[0]
	rows, err := tx.Query(selectSQL, sagaType.String())
	defer func() {
		err := rows.Close()
		if err != nil {
			store.Log().WithError(err).Error("could not close rows")
		}
	}()
	if err != nil {
		return nil, err
	}
	if instances, err = store.scanInstances(rows); err != nil {
		store.Log().WithError(err).Error("SagaStore failed to scan saga db record")
		return nil, err
	}
	return instances, nil
}
func (store *SagaStore) UpdateSaga(tx *sql.Tx, instance *saga.Instance) (err error) {
	tblName := GetSagatableName(store.SvcName)
	currentVersion := instance.ConcurrencyCtrl
	nextVersion := instance.ConcurrencyCtrl + 1
	instance.ConcurrencyCtrl = nextVersion
	var buf []byte
	if buf, err = store.serilizeSaga(instance); err != nil {
		store.Log().WithError(err).WithField("saga_id", instance.ID).Error("SagaStore failed to encode saga")
		return err
	}
	updateSQL := `UPDATE ` + tblName + ` SET saga_data=` + store.ParamsMarkers[0] + `, version=` + store.ParamsMarkers[1] + `
WHERE saga_id=` + store.ParamsMarkers[2] + ` AND version=` + store.ParamsMarkers[3] + ``
	result, err := tx.Exec(updateSQL, buf, nextVersion, instance.ID, currentVersion)
	if err != nil {
		return err
	} else if rowsAffected, ee := result.RowsAffected(); ee != nil || rowsAffected == 0 {
		return fmt.Errorf("saga with saga_id:%v had stale data when updating. :%v", instance.ID, ee)
	}
	return nil
}