inbox

Documentation

Index

• type ConsumerSessionMessage
• type Inbox
  • func NewSub(db *pg.DB, opt ...Opts) *Inbox
  • func (i *Inbox) Cleanup(sarama.ConsumerGroupSession) error
  • func (i *Inbox) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (i *Inbox) Handler(handler MessageHandler) *Inbox
  • func (i *Inbox) Setup(sarama.ConsumerGroupSession) error
  • func (i *Inbox) Shutdown(ctx context.Context) error
  • func (i *Inbox) Start(ctx context.Context) (err error)
• type InboxManager
  • func NewInboxManager(db *pg.DB) *InboxManager
  • func (im *InboxManager) ProcessMessage(ctx context.Context, message common.Message) error
• type Inboxes
• type MessageHandler
• type MultiBatchConsumerConfig
• type Opts
  • func WithConsumer(consumer sarama.ConsumerGroup) Opts
  • func WithMultiBatchConfig(cfg *MultiBatchConsumerConfig) Opts
  • func WithTopics(topics []string) Opts

Types

type ConsumerSessionMessage

type ConsumerSessionMessage struct {
	Message *sarama.ConsumerMessage
	Session sarama.ConsumerGroupSession
}

type Inbox

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

func NewSub

func NewSub(db *pg.DB, opt ...Opts) *Inbox

func (*Inbox) Cleanup

func (i *Inbox) Cleanup(sarama.ConsumerGroupSession) error

func (*Inbox) ConsumeClaim

func (i *Inbox) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

func (*Inbox) Handler

func (i *Inbox) Handler(handler MessageHandler) *Inbox

func (*Inbox) Setup

func (i *Inbox) Setup(sarama.ConsumerGroupSession) error

func (*Inbox) Shutdown

func (i *Inbox) Shutdown(ctx context.Context) error

func (*Inbox) Start

func (i *Inbox) Start(ctx context.Context) (err error)

type InboxManager

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

func NewInboxManager

func NewInboxManager(db *pg.DB) *InboxManager

func (*InboxManager) ProcessMessage

func (im *InboxManager) ProcessMessage(ctx context.Context, message common.Message) error

type Inboxes

type Inboxes struct {
	ID          string                 `pg:"id,type:uuid,default:uuid_generate_v4()"`
	Payload     map[string]interface{} `pg:"type:jsonb,notnull"`
	Topic       string                 `pg:"type:varchar(255),notnull"`
	IsAccepted  bool                   `pg:"type:bool,notnull,default:false"`
	Publisher   string                 `pg:"type:varchar(255),default:null"`
	CreatedAt   time.Time              `pg:"type:timestamptz,default:now()"`
	ProcessedAt time.Time              `pg:"type:timestamptz,default:null"`
	CreatedBy   string                 `pg:"type:varchar(255)"`
	UpdatedAt   time.Time              `pg:"type:timestamptz,default:now()"`
	UpdatedBy   string                 `pg:"type:varchar(255)"`
}

type MessageHandler

type MessageHandler interface {
	Dispatch(ctx context.Context, message common.Message) error
}

type MultiBatchConsumerConfig

type MultiBatchConsumerConfig struct {
	BufferCapacity        int
	MaxBufSize            int
	TickerIntervalSeconds int
	BufChan               chan batchMessages
}

type Opts

type Opts func(*Inbox) error

func WithConsumer

func WithConsumer(consumer sarama.ConsumerGroup) Opts

func WithMultiBatchConfig

func WithMultiBatchConfig(cfg *MultiBatchConsumerConfig) Opts

func WithTopics

func WithTopics(topics []string) Opts