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
}

type Bus interface {
	HandlerRegister
	RegisterDeadletterHandler
	BusSwitch
	Messaging
	SagaRegister
	Health
}

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 DefaultBus struct {
	*Safety
	Outgoing      *AMQPOutbox
	Outbox        TxOutbox
	PrefetchCount uint
	AmqpConnStr   string

	AMQPChannel *amqp.Channel

	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            SagaRegister
	TxProvider      TxProvider
	IsTxnl          bool
	WorkerNum       uint
	Serializer      Serializer
	DLX             string
	DefaultPolicies []MessagePolicy
	Confirm         bool

	DbPingTimeout time.Duration
	// 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 {
	Reply(ctx context.Context, message *BusMessage) error
	Bus() Messaging
	Tx() *sql.Tx
	Ctx() context.Context
	Routing() (exchange, routingKey string)
}

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 RegisterDeadletterHandler interface {
	HandleDeadletter(handler func(tx *sql.Tx, poision amqp.Delivery) 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 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 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
}