queue

Documentation

Index

• func AddCountsRoute(lggr logr.Logger, mux *http.ServeMux, q CountReader)
• type CountReader
• type Counter
• type Counts
  • func GetCounts(ctx context.Context, lggr logr.Logger, httpCl *http.Client, ...) (*Counts, error)
  • func NewCounts() *Counts
  • func (q *Counts) Aggregate() int
  • func (q *Counts) MarshalJSON() ([]byte, error)
  • func (q *Counts) String() string
  • func (q *Counts) UnmarshalJSON(data []byte) error
• type FakeCountReader
  • func (f *FakeCountReader) Current() (*Counts, error)
• type FakeCounter
  • func NewFakeCounter() *FakeCounter
  • func (f *FakeCounter) Current() (*Counts, error)
  • func (f *FakeCounter) Ensure(host string)
  • func (f *FakeCounter) Remove(host string) bool
  • func (f *FakeCounter) Resize(host string, i int) error
• type HostAndCount
• type Memory
  • func NewMemory() *Memory
  • func (r *Memory) Current() (*Counts, error)
  • func (r *Memory) Ensure(host string)
  • func (r *Memory) Remove(host string) bool
  • func (r *Memory) Resize(host string, delta int) error

Functions

func AddCountsRoute

func AddCountsRoute(lggr logr.Logger, mux *http.ServeMux, q CountReader)

Types

type CountReader

type CountReader interface {
	// Current returns the current count of pending requests
	// for the given hostname
	Current() (*Counts, error)
}

CountReader represents the size of a virtual HTTP queue, possibly distributed across multiple HTTP server processes. It only can access the current size of the queue, not any other information about requests.

It is concurrency safe.

type Counter

type Counter interface {
	CountReader
	// Resize resizes the queue size by delta for the given host.
	Resize(host string, delta int) error
	// Ensure ensures that host is represented in this counter.
	// If host already has a nonzero value, then it is unchanged. If
	// it is missing, it is set to 0.
	Ensure(host string)
	// Remove tries to remove the given host and its
	// associated counts from the queue. returns true if it existed,
	// false otherwise.
	Remove(host string) bool
}

QueueCounter represents a virtual HTTP queue, possibly distributed across multiple HTTP server processes. It can only increase or decrease the size of the queue or read the current size of the queue, but not read or modify any other information about it.

Both the mutation and read functionality is concurrency safe, but the read functionality is point-in-time only

type Counts

type Counts struct {
	json.Marshaler
	json.Unmarshaler
	fmt.Stringer
	Counts map[string]int
}

Counts is a snapshot of the HTTP pending request queue counts for each host. This is a json.Marshaler, json.Unmarshaler, and fmt.Stringer implementation.

Use NewQueueCounts to create a new one of these.

func GetCounts

func GetCounts(
	ctx context.Context,
	lggr logr.Logger,
	httpCl *http.Client,
	interceptorURL url.URL,
) (*Counts, error)

GetQueueCounts issues an RPC call to get the queue counts from the given hostAndPort. Note that the hostAndPort should not end with a "/" and shouldn't include a path.

func NewCounts

func NewCounts() *Counts

NewQueueCounts creates a new empty QueueCounts struct

func (*Counts) Aggregate

func (q *Counts) Aggregate() int

Aggregate returns the total count across all hosts

func (*Counts) MarshalJSON

func (q *Counts) MarshalJSON() ([]byte, error)

MarshalJSON implements json.Marshaler

func (*Counts) String

func (q *Counts) String() string

String implements fmt.Stringer

func (*Counts) UnmarshalJSON

func (q *Counts) UnmarshalJSON(data []byte) error

UnmarshalJSON implements json.Unmarshaler

type FakeCountReader

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

func (*FakeCountReader) Current

func (f *FakeCountReader) Current() (*Counts, error)

type FakeCounter

type FakeCounter struct {
	RetMap        map[string]int
	ResizedCh     chan HostAndCount
	ResizeTimeout time.Duration
	// contains filtered or unexported fields
}

func NewFakeCounter

func NewFakeCounter() *FakeCounter

func (*FakeCounter) Current

func (f *FakeCounter) Current() (*Counts, error)

func (*FakeCounter) Ensure

func (f *FakeCounter) Ensure(host string)

func (*FakeCounter) Remove

func (f *FakeCounter) Remove(host string) bool

func (*FakeCounter) Resize

func (f *FakeCounter) Resize(host string, i int) error

type HostAndCount

type HostAndCount struct {
	Host  string
	Count int
}

type Memory

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

Memory is a Counter implementation that holds the HTTP queue in memory only. Always use NewMemory to create one of these.

func NewMemory

func NewMemory() *Memory

NewMemoryQueue creates a new empty in-memory queue

func (*Memory) Current

func (r *Memory) Current() (*Counts, error)

Current returns the current size of the queue.

func (*Memory) Ensure

func (r *Memory) Ensure(host string)

func (*Memory) Remove

func (r *Memory) Remove(host string) bool

func (*Memory) Resize

func (r *Memory) Resize(host string, delta int) error

Resize changes the size of the queue. Further calls to Current() return the newly calculated size if no other Resize() calls were made in the interim.