gbus

package

Go to main page

Versions in this module

v1

v1.1.8

Dec 2, 2019

v1.1.7

Oct 22, 2019

Changes in this version

type Semantics

+ const REPLY

v1.1.6

Oct 20, 2019

Changes in this version

+ var FirstDeathRoutingKeyHeaderName = "x-first-death-routing-key"

+ var ResurrectedHeaderName = "x-resurrected-from-death"

+ func GetDeliveryLogEntries(delivery amqp.Delivery) logrus.Fields

type BusConfiguration

+ OutboxCfg OutboxConfiguration

type BusMessage

+ IdempotencyKey string

+ func (bm *BusMessage) SetIdempotencyKey(idempotencyKey string)

+ type OutboxConfiguration struct

+ Ackers uint

+ MetricsInterval time.Duration

+ PageSize uint

+ ScavengeInterval time.Duration

+ SendInterval time.Duration

v1.1.5

Sep 29, 2019

v1.1.4

Sep 28, 2019

v1.1.3

Sep 25, 2019

Changes in this version

type BusMessage

+ func (bm *BusMessage) TargetSaga(sagaID string)

type SagaInvocation

+ SagaID func() string

v1.1.2

Sep 17, 2019

Changes in this version

type Invocation

+ InvokingSvc func() string

+ type SagaInvocation interface

+ HostingSvc func() string

+ ReplyToInitiator func(ctx context.Context, message *BusMessage) error

v1.1.1

Aug 31, 2019

Changes in this version

+ var RPCHeaderName = "x-grabbit-msg-rpc-id"

+ func GetMessageName(delivery amqp.Delivery) string

type BusMessage

+ func NewFromDelivery(delivery amqp.Delivery) (*BusMessage, error)

type DefaultBus

+ func (b *DefaultBus) SetGlobalRawMessageHandler(handler RawMessageHandler)

+ type RawMessageHandler func(tx *sql.Tx, delivery *amqp.Delivery) error

+ func (dlmg RawMessageHandler) Name() string

+ type RawMessageHandling interface

+ SetGlobalRawMessageHandler func(handler RawMessageHandler)

v1.1.0

Aug 18, 2019

Changes in this version

type AMQPOutbox

+ func (out *AMQPOutbox) Shutdown()

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

+ func (dlmg DeadLetterMessageHandler) Name() string

+ type Deadlettering interface

+ HandleDeadletter func(handler DeadLetterMessageHandler)

+ ReturnDeadToQueue func(ctx context.Context, publishing *amqp.Publishing) error

type DefaultBus

+ func (b *DefaultBus) ReturnDeadToQueue(ctx context.Context, publishing *amqp.Publishing) error

+ type DeliveryInfo struct

+ Attempt uint

+ MaxRetryCount uint

type Invocation

+ DeliveryInfo func() DeliveryInfo

type MessageHandler

+ func (mg MessageHandler) Name() string

+ type SagaGlue interface

+ Start func() error

+ Stop func() error

+ type TimeoutManager interface

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

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

+ SetTimeoutFunction func(func(tx *sql.Tx, sagaID string) error)

+ Start func() error

+ Stop func() error

v1.0.3

Jun 23, 2019

Changes in this version

type Builder

+ WithLogger func(logger logrus.FieldLogger) Builder

type DefaultBus

+ func (b *DefaultBus) Log() logrus.FieldLogger

+ type Glogged struct

+ func (gl *Glogged) Log() logrus.FieldLogger

+ func (gl *Glogged) SetLogger(entry logrus.FieldLogger)

+ type Logged interface

+ Log func() logrus.FieldLogger

+ SetLogger func(entry logrus.FieldLogger)

v1.0.2

Jun 8, 2019

Changes in this version

+ var BaseRetryDuration = 10 * time.Millisecond

type Builder

+ WithConfiguration func(config BusConfiguration) Builder

+ type BusConfiguration struct

+ BaseRetryDuration int

+ MaxRetryCount uint

v1.0.1

May 1, 2019

v1.0.0

Apr 24, 2019

Changes in this version

+ var MaxRetryCount uint = 3

+ var RpcHeaderName = "x-grabbit-msg-rpc-id"

+ func GetFqn(obj interface{}) string

+ func GetTypeFQN(t reflect.Type) string

+ type AMQPOutbox struct

+ SvcName string

+ func (out *AMQPOutbox) NotifyConfirm(ack, nack chan uint64)

+ func (out *AMQPOutbox) Post(exchange, routingKey string, amqpMessage amqp.Publishing) (uint64, error)

+ type Builder interface

+ Build func(svcName string) Bus

+ ConfigureHealthCheck func(timeoutInSeconds time.Duration) Builder

+ PurgeOnStartUp func() Builder

+ Txnl func(provider, connStr string) Builder

+ WithConfirms func() Builder

+ WithDeadlettering func(deadletterExchange string) Builder

+ WithPolicies func(policies ...MessagePolicy) Builder

+ WithSerializer func(serializer Serializer) Builder

+ WorkerNum func(workers uint, prefetchCount uint) Builder

+ type Bus interface

+ type BusMessage struct

+ CorrelationID string

+ ID string

+ Payload Message

+ PayloadFQN string

+ RPCID string

+ SagaCorrelationID string

+ SagaID string

+ Semantics Semantics

+ func NewBusMessage(payload Message) *BusMessage

+ func NewFromAMQPHeaders(headers amqp.Table) *BusMessage

+ func (bm *BusMessage) GetAMQPHeaders() (headers amqp.Table)

+ func (bm *BusMessage) GetTraceLog() (fields []log.Field)

+ func (bm *BusMessage) SetFromAMQPHeaders(headers amqp.Table)

+ func (bm *BusMessage) SetPayload(payload Message)

+ type BusSwitch interface

+ Shutdown func() error

+ Start func() error

+ type DefaultBus struct

+ AMQPChannel *amqp.Channel

+ AmqpConnStr string

+ Confirm bool

+ ConsumerLock *sync.Mutex

+ DLX string

+ DbPingTimeout time.Duration

+ DefaultPolicies []MessagePolicy

+ DelayedSubscriptions [][]string

+ Glue SagaRegister

+ HandlersLock *sync.Mutex

+ IsTxnl bool

+ Outbox TxOutbox

+ Outgoing *AMQPOutbox

+ PrefetchCount uint

+ PurgeOnStartup bool

+ RPCHandlers map[string]MessageHandler

+ RPCLock *sync.Mutex

+ RegisteredSchemas map[string]bool

+ Registrations []*Registration

+ SenderLock *sync.Mutex

+ Serializer Serializer

+ SvcName string

+ TxProvider TxProvider

+ WorkerNum uint

+ func (b *DefaultBus) GetHealth() HealthCard

+ func (b *DefaultBus) HandleDeadletter(handler func(tx *sql.Tx, poision amqp.Delivery) error)

+ func (b *DefaultBus) HandleEvent(exchange, topic string, event Message, handler MessageHandler) error

+ func (b *DefaultBus) HandleMessage(message Message, handler MessageHandler) error

+ func (b *DefaultBus) NotifyHealth(health chan error)

+ func (b *DefaultBus) Publish(ctx context.Context, exchange, topic string, message *BusMessage, ...) error

+ func (b *DefaultBus) RPC(ctx context.Context, service string, request, reply *BusMessage, ...) (*BusMessage, error)

+ func (b *DefaultBus) RegisterSaga(saga Saga, conf ...SagaConfFn) error

+ func (b *DefaultBus) Send(ctx context.Context, toService string, message *BusMessage, ...) error

+ func (b *DefaultBus) Shutdown() (shutdwonErr error)

+ func (b *DefaultBus) Start() error

+ type HandlerRegister interface

+ HandleEvent func(exchange, topic string, event Message, handler MessageHandler) error

+ HandleMessage func(message Message, handler MessageHandler) error

+ type Health interface

+ GetHealth func() HealthCard

+ NotifyHealth func(health chan error)

+ type HealthCard struct

+ DbConnected bool

+ RabbitBackPressure bool

+ RabbitConnected bool

+ type Invocation interface

+ Bus func() Messaging

+ Ctx func() context.Context

+ Reply func(ctx context.Context, message *BusMessage) error

+ Routing func() (exchange, routingKey string)

+ Tx func() *sql.Tx

+ type Message interface

+ SchemaName func() string

+ type MessageFilter struct

+ Exchange string

+ MsgName string

+ RoutingKey string

+ func NewMessageFilter(exchange, routingKey string, message Message) *MessageFilter

+ func (filter *MessageFilter) Matches(exchange, routingKey, msgName string) bool

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

+ type MessagePolicy interface

+ Apply func(publishing *amqp.Publishing)

+ type Messaging interface

+ Publish func(ctx context.Context, exchange, topic string, event *BusMessage, ...) error

+ RPC func(ctx context.Context, service string, request, reply *BusMessage, ...) (*BusMessage, error)

+ Send func(ctx context.Context, toService string, command *BusMessage, ...) error

+ type RegisterDeadletterHandler interface

+ HandleDeadletter func(handler func(tx *sql.Tx, poision amqp.Delivery) error)

+ type Registration struct

+ Handler MessageHandler

+ func NewRegistration(exchange, routingKey string, message Message, handler MessageHandler) *Registration

+ func (sub *Registration) Matches(exchange, routingKey, msgName string) bool

+ type RequestSagaTimeout interface

+ Timeout func(invocation Invocation, message *BusMessage) error

+ TimeoutDuration func() time.Duration

+ type Safety struct

+ func (s *Safety) SafeWithRetries(funk func() error, retries uint) error

+ type Saga interface

+ IsComplete func() bool

+ New func() Saga

+ RegisterAllHandlers func(register HandlerRegister)

+ StartedBy func() []Message

+ type SagaConfFn func(Saga) Saga

+ type SagaRegister interface

+ RegisterSaga func(saga Saga, conf ...SagaConfFn) error

+ type SagaTimeoutMessage struct

+ SagaID string

+ func (SagaTimeoutMessage) SchemaName() string

+ type Semantics string

+ const CMD

+ const EVT

+ type Serializer interface

+ Decode func(buffer []byte, schemaName string) (Message, error)

+ Encode func(message Message) ([]byte, error)

+ Name func() string

+ Register func(obj Message)

+ type TxOutbox interface

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

+ Start func(amqpOut *AMQPOutbox) error

+ Stop func() error

+ type TxProvider interface

+ Dispose func()

+ New func() (*sql.Tx, error)

+ Ping func(timeoutInSeconds time.Duration) bool

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL