brokerstore

Documentation

Index

• type BrokerStore
  • func (b *BrokerStore) FlushAllProducers(ctx context.Context)
  • func (b *BrokerStore) GetAdmin(ctx context.Context, options messagebroker.AdminClientOptions) (messagebroker.Admin, error)
  • func (b *BrokerStore) GetConsumer(ctx context.Context, op messagebroker.ConsumerClientOptions) (messagebroker.Consumer, error)
  • func (b *BrokerStore) GetProducer(ctx context.Context, op messagebroker.ProducerClientOptions) (messagebroker.Producer, error)
  • func (b *BrokerStore) IsTopicCleanUpEnabled() bool
  • func (b *BrokerStore) RemoveConsumer(ctx context.Context, op messagebroker.ConsumerClientOptions) bool
  • func (b *BrokerStore) RemoveProducer(ctx context.Context, op messagebroker.ProducerClientOptions) bool
• type IBrokerStore
  • func NewBrokerStore(variant string, config *messagebroker.BrokerConfig) (IBrokerStore, error)
• type Key
  • func NewKey(name string, id string) *Key
  • func (key *Key) Prefix() string
  • func (key *Key) String() string

Types

type BrokerStore

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

BrokerStore holds broker clients

func (*BrokerStore) FlushAllProducers

func (b *BrokerStore) FlushAllProducers(ctx context.Context)

FlushAllProducers will iterate over the producer map, and flush all messages in the producer buffer

func (*BrokerStore) GetAdmin

func (b *BrokerStore) GetAdmin(ctx context.Context, options messagebroker.AdminClientOptions) (messagebroker.Admin, error)

GetAdmin returns for an existing admin instance, if available returns that else creates as new instance

func (*BrokerStore) GetConsumer

func (b *BrokerStore) GetConsumer(ctx context.Context, op messagebroker.ConsumerClientOptions) (messagebroker.Consumer, error)

GetConsumer returns for an existing consumer instance, if available returns that else creates as new instance

func (*BrokerStore) GetProducer

func (b *BrokerStore) GetProducer(ctx context.Context, op messagebroker.ProducerClientOptions) (messagebroker.Producer, error)

GetProducer returns for an existing producer instance, if available returns that else creates as new instance

func (*BrokerStore) IsTopicCleanUpEnabled

func (b *BrokerStore) IsTopicCleanUpEnabled() bool

IsTopicCleanUpEnabled checks if topic clean-up config is enabled or not

func (*BrokerStore) RemoveConsumer

func (b *BrokerStore) RemoveConsumer(ctx context.Context, op messagebroker.ConsumerClientOptions) bool

RemoveConsumer deletes the consumer from the store

func (*BrokerStore) RemoveProducer

func (b *BrokerStore) RemoveProducer(ctx context.Context, op messagebroker.ProducerClientOptions) bool

RemoveProducer deletes the producer from the store followed by a shutdown

type IBrokerStore

type IBrokerStore interface {

	// GetConsumer returns for an existing consumer instance, if available returns that else creates as new instance
	GetConsumer(ctx context.Context, op messagebroker.ConsumerClientOptions) (messagebroker.Consumer, error)

	// RemoveConsumer deletes the consumer from the store
	RemoveConsumer(ctx context.Context, op messagebroker.ConsumerClientOptions) bool

	// GetProducer returns for an existing producer instance, if available returns that else creates as new instance
	GetProducer(ctx context.Context, op messagebroker.ProducerClientOptions) (messagebroker.Producer, error)

	// RemoveProducer deletes the producer from the store followed by a shutdown
	RemoveProducer(ctx context.Context, op messagebroker.ProducerClientOptions) bool

	// GetAdmin returns for an existing admin instance, if available returns that else creates as new instance
	GetAdmin(ctx context.Context, op messagebroker.AdminClientOptions) (messagebroker.Admin, error)

	// FlushAllProducers will iterate over the producer map, and flush all messages in the producer buffer
	FlushAllProducers(ctx context.Context)

	// IsTopicCleanUpEnabled returns true if topicCleanUp is enabled in the env
	IsTopicCleanUpEnabled() bool
}

IBrokerStore ...

func NewBrokerStore

func NewBrokerStore(variant string, config *messagebroker.BrokerConfig) (IBrokerStore, error)

NewBrokerStore returns a concrete implementation IBrokerStore

type Key

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

Key ...

func NewKey

func NewKey(name string, id string) *Key

NewKey creates a new key for broker map

func (*Key) Prefix

func (key *Key) Prefix() string

Prefix returns only the topic name to run a match all query

func (*Key) String

func (key *Key) String() string