pullapi

package
v2.11.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Aug 18, 2026 License: Apache-2.0 Imports: 16 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Authorizer

type Authorizer func(r *http.Request) bool

func BearerTokenAuthorizer

func BearerTokenAuthorizer(tokens [][]byte) Authorizer

type DequeueParams

type DequeueParams struct {
	Batch       int
	MaxWait     time.Duration
	HasMaxWait  bool
	LeaseTTL    time.Duration
	HasLeaseTTL bool
}

DequeueParams captures normalized dequeue input for route-bound operations.

type DequeueResult

type DequeueResult struct {
	Items []queue.Envelope
}

DequeueResult is the route-bound dequeue output.

type Handler

type Handler interface {
	ServeHTTP(w http.ResponseWriter, r *http.Request)
}

type LeaseBatchResult

type LeaseBatchResult struct {
	Succeeded int
	Conflicts []queue.LeaseBatchConflict
}

LeaseBatchResult captures aggregate lease mutation output.

type OpError

type OpError struct {
	StatusCode int
	Code       string
	Detail     string
}

OpError is a transport-neutral operation error for pull worker flows.

func (*OpError) Error

func (e *OpError) Error() string

type Server

type Server struct {
	Store                queue.Store
	Target               string
	ResolveRoute         func(endpoint string) (route string, ok bool)
	Authorize            Authorizer
	ObserveDequeue       func(route string, statusCode int, items []queue.Envelope)
	ObserveAck           func(route string, statusCode int, leaseID string, leaseExpired bool)
	ObserveNack          func(route string, statusCode int, leaseID string, leaseExpired bool)
	ObserveExtend        func(route string, statusCode int, leaseID string, extendBy time.Duration, leaseExpired bool)
	DefaultLeaseTTL      time.Duration
	MaxBatch             int
	MaxLeaseBatch        int
	MaxLeaseTTL          time.Duration
	DefaultMaxWait       time.Duration
	MaxWait              time.Duration
	SSEKeepalive         time.Duration
	SSEMaxConnection     time.Duration
	ObserveSSEConnect    func(route string)
	ObserveSSEDisconnect func(route string, statusCode int, messagesSent int, duration time.Duration)

	// LeaseRouteScoped reports whether the running config uses per-route pull
	// credentials. When it does, lease operations are checked against the route
	// they were issued for; when it does not, every client is authorized for
	// every route anyway and the check is skipped. See lease_scope.go.
	LeaseRouteScoped func() bool

	RecentLeaseOpTTL time.Duration
	RecentLeaseOpCap int
	// contains filtered or unexported fields
}

func NewServer

func NewServer(store queue.Store) *Server

func (*Server) AckBatch

func (s *Server) AckBatch(route string, leaseIDs []string) (LeaseBatchResult, *OpError)

func (*Server) AckSingle

func (s *Server) AckSingle(route string, leaseID string) *OpError

func (*Server) Dequeue

func (s *Server) Dequeue(ctx context.Context, route string, params DequeueParams) (DequeueResult, *OpError)

Dequeue takes ctx so a long poll ends when the caller does. Without it a worker that timed out left the handler blocked inside the store for the rest of max_wait, and items arriving meanwhile were leased into a connection that no longer existed -- invisible for the whole lease TTL, until the expiry sweep reclaimed them.

func (*Server) Extend

func (s *Server) Extend(route string, leaseID string, extendBy time.Duration) *OpError

func (*Server) NackBatch

func (s *Server) NackBatch(route string, leaseIDs []string, dead bool, reason string, delay time.Duration) (LeaseBatchResult, *OpError)

func (*Server) NackSingle

func (s *Server) NackSingle(route string, leaseID string, dead bool, reason string, delay time.Duration) *OpError

func (*Server) ServeHTTP

func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request)

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL