msq

Documentation

Index

• type Connection
  • func (c *Connection) Attempt() error
  • func (c *Connection) Close() error
  • func (c *Connection) Database() *gorm.DB
  • func (c *Connection) SetupDatabase() error
• type ConnectionConfig
• type Headers
  • func CreateHeader(className string) Headers
• type Listener
  • func (l *Listener) Context() context.Context
  • func (l *Listener) Start(handle func([]MessengerMessage) bool, num int)
  • func (l *Listener) Stop()
• type ListenerConfig
• type MessengerMessage
  • func (e *MessengerMessage) GetHeaders() (Payload, error)
  • func (e *MessengerMessage) GetPayload() (Payload, error)
• type Payload
  • func (p *Payload) Marshal() ([]byte, error)
  • func (p *Payload) UnMarshal(data []byte) (*Payload, error)
• type Queue
  • func Connect(config ConnectionConfig) (*Queue, error)
  • func (q *Queue) Configure(config *QueueConfig)
  • func (q *Queue) Done(event *MessengerMessage) error
  • func (q *Queue) Failed() ([]*MessengerMessage, error)
  • func (q *Queue) Pop() (*MessengerMessage, error)
  • func (q *Queue) Push(payload Payload, headers Headers) (*MessengerMessage, error)
  • func (q *Queue) ReQueue(event *MessengerMessage) error
• type QueueConfig

Types

type Connection

type Connection struct {
	Config ConnectionConfig
	// contains filtered or unexported fields
}

func (*Connection) Attempt

func (c *Connection) Attempt() error

func (*Connection) Close

func (c *Connection) Close() error

func (*Connection) Database

func (c *Connection) Database() *gorm.DB

func (*Connection) SetupDatabase

func (c *Connection) SetupDatabase() error

type ConnectionConfig

type ConnectionConfig struct {
	Type     string
	Host     string
	Proto    string
	Port     int
	Username string
	Password string
	Database string
	Charset  string
	Locale   string
	Logging  bool
}

type Headers

type Headers map[string]interface{}

Symfony headers required type, the message class qualified name X-Message-Stamp-Symfony\\Component\\Messenger\\Stamp\\BusNameStamp, [{\"busName\":\"messenger.bus.default\"}] Content-Type, application\/json

func CreateHeader

func CreateHeader(className string) Headers

type Listener

type Listener struct {
	Running bool
	Queue   Queue
	Config  ListenerConfig
	// contains filtered or unexported fields
}

func (*Listener) Context

func (l *Listener) Context() context.Context

func (*Listener) Start

func (l *Listener) Start(handle func([]MessengerMessage) bool, num int)

func (*Listener) Stop

func (l *Listener) Stop()

type ListenerConfig

type ListenerConfig struct {
	Interval time.Duration
	Timeout  time.Duration
}

type MessengerMessage

type MessengerMessage struct {
	ID          uint `gorm:"primary_key"`
	CreatedAt   time.Time
	AvailableAt time.Time
	DeliveredAt time.Time
	QueueName   string `gorm:"type:varchar(120);index:queue_name;not null"`
	Body        string `gorm:"type:text;body"`
	Headers     string `gorm:"type:text;headers"`
	Retries     int    `gorm:"type:int;retries"`
}

func (*MessengerMessage) GetHeaders

func (e *MessengerMessage) GetHeaders() (Payload, error)

func (*MessengerMessage) GetPayload

func (e *MessengerMessage) GetPayload() (Payload, error)

type Payload

type Payload map[string]interface{}

func (*Payload) Marshal

func (p *Payload) Marshal() ([]byte, error)

func (*Payload) UnMarshal

func (p *Payload) UnMarshal(data []byte) (*Payload, error)

type Queue

type Queue struct {
	Connection *Connection
	Config     *QueueConfig
}

func Connect

func Connect(config ConnectionConfig) (*Queue, error)

func (*Queue) Configure

func (q *Queue) Configure(config *QueueConfig)

func (*Queue) Done

func (q *Queue) Done(event *MessengerMessage) error

func (*Queue) Failed

func (q *Queue) Failed() ([]*MessengerMessage, error)

func (*Queue) Pop

func (q *Queue) Pop() (*MessengerMessage, error)

func (*Queue) Push

func (q *Queue) Push(payload Payload, headers Headers) (*MessengerMessage, error)

func (*Queue) ReQueue

func (q *Queue) ReQueue(event *MessengerMessage) error

type QueueConfig

type QueueConfig struct {
	Name       string
	MaxRetries int64
	MessageTTL time.Duration
}