queue

Documentation

Index

• Variables
• type Handler
• type Query
• type Queue
  • func NewQueue(store kt.TxnDatastoreExtended, handler Handler) (*Queue, error)
  • func (q *Queue) AddRequest(params RequestParams) (*openapi.PinStatus, error)
  • func (q *Queue) Close() error
  • func (q *Queue) GetRequest(group, id string) (*openapi.PinStatus, error)
  • func (q *Queue) ListRequests(group string, query Query) ([]openapi.PinStatus, error)
  • func (q *Queue) NewID(t time.Time) (string, error)
  • func (q *Queue) RemoveRequest(group, id string) error
  • func (q *Queue) ReplaceRequest(group, id string, pin openapi.Pin) (*openapi.PinStatus, error)
• type Request
• type RequestParams

Variables

var (

	// StartDelay is the time delay before the queue will process queued request on start.
	StartDelay = time.Second * 10

	// MaxConcurrency is the maximum number of requests that will be handled concurrently.
	MaxConcurrency = 100

	// ErrNotFound indicates the requested request was not found.
	ErrNotFound = errors.New("request not found")

	// ErrInProgress indicates the request is in progress and cannot be altered.
	ErrInProgress = errors.New("request in progress")
)

Types

type Handler

type Handler func(ctx context.Context, request Request) error

Handler is called when a request moves from "queued" to "pinning". This separates the queue's job from the actual handling of a request, making the queue logic easier to test.

type Query

type Query struct {
	Cids     []string
	Name     string
	Match    openapi.TextMatchingStrategy
	Statuses []openapi.Status
	Before   time.Time
	After    time.Time
	Limit    int
	Meta     map[string]string
}

Query is used to query for requests (a more typed version of openapi.Query).

type Queue

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

Queue is a persistent worker-based task queue.

func NewQueue

func NewQueue(store kt.TxnDatastoreExtended, handler Handler) (*Queue, error)

NewQueue returns a new Queue using handler to process requests.

func (*Queue) AddRequest

func (q *Queue) AddRequest(params RequestParams) (*openapi.PinStatus, error)

AddRequest adds a new request to the queue. The new request will be handled immediately if workers are not busy.

func (*Queue) Close

func (q *Queue) Close() error

Close the queue and cancel active "pinning" requests.

func (*Queue) GetRequest

func (q *Queue) GetRequest(group, id string) (*openapi.PinStatus, error)

GetRequest returns a request by group key and id.

func (*Queue) ListRequests

func (q *Queue) ListRequests(group string, query Query) ([]openapi.PinStatus, error)

ListRequests lists requests for a group key by applying a Query.

func (*Queue) NewID

func (q *Queue) NewID(t time.Time) (string, error)

NewID returns new monotonically increasing request ids.

func (*Queue) RemoveRequest

func (q *Queue) RemoveRequest(group, id string) error

RemoveRequest removes a request. Note: In-progress ("pinning") requests cannot be removed.

func (*Queue) ReplaceRequest

func (q *Queue) ReplaceRequest(group, id string, pin openapi.Pin) (*openapi.PinStatus, error)

ReplaceRequest replaces a request's openapi.PinStatus.Pin. Note: In-progress ("pinning") requests cannot be replaced.

type Request

type Request struct {
	openapi.PinStatus

	Thread   core.ID
	Key      string
	Identity did.Token

	// Replace indicates openapi.PinStatus.Pin.Cid is marked for replacement.
	Replace bool
	// Remove indicates the request is marked for removal.
	Remove bool
}

Request is a wrapper for openapi.PinStatus that is persisted to the datastore.

type RequestParams

type RequestParams struct {
	openapi.Pin

	// Time the request was received (used for openapi.PinStatus.Created).
	Time time.Time

	Thread   core.ID
	Key      string
	Identity did.Token
}

RequestParams are used to create a new request.