gbus

type AMQPOutbox struct {
	SvcName string
	// contains filtered or unexported fields
}

type Builder interface {
	PurgeOnStartUp() Builder
	WithDeadlettering(deadletterExchange string) Builder
	/*
		Txnl sets the bus to be transactional using a persisted saga store
		provider: mysql for mysql database
		connStr: connection string in the format of the passed in provider
	*/
	Txnl(provider, connStr string) Builder
	//WithSerializer provides the ability to plugin custom serializers
	WithSerializer(serializer Serializer) Builder
	/*
		 		WorkerNum sets the number of worker go routines consuming messages from the queue
				The default value if this option is not set is 1
	*/
	WorkerNum(workers uint, prefetchCount uint) Builder

	/*
	   WithConfirms enables publisher confirms
	*/
	WithConfirms() Builder

	//WithPolicies defines the default policies that are applied for evey outgoing amqp messge
	WithPolicies(policies ...MessagePolicy) Builder

	//ConfigureHealthCheck defines the default timeout in seconds for the db ping check
	ConfigureHealthCheck(timeoutInSeconds time.Duration) Builder

	//RetriesNum defines the number of retries upon error
	WithConfiguration(config BusConfiguration) Builder

	//Build the bus
	Build(svcName string) Bus

	//WithLogger set custom logger instance
	WithLogger(logger logrus.FieldLogger) Builder
}

type Bus interface {
	HandlerRegister
	Deadlettering
	BusSwitch
	Messaging
	SagaRegister
	Health
	Logged
}

type BusConfiguration struct {
	MaxRetryCount     uint
	BaseRetryDuration int
}

type BusMessage struct {
	ID                string
	CorrelationID     string
	SagaID            string
	SagaCorrelationID string
	Semantics         Semantics /*cmd or evt*/
	Payload           Message
	PayloadFQN        string
	RPCID             string
}

type BusSwitch interface {
	/*
		Start starts the bus, once the bus is started messages get consiumed from the queue
		and handlers get invoced.
		Register all handlers prior to calling GBus.Start()
	*/
	Start() error
	/*
		Shutdown the bus and close connection to the underlying broker
	*/
	Shutdown() error
}

type DeadLetterMessageHandler func(tx *sql.Tx, poison amqp.Delivery) error

type Deadlettering interface {
	HandleDeadletter(handler DeadLetterMessageHandler)
	ReturnDeadToQueue(ctx context.Context, publishing *amqp.Publishing) error
}

type DefaultBus struct {
	*Safety
	*Glogged
	Outbox        TxOutbox
	PrefetchCount uint
	AmqpConnStr   string

	SvcName string

	Registrations []*Registration

	RPCHandlers map[string]MessageHandler

	HandlersLock         *sync.Mutex
	RPCLock              *sync.Mutex
	SenderLock           *sync.Mutex
	ConsumerLock         *sync.Mutex
	RegisteredSchemas    map[string]bool
	DelayedSubscriptions [][]string
	PurgeOnStartup       bool

	Glue       SagaGlue
	TxProvider TxProvider

	WorkerNum       uint
	Serializer      Serializer
	DLX             string
	DefaultPolicies []MessagePolicy
	Confirm         bool

	DbPingTimeout time.Duration
	// contains filtered or unexported fields
}

type DeliveryInfo struct {
	Attempt       uint
	MaxRetryCount uint
}

type Glogged struct {
	// contains filtered or unexported fields
}

type HandlerRegister interface {
	/*
		HandleMessage registers a handler to a specific message type
		Use this method to register handlers for commands and reply messages
		Use the HandleEvent method to subscribe on events and register a handler
	*/
	HandleMessage(message Message, handler MessageHandler) error
	/*
		HandleEvent registers a handler for a specific message type published
		to an exchange with a specific topic
	*/
	HandleEvent(exchange, topic string, event Message, handler MessageHandler) error
}

type Health interface {
	NotifyHealth(health chan error)
	GetHealth() HealthCard
}

type HealthCard struct {
	DbConnected        bool
	RabbitConnected    bool
	RabbitBackPressure bool
}

type Invocation interface {
	Logged
	Reply(ctx context.Context, message *BusMessage) error
	Bus() Messaging
	Tx() *sql.Tx
	Ctx() context.Context
	Routing() (exchange, routingKey string)
	DeliveryInfo() DeliveryInfo
}

type Logged interface {
	SetLogger(entry logrus.FieldLogger)
	Log() logrus.FieldLogger
}

type Message interface {
	SchemaName() string
}

type MessageFilter struct {
	Exchange   string
	RoutingKey string
	MsgName    string
}

type MessageHandler func(invocation Invocation, message *BusMessage) error

type MessagePolicy interface {
	Apply(publishing *amqp.Publishing)
}

type Messaging interface {
	/*
		Send a command or a command response to a specific service
		one-to-one semantics
	*/
	Send(ctx context.Context, toService string, command *BusMessage, policies ...MessagePolicy) error

	/*
		Publish and event, one-to-many semantics
	*/
	Publish(ctx context.Context, exchange, topic string, event *BusMessage, policies ...MessagePolicy) error

	/*
		RPC calls the service passing him the request BusMessage and blocks until a reply is
		received or timeout experied.

	*/
	RPC(ctx context.Context, service string, request, reply *BusMessage, timeout time.Duration) (*BusMessage, error)
}

type Registration struct {
	Handler MessageHandler
	// contains filtered or unexported fields
}

type RequestSagaTimeout interface {
	TimeoutDuration() time.Duration
	Timeout(tx *sql.Tx, bus Messaging) error
}

type Safety struct{}

type Saga interface {
	//StartedBy returns the messages that when received should create a new saga instance
	StartedBy() []Message
	/*
		RegisterAllHandlers passes in the HandlerRegister so that the saga can register
		the messages that it handles
	*/
	RegisterAllHandlers(register HandlerRegister)

	//IsComplete retruns if the saga is complete and can be discarded
	IsComplete() bool

	//New is a factory method used by the bus to crerate new instances of a saga
	New() Saga
}

type SagaConfFn func(Saga) Saga

type SagaGlue interface {
	SagaRegister
	Logged
	Start() error
	Stop() error
}

type SagaRegister interface {
	RegisterSaga(saga Saga, conf ...SagaConfFn) error
}

type SagaTimeoutMessage struct {
	SagaID string
}

type Semantics string

type Serializer interface {
	Name() string
	Encode(message Message) ([]byte, error)
	Decode(buffer []byte, schemaName string) (Message, error)
	Register(obj Message)
}

type TimeoutManager interface {
	//RegisterTimeout requests the TimeoutManager to register a timeout for a specific saga instance
	RegisterTimeout(tx *sql.Tx, sagaID string, duration time.Duration) error
	//ClearTimeout clears a timeout for a specific saga
	ClearTimeout(tx *sql.Tx, sagaID string) error
	//SetTimeoutFunction accepts the function that the TimeoutManager should invoke once a timeout expires
	SetTimeoutFunction(func(tx *sql.Tx, sagaID string) error)
	//Start starts the timeout manager
	Start() error
	//Stop shuts the timeout manager down
	Stop() error
}

type TxOutbox interface {
	Save(tx *sql.Tx, exchange, routingKey string, amqpMessage amqp.Publishing) error
	Start(amqpOut *AMQPOutbox) error
	Stop() error
}

type TxProvider interface {
	New() (*sql.Tx, error)
	Dispose()
	Ping(timeoutInSeconds time.Duration) bool
}