httpclient

Documentation

Overview

Package httpclient provides a typed Go client for consuming the PostgreSQL queue management REST API.

Create a client with:

client, err := httpclient.New("http://localhost:8080/api/v1")
if err != nil {
   panic(err)
}

Then use the client to query resources:

queues, err := client.ListQueues(ctx)
tasks, err := client.ListTasks(ctx)
tickers, err := client.ListTickers(ctx)

Retain tasks with optional queue filtering:

// From specific queue
task, err := client.RetainTask(ctx, "worker-1", "emails")

// From any queue
task, err := client.RetainTask(ctx, "worker-1")

Index

• type Client
  • func New(url string, opts ...client.ClientOpt) (*Client, error)
  • func (c *Client) CreateQueue(ctx context.Context, queue schema.QueueMeta) (*schema.Queue, error)
  • func (c *Client) CreateTask(ctx context.Context, queue string, meta schema.TaskMeta) (*schema.Task, error)
  • func (c *Client) CreateTicker(ctx context.Context, ticker schema.TickerMeta) (*schema.Ticker, error)
  • func (c *Client) DeleteQueue(ctx context.Context, name string) (*schema.Queue, error)
  • func (c *Client) DeleteTicker(ctx context.Context, name string) (*schema.Ticker, error)
  • func (c *Client) GetQueue(ctx context.Context, name string) (*schema.Queue, error)
  • func (c *Client) GetTicker(ctx context.Context, name string) (*schema.Ticker, error)
  • func (c *Client) ListNamespaces(ctx context.Context) (*schema.NamespaceList, error)
  • func (c *Client) ListQueues(ctx context.Context, opts ...Opt) (*schema.QueueList, error)
  • func (c *Client) ListTasks(ctx context.Context, opts ...Opt) (*schema.TaskList, error)
  • func (c *Client) ListTickers(ctx context.Context, opts ...Opt) (*schema.TickerList, error)
  • func (c *Client) NextTicker(ctx context.Context, callback func(client.TextStreamEvent) error) error
  • func (c *Client) ReleaseTask(ctx context.Context, id uint64, result any) (*schema.TaskWithStatus, error)
  • func (c *Client) RetainTask(ctx context.Context, worker string, queue ...string) (*schema.TaskWithStatus, error)
  • func (c *Client) UpdateQueue(ctx context.Context, name string, meta schema.QueueMeta) (*schema.Queue, error)
  • func (c *Client) UpdateTicker(ctx context.Context, name string, meta schema.TickerMeta) (*schema.Ticker, error)
• type Opt
  • func OptSet(key, value string) Opt
  • func WithLimit(limit uint64) Opt
  • func WithOffset(offset uint64) Opt
  • func WithOffsetLimit(offset uint64, limit *uint64) Opt
  • func WithQueue(queue string) Opt
  • func WithStatus(status string) Opt
  • func WithWorker(worker string) Opt

Types

type Client

type Client struct {
	*client.Client
}

func New

func New(url string, opts ...client.ClientOpt) (*Client, error)

New creates a new HTTP client for the PostgreSQL queue API.

func (*Client) CreateQueue

func (c *Client) CreateQueue(ctx context.Context, queue schema.QueueMeta) (*schema.Queue, error)

func (*Client) CreateTask

func (c *Client) CreateTask(ctx context.Context, queue string, meta schema.TaskMeta) (*schema.Task, error)

CreateTask creates a new task in a queue (POST /task).

func (*Client) CreateTicker

func (c *Client) CreateTicker(ctx context.Context, ticker schema.TickerMeta) (*schema.Ticker, error)

func (*Client) DeleteQueue

func (c *Client) DeleteQueue(ctx context.Context, name string) (*schema.Queue, error)

func (*Client) DeleteTicker

func (c *Client) DeleteTicker(ctx context.Context, name string) (*schema.Ticker, error)

func (*Client) GetQueue

func (c *Client) GetQueue(ctx context.Context, name string) (*schema.Queue, error)

func (*Client) GetTicker

func (c *Client) GetTicker(ctx context.Context, name string) (*schema.Ticker, error)

func (*Client) ListNamespaces

func (c *Client) ListNamespaces(ctx context.Context) (*schema.NamespaceList, error)

func (*Client) ListQueues

func (c *Client) ListQueues(ctx context.Context, opts ...Opt) (*schema.QueueList, error)

func (*Client) ListTasks

func (c *Client) ListTasks(ctx context.Context, opts ...Opt) (*schema.TaskList, error)

ListTasks returns all tasks with optional filtering by status (GET /task).

func (*Client) ListTickers

func (c *Client) ListTickers(ctx context.Context, opts ...Opt) (*schema.TickerList, error)

func (*Client) NextTicker

func (c *Client) NextTicker(ctx context.Context, callback func(client.TextStreamEvent) error) error

func (*Client) ReleaseTask

func (c *Client) ReleaseTask(ctx context.Context, id uint64, result any) (*schema.TaskWithStatus, error)

ReleaseTask releases a task (PATCH /task/{id}). If result is non-nil, marks the task as failed with result as the error payload. If result is nil, marks the task as completed successfully.

func (*Client) RetainTask

func (c *Client) RetainTask(ctx context.Context, worker string, queue ...string) (*schema.TaskWithStatus, error)

RetainTask gets the next available task (PUT /task or PUT /task/{queue}?worker=XX). If queue is empty, tasks from any queue are considered.

func (*Client) UpdateQueue

func (c *Client) UpdateQueue(ctx context.Context, name string, meta schema.QueueMeta) (*schema.Queue, error)

func (*Client) UpdateTicker

func (c *Client) UpdateTicker(ctx context.Context, name string, meta schema.TickerMeta) (*schema.Ticker, error)

type Opt

type Opt func(*opt) error

Opt is an option to set on the client request.

func OptSet

func OptSet(key, value string) Opt

OptSet is a generic option to set a query parameter.

func WithLimit

func WithLimit(limit uint64) Opt

WithLimit sets the limit query parameter.

func WithOffset

func WithOffset(offset uint64) Opt

WithOffset sets the offset query parameter.

func WithOffsetLimit

func WithOffsetLimit(offset uint64, limit *uint64) Opt

WithOffsetLimit sets offset and limit query parameters.

func WithQueue

func WithQueue(queue string) Opt

WithQueue sets the queue query parameter.

func WithStatus

func WithStatus(status string) Opt

WithStatus sets the status query parameter for filtering tasks.

func WithWorker

func WithWorker(worker string) Opt

WithWorker sets the worker query parameter.