queue

Documentation

Index

• type InMemoryQueueStorage
  • func (q *InMemoryQueueStorage) AddRequest(r []byte) error
  • func (q *InMemoryQueueStorage) GetRequest() ([]byte, error)
  • func (q *InMemoryQueueStorage) Init() error
  • func (q *InMemoryQueueStorage) QueueSize() (int, error)
• type Queue
  • func New(threads int, s Storage) (*Queue, error)
  • func (q *Queue) AddRequest(r *fastcolly.Request) error
  • func (q *Queue) AddURL(URL string) error
  • func (q *Queue) IsEmpty() bool
  • func (q *Queue) Run(c *fastcolly.Collector, doCall func(r *fastcolly.Request, err error), ...) error
  • func (q *Queue) Size() (int, error)
• type Storage

Types

type InMemoryQueueStorage

type InMemoryQueueStorage struct {
	// MaxSize defines the capacity of the queue.
	// New requests are discarded if the queue size reaches MaxSize
	MaxSize int
	// contains filtered or unexported fields
}

InMemoryQueueStorage is the default implementation of the Storage interface. InMemoryQueueStorage holds the request queue in memory.

func (*InMemoryQueueStorage) AddRequest

func (q *InMemoryQueueStorage) AddRequest(r []byte) error

AddRequest implements Storage.AddRequest() function

func (*InMemoryQueueStorage) GetRequest

func (q *InMemoryQueueStorage) GetRequest() ([]byte, error)

GetRequest implements Storage.GetRequest() function

func (*InMemoryQueueStorage) Init

func (q *InMemoryQueueStorage) Init() error

Init implements Storage.Init() function

func (*InMemoryQueueStorage) QueueSize

func (q *InMemoryQueueStorage) QueueSize() (int, error)

QueueSize implements Storage.QueueSize() function

type Queue

type Queue struct {
	// Threads defines the number of consumer threads
	Threads int
	// contains filtered or unexported fields
}

Queue is a request queue which uses a Collector to consume requests in multiple threads

func New

func New(threads int, s Storage) (*Queue, error)

New creates a new queue with a Storage specified in argument A standard InMemoryQueueStorage is used if Storage argument is nil.

func (*Queue) AddRequest

func (q *Queue) AddRequest(r *fastcolly.Request) error

AddRequest adds a new Request to the queue

func (*Queue) AddURL

func (q *Queue) AddURL(URL string) error

AddURL adds a new URL to the queue

func (*Queue) IsEmpty

func (q *Queue) IsEmpty() bool

IsEmpty returns true if the queue is empty

func (*Queue) Run

func (q *Queue) Run(c *fastcolly.Collector, doCall func(r *fastcolly.Request, err error), exit <-chan struct{}, done chan<- struct{}) error

Run starts consumer threads and calls the Collector to perform requests. Run blocks while the queue has active requests

func (*Queue) Size

func (q *Queue) Size() (int, error)

Size returns the size of the queue

type Storage

type Storage interface {
	// Init initializes the storage
	Init() error
	// AddRequest adds a serialized request to the queue
	AddRequest([]byte) error
	// GetRequest pops the next request from the queue
	// or returns error if the queue is empty
	GetRequest() ([]byte, error)
	// QueueSize returns with the size of the queue
	QueueSize() (int, error)
}

Storage is the interface of the queue's storage backend