queue

Documentation

Index

• Variables
• func Init(t QueueType, params map[string]any) error
• type MemoryQueue
  • func NewMemoryQueue() *MemoryQueue
  • func (q *MemoryQueue) EventSeen(id string) bool
  • func (q *MemoryQueue) Ping() error
  • func (q *MemoryQueue) Publish(ctx context.Context, item event.VaultEvent) error
  • func (q *MemoryQueue) Push(evt event.VaultEvent) error
  • func (q *MemoryQueue) SeenEvent(id string)
  • func (q *MemoryQueue) Start(params map[string]any) error
  • func (q *MemoryQueue) Stop() error
  • func (q *MemoryQueue) Subscribe(ctx context.Context) (chan event.VaultEvent, error)
• type NATSQueue
  • func NewNATSQueue() *NATSQueue
  • func (q *NATSQueue) EventSeen(id string) bool
  • func (q *NATSQueue) Ping() error
  • func (q *NATSQueue) Publish(ctx context.Context, e event.VaultEvent) error
  • func (q *NATSQueue) Push(evt event.VaultEvent) error
  • func (q *NATSQueue) SeenEvent(id string)
  • func (q *NATSQueue) Start(params map[string]any) error
  • func (q *NATSQueue) Stop() error
  • func (q *NATSQueue) Subscribe(ctx context.Context) (chan event.VaultEvent, error)
• type Queue
  • func NewQueue(t QueueType) (Queue, error)
• type QueueType
• type RedisQueue
  • func NewRedisQueue() *RedisQueue
  • func (q *RedisQueue) EventSeen(id string) bool
  • func (q *RedisQueue) Ping() error
  • func (q *RedisQueue) Publish(ctx context.Context, item event.VaultEvent) error
  • func (q *RedisQueue) Push(evt event.VaultEvent) error
  • func (q *RedisQueue) SeenEvent(id string)
  • func (q *RedisQueue) Start(params map[string]any) error
  • func (q *RedisQueue) Stop() error
  • func (q *RedisQueue) Subscribe(ctx context.Context) (chan event.VaultEvent, error)
• type SQSQueue
  • func NewSQSQueue() *SQSQueue
  • func (q *SQSQueue) EventSeen(id string) bool
  • func (q *SQSQueue) Ping() error
  • func (q *SQSQueue) Publish(ctx context.Context, e event.VaultEvent) error
  • func (q *SQSQueue) Push(evt event.VaultEvent) error
  • func (q *SQSQueue) SeenEvent(id string)
  • func (q *SQSQueue) Start(params map[string]any) error
  • func (q *SQSQueue) Stop() error
  • func (q *SQSQueue) Subscribe(ctx context.Context) (chan event.VaultEvent, error)
• type TLSConfig
• type UnboundedChannel
  • func NewUnboundedChannel() *UnboundedChannel
  • func (u *UnboundedChannel) Len() int
  • func (u *UnboundedChannel) Receive(ctx context.Context) (interface{}, error)
  • func (u *UnboundedChannel) Send(v interface{})

Variables

var (
	Q      Queue
	Dedupe bool
)

Functions

func Init

func Init(t QueueType, params map[string]any) error

Types

type MemoryQueue

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

func NewMemoryQueue

func NewMemoryQueue() *MemoryQueue

func (*MemoryQueue) EventSeen

func (q *MemoryQueue) EventSeen(id string) bool

func (*MemoryQueue) Ping

func (q *MemoryQueue) Ping() error

func (*MemoryQueue) Publish

func (q *MemoryQueue) Publish(ctx context.Context, item event.VaultEvent) error

func (*MemoryQueue) Push

func (q *MemoryQueue) Push(evt event.VaultEvent) error

func (*MemoryQueue) SeenEvent

func (q *MemoryQueue) SeenEvent(id string)

func (*MemoryQueue) Start

func (q *MemoryQueue) Start(params map[string]any) error

func (*MemoryQueue) Stop

func (q *MemoryQueue) Stop() error

func (*MemoryQueue) Subscribe

func (q *MemoryQueue) Subscribe(ctx context.Context) (chan event.VaultEvent, error)

type NATSQueue

type NATSQueue struct {
	Url     string     `json:"url" yaml:"url"`
	Subject string     `json:"subject" yaml:"subject"`
	TLS     *TLSConfig `json:"tls" yaml:"tls"`
	// contains filtered or unexported fields
}

func NewNATSQueue

func NewNATSQueue() *NATSQueue

func (*NATSQueue) EventSeen

func (q *NATSQueue) EventSeen(id string) bool

func (*NATSQueue) Ping

func (q *NATSQueue) Ping() error

func (*NATSQueue) Publish

func (q *NATSQueue) Publish(ctx context.Context, e event.VaultEvent) error

func (*NATSQueue) Push

func (q *NATSQueue) Push(evt event.VaultEvent) error

func (*NATSQueue) SeenEvent

func (q *NATSQueue) SeenEvent(id string)

func (*NATSQueue) Start

func (q *NATSQueue) Start(params map[string]any) error

func (*NATSQueue) Stop

func (q *NATSQueue) Stop() error

func (*NATSQueue) Subscribe

func (q *NATSQueue) Subscribe(ctx context.Context) (chan event.VaultEvent, error)

type Queue

type Queue interface {
	Start(params map[string]any) error
	Stop() error
	Publish(context.Context, event.VaultEvent) error
	Push(event.VaultEvent) error
	Subscribe(ctx context.Context) (chan event.VaultEvent, error)
	SeenEvent(string)
	EventSeen(string) bool
	Ping() error
}

func NewQueue

func NewQueue(t QueueType) (Queue, error)

type QueueType

type QueueType string

const (
	QueueTypeMemory QueueType = "memory"
	QueueTypeNATS   QueueType = "nats"
	QueueTypeRedis  QueueType = "redis"
	QueueTypeSQS    QueueType = "sqs"
)

type RedisQueue

type RedisQueue struct {
	Host     string     `yaml:"host" json:"host"`
	Port     int        `yaml:"port" json:"port"`
	Database int        `yaml:"database" json:"database"`
	Password string     `yaml:"password" json:"password"`
	TLS      *TLSConfig `yaml:"tls" json:"tls"`
	// contains filtered or unexported fields
}

func NewRedisQueue

func NewRedisQueue() *RedisQueue

func (*RedisQueue) EventSeen

func (q *RedisQueue) EventSeen(id string) bool

func (*RedisQueue) Ping

func (q *RedisQueue) Ping() error

func (*RedisQueue) Publish

func (q *RedisQueue) Publish(ctx context.Context, item event.VaultEvent) error

func (*RedisQueue) Push

func (q *RedisQueue) Push(evt event.VaultEvent) error

func (*RedisQueue) SeenEvent

func (q *RedisQueue) SeenEvent(id string)

func (*RedisQueue) Start

func (q *RedisQueue) Start(params map[string]any) error

func (*RedisQueue) Stop

func (q *RedisQueue) Stop() error

func (*RedisQueue) Subscribe

func (q *RedisQueue) Subscribe(ctx context.Context) (chan event.VaultEvent, error)

type SQSQueue

type SQSQueue struct {
	Url     string `json:"url" yaml:"url"`
	Region  string `json:"region" yaml:"region"`
	RoleArn string `json:"roleArn" yaml:"roleArn"`
	// contains filtered or unexported fields
}

func NewSQSQueue

func NewSQSQueue() *SQSQueue

func (*SQSQueue) EventSeen

func (q *SQSQueue) EventSeen(id string) bool

func (*SQSQueue) Ping

func (q *SQSQueue) Ping() error

func (*SQSQueue) Publish

func (q *SQSQueue) Publish(ctx context.Context, e event.VaultEvent) error

func (*SQSQueue) Push

func (q *SQSQueue) Push(evt event.VaultEvent) error

func (*SQSQueue) SeenEvent

func (q *SQSQueue) SeenEvent(id string)

func (*SQSQueue) Start

func (q *SQSQueue) Start(params map[string]any) error

func (*SQSQueue) Stop

func (q *SQSQueue) Stop() error

func (*SQSQueue) Subscribe

func (q *SQSQueue) Subscribe(ctx context.Context) (chan event.VaultEvent, error)

type TLSConfig

type TLSConfig struct {
	CA   string `json:"ca" yaml:"ca"`
	Cert string `json:"cert" yaml:"cert"`
	Key  string `json:"key" yaml:"key"`
}

type UnboundedChannel

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

UnboundedChannel implements an unbounded FIFO queue backed by a linked list

func NewUnboundedChannel

func NewUnboundedChannel() *UnboundedChannel

func (*UnboundedChannel) Len

func (u *UnboundedChannel) Len() int

Len returns the current length of the queue

func (*UnboundedChannel) Receive

func (u *UnboundedChannel) Receive(ctx context.Context) (interface{}, error)

Receive removes and returns the first item from the queue It blocks if the queue is empty

func (*UnboundedChannel) Send

func (u *UnboundedChannel) Send(v interface{})

Send adds an item to the queue