queue

Documentation

Overview

Package queue provides queue implementations that can be used for asynchronous scheduling of fetch actions.

Index

• type GCP
  • func (q *GCP) EnqueueScan(ctx context.Context, task Task, opts *Options) (enqueued bool, err error)
• type InMemory
  • func NewInMemory(ctx context.Context, workerCount int, processFunc inMemoryProcessFunc) *InMemory
  • func (q *InMemory) EnqueueScan(ctx context.Context, task Task, _ *Options) (bool, error)
  • func (q *InMemory) WaitForTesting(ctx context.Context)
• type Options
• type Queue
  • func New(ctx context.Context, cfg *config.Config, processFunc inMemoryProcessFunc) (Queue, error)
• type Task

Types

type GCP

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

GCP provides a Queue implementation backed by the Google Cloud Tasks API.

func (*GCP) EnqueueScan

func (q *GCP) EnqueueScan(ctx context.Context, task Task, opts *Options) (enqueued bool, err error)

EnqueueScan enqueues a scan task on GCP. It returns an error if there was an error hashing the task name, or an error pushing the task to GCP. If the task was a duplicate, it returns (false, nil).

type InMemory

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

InMemory is a Queue implementation that schedules in-process fetch operations. Unlike the GCP task queue, it will not automatically retry tasks on failure.

This should only be used for local development.

func NewInMemory

func NewInMemory(ctx context.Context, workerCount int, processFunc inMemoryProcessFunc) *InMemory

NewInMemory creates a new InMemory that asynchronously fetches from proxyClient and stores in db. It uses workerCount parallelism to execute these fetches.

func (*InMemory) EnqueueScan

func (q *InMemory) EnqueueScan(ctx context.Context, task Task, _ *Options) (bool, error)

EnqueueScan pushes a scan task into the local queue to be processed asynchronously.

func (*InMemory) WaitForTesting

func (q *InMemory) WaitForTesting(ctx context.Context)

WaitForTesting waits for all queued requests to finish. It should only be used by test code.

type Options

type Options struct {
	// Namespace prefixes the URL path.
	Namespace string
	// DisableProxyFetch reports whether proxyfetch should be set to off when
	// making a fetch request.
	DisableProxyFetch bool

	// TaskNameSuffix is appended to the task name to force reprocessing of
	// tasks that would normally be de-duplicated.
	TaskNameSuffix string
}

Options is used to provide option arguments for a task queue.

type Queue

type Queue interface {
	// EnqueueScan enqueues a scan request.
	// It reports whether a new task was actually added.
	EnqueueScan(context.Context, Task, *Options) (bool, error)
}

A Queue provides an interface for asynchronous scheduling of fetch actions.

func New

func New(ctx context.Context, cfg *config.Config, processFunc inMemoryProcessFunc) (Queue, error)

New creates a new Queue with name queueName based on the configuration in cfg. When running locally, Queue uses numWorkers concurrent workers.

type Task

type Task interface {
	Name() string   // Human-readable string for the task. Need not be unique.
	Path() string   // URL path
	Params() string // URL query params
}

A Task can produce information needed for Cloud Tasks.