mysql

Documentation

Index

• Constants
• func EnsureSchema(db *sql.DB, svcName string)
• func GetTimeoutsTableName(svcName string) string
• func NewSagaStore(svcName string, txProvider gbus.TxProvider) saga.Store
• func NewTxProvider(connStr string) (*tx.Provider, error)
• type SagaStore
• type TimeoutManager
  • func NewTimeoutManager(bus gbus.Bus, txp gbus.TxProvider, logger func() logrus.FieldLogger, ...) *TimeoutManager
  • func (tm *TimeoutManager) ClearTimeout(tx *sql.Tx, sagaID string) error
  • func (tm *TimeoutManager) RegisterTimeout(tx *sql.Tx, sagaID string, duration time.Duration) error
  • func (tm *TimeoutManager) SetTimeoutFunction(timeoutFunc func(tx *sql.Tx, sagaID string) error)
  • func (tm *TimeoutManager) Start() error
  • func (tm *TimeoutManager) Stop() error
• type TxOutbox
  • func NewOutbox(svcName string, txProv gbus.TxProvider, purgeOnStartup bool, ...) *TxOutbox
  • func (outbox *TxOutbox) Save(tx *sql.Tx, exchange, routingKey string, amqpMessage amqp.Publishing) error
  • func (outbox *TxOutbox) Start(amqpOut *gbus.AMQPOutbox) error
  • func (outbox *TxOutbox) Stop() error

Constants

const (
	Pending int = iota + 1
	Sent
)

Functions

func EnsureSchema

func EnsureSchema(db *sql.DB, svcName string)

EnsureSchema implements Grabbit's migrations strategy

func GetTimeoutsTableName

func GetTimeoutsTableName(svcName string) string

GetTimeoutsTableName returns the table name in which to store timeouts

func NewSagaStore

func NewSagaStore(svcName string, txProvider gbus.TxProvider) saga.Store

NewSagaStore creates a bew SagaStore

func NewTxProvider

func NewTxProvider(connStr string) (*tx.Provider, error)

NewTxProvider returns a new PgProvider

Types

type SagaStore

type SagaStore struct {
	*tx.SagaStore
}

SagaStore implements the saga/store interface on top of MySQL

type TimeoutManager

type TimeoutManager struct {
	Bus         gbus.Bus
	Log         func() logrus.FieldLogger
	TimeoutSaga func(*sql.Tx, string) error
	Txp         gbus.TxProvider
	SvcName     string
	// contains filtered or unexported fields
}

TimeoutManager is a mysql implementation of a persistent timeoutmanager

func NewTimeoutManager

func NewTimeoutManager(bus gbus.Bus, txp gbus.TxProvider, logger func() logrus.FieldLogger, svcName string, purge bool) *TimeoutManager

NewTimeoutManager creates a new instance of a mysql based TimeoutManager

func (*TimeoutManager) ClearTimeout

func (tm *TimeoutManager) ClearTimeout(tx *sql.Tx, sagaID string) error

ClearTimeout clears a timeout for a specific saga

func (*TimeoutManager) RegisterTimeout

func (tm *TimeoutManager) RegisterTimeout(tx *sql.Tx, sagaID string, duration time.Duration) error

RegisterTimeout requests a timeout from the timeout manager

func (*TimeoutManager) SetTimeoutFunction

func (tm *TimeoutManager) SetTimeoutFunction(timeoutFunc func(tx *sql.Tx, sagaID string) error)

SetTimeoutFunction accepts the timeouting function

func (*TimeoutManager) Start

func (tm *TimeoutManager) Start() error

Start starts the timeout manager

func (*TimeoutManager) Stop

func (tm *TimeoutManager) Stop() error

Stop shuts down the timeout manager

type TxOutbox

type TxOutbox struct {
	*gbus.Glogged

	ID string
	// contains filtered or unexported fields
}

TxOutbox is a mysql based transactional outbox

func NewOutbox

func NewOutbox(svcName string, txProv gbus.TxProvider, purgeOnStartup bool, cfg gbus.OutboxConfiguration) *TxOutbox

NewOutbox creates a new mysql transactional outbox

func (*TxOutbox) Save

func (outbox *TxOutbox) Save(tx *sql.Tx, exchange, routingKey string, amqpMessage amqp.Publishing) error

Save stores a message in a DB to ensure delivery

func (*TxOutbox) Start

func (outbox *TxOutbox) Start(amqpOut *gbus.AMQPOutbox) error

Start starts the transactional outbox that is used to send messages in sync with domain object change

func (*TxOutbox) Stop

func (outbox *TxOutbox) Stop() error

Stop forcess the transactional outbox to stop processing additional messages