Documentation
¶
Index ¶
- type Authorizer
- type DequeueParams
- type DequeueResult
- type Handler
- type LeaseBatchResult
- type OpError
- type Server
- func (s *Server) AckBatch(route string, leaseIDs []string) (LeaseBatchResult, *OpError)
- func (s *Server) AckSingle(route string, leaseID string) *OpError
- func (s *Server) Dequeue(ctx context.Context, route string, params DequeueParams) (DequeueResult, *OpError)
- func (s *Server) Extend(route string, leaseID string, extendBy time.Duration) *OpError
- func (s *Server) NackBatch(route string, leaseIDs []string, dead bool, reason string, delay time.Duration) (LeaseBatchResult, *OpError)
- func (s *Server) NackSingle(route string, leaseID string, dead bool, reason string, delay time.Duration) *OpError
- func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request)
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Authorizer ¶
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 ¶
DequeueResult is the route-bound dequeue output.
type LeaseBatchResult ¶
type LeaseBatchResult struct {
Succeeded int
Conflicts []queue.LeaseBatchConflict
}
LeaseBatchResult captures aggregate lease mutation output.
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 (*Server) AckBatch ¶
func (s *Server) AckBatch(route string, leaseIDs []string) (LeaseBatchResult, *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) NackSingle ¶
Click to show internal directories.
Click to hide internal directories.