queue

Documentation

Overview

Package queue provides a generic queue resource plus an in-memory adapter for demos and tests. Optional NATS/Kafka wrappers stay thin and injection-based so core ShiftLock does not require broker SDKs.

Index

• type Backend
• type Config
• type Kafka
• type Memory
  • func NewMemory(maxMsgs int) *Memory
  • func (m *Memory) Consume() (string, bool)
  • func (m *Memory) Depth(context.Context) (int, error)
  • func (m *Memory) Pause(context.Context) error
  • func (m *Memory) Paused() bool
  • func (m *Memory) Ping(context.Context) error
  • func (m *Memory) Publish(msg string) error
  • func (m *Memory) Resume(context.Context) error
• type NATS
• type Resource
  • func New(cfg Config) (*Resource, error)
  • func (r *Resource) Capabilities() resource.ResourceCapabilities
  • func (r *Resource) Depth(ctx context.Context) (int, error)
  • func (r *Resource) Describe() resource.Description
  • func (r *Resource) Health(ctx context.Context) resource.ResourceHealth
  • func (r *Resource) ID() resource.ResourceID
  • func (r *Resource) Kind() resource.Kind
  • func (r *Resource) Pause(ctx context.Context) error
  • func (r *Resource) Resume(ctx context.Context) error
  • func (r *Resource) Snapshot(ctx context.Context) (map[string]string, error)

Types

type Backend

type Backend interface {
	Ping(ctx context.Context) error
	Pause(ctx context.Context) error
	Resume(ctx context.Context) error
	Depth(ctx context.Context) (int, error)
}

Backend is the minimal queue control surface.

type Config

type Config struct {
	ID          resource.ResourceID
	DisplayName string
	Backend     Backend
	// MaxDepth soft capacity signal for health (0 = 10000).
	MaxDepth int
}

Config configures a queue resource.

type Kafka

type Kafka struct {
	Backend Backend
}

Kafka is a thin optional wrapper documenting injection of a pause/ping backend.

type Memory

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

Memory is an in-process queue backend for demos/tests.

func NewMemory

func NewMemory(maxMsgs int) *Memory

NewMemory creates a bounded memory queue (default max 1024).

func (*Memory) Consume

func (m *Memory) Consume() (string, bool)

Consume removes and returns the next message.

func (*Memory) Depth

func (m *Memory) Depth(context.Context) (int, error)

func (*Memory) Pause

func (m *Memory) Pause(context.Context) error

func (*Memory) Paused

func (m *Memory) Paused() bool

Paused reports pause state.

func (*Memory) Ping

func (m *Memory) Ping(context.Context) error

func (*Memory) Publish

func (m *Memory) Publish(msg string) error

Publish enqueues a message (rejects when paused or full).

func (*Memory) Resume

func (m *Memory) Resume(context.Context) error

type NATS

type NATS struct {
	Backend Backend
}

NATS is a thin optional wrapper documenting injection of a pause/ping backend. No NATS SDK dependency is pulled into the module.

type Resource

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

Resource implements resource.Resource for queues.

func New

func New(cfg Config) (*Resource, error)

New constructs a queue resource.

func (*Resource) Capabilities

func (r *Resource) Capabilities() resource.ResourceCapabilities

func (*Resource) Depth

func (r *Resource) Depth(ctx context.Context) (int, error)

Depth returns approximate backlog.

func (*Resource) Describe

func (r *Resource) Describe() resource.Description

func (*Resource) Health

func (r *Resource) Health(ctx context.Context) resource.ResourceHealth

func (*Resource) ID

func (r *Resource) ID() resource.ResourceID

func (*Resource) Kind

func (r *Resource) Kind() resource.Kind

func (*Resource) Pause

func (r *Resource) Pause(ctx context.Context) error

Pause pauses consumers (application-level).

func (*Resource) Resume

func (r *Resource) Resume(ctx context.Context) error

Resume resumes consumers.

func (*Resource) Snapshot

func (r *Resource) Snapshot(ctx context.Context) (map[string]string, error)

Snapshot is sanitized (no payloads).