queue

Documentation

Index

• Variables
• type Item
  • func FileToItems(path string) (seeds []Item, err error)
  • func NewItem(URL *url.URL, parentURL *url.URL, itemType string, hop uint64, ID string, ...) (*Item, error)
• type PersistentGroupedQueue
  • func NewPersistentGroupedQueue(queueDirPath string, useHandover bool, useCommit bool) (*PersistentGroupedQueue, error)
  • func (q *PersistentGroupedQueue) BatchEnqueue(items ...*Item) error
  • func (q *PersistentGroupedQueue) CanDequeue() bool
  • func (q *PersistentGroupedQueue) CanEnqueue() bool
  • func (q *PersistentGroupedQueue) Close() error
  • func (q *PersistentGroupedQueue) Dequeue() (*Item, error)
  • func (q *PersistentGroupedQueue) Enqueue(item *Item) error
  • func (q *PersistentGroupedQueue) FreezeDequeue()
  • func (q *PersistentGroupedQueue) GetElementsPerHost() *map[string]int
  • func (q *PersistentGroupedQueue) GetStats() QueueStats
  • func (q *PersistentGroupedQueue) ReadItemAt(position uint64, itemSize uint64) ([]byte, error)
  • func (q *PersistentGroupedQueue) TempDisableHandover(enableBack chan struct{}, syncHandover chan struct{}) bool
• type QueueStats

Variables

var (
	// ErrQueueFull is returned when the queue is full
	ErrQueueFull = errors.New("queue is full")

	// ErrQueueEmpty is returned when the queue is empty
	ErrQueueEmpty = errors.New("queue is empty")

	// ErrQueueClosed is returned when the queue is closed
	ErrQueueClosed = errors.New("queue is closed")

	// ErrQueueTimeout is returned when the queue operation times out
	ErrQueueTimeout = errors.New("queue operation timed out")

	// ErrQueueAlreadyClosed is returned when the queue is already closed
	ErrQueueAlreadyClosed = errors.New("queue is already closed")

	// ErrDequeueClosed is returned when the dequeue operation is called on a closed or empty queue
	ErrDequeueClosed = errors.New("dequeue operation called on a closed or empty queue")

	// ErrCommitValueReceived is returned when a commit value is received when it should not have been
	ErrCommitValueReceived = errors.New("commit value received when it should not have been")

	// ErrCommitValueNotReceived is returned when a commit value is not received when it should have been
	ErrCommitValueNotReceived = errors.New("commit value not received when it should have been")
)

Types

type Item

type Item struct {
	URL             *url.URL
	ParentURL       *url.URL
	Hop             uint64
	Type            string
	ID              string
	BypassSeencheck bool
	Hash            uint64
	LocallyCrawled  uint64
	Redirect        uint64
}

func FileToItems

func FileToItems(path string) (seeds []Item, err error)

func NewItem

func NewItem(URL *url.URL, parentURL *url.URL, itemType string, hop uint64, ID string, bypassSeencheck bool) (*Item, error)

type PersistentGroupedQueue

type PersistentGroupedQueue struct {
	Paused *utils.TAtomBool
	Empty  *utils.TAtomBool

	HandoverOpen *utils.TAtomBool
	// contains filtered or unexported fields
}

func NewPersistentGroupedQueue

func NewPersistentGroupedQueue(queueDirPath string, useHandover bool, useCommit bool) (*PersistentGroupedQueue, error)

func (*PersistentGroupedQueue) BatchEnqueue

func (q *PersistentGroupedQueue) BatchEnqueue(items ...*Item) error

BatchEnqueue adds 1 or many items to the queue in a single operation. If multiple items are provided, the order in which they will be enqueued is not guaranteed. It WILL be less efficient than Enqueue for single items.

func (*PersistentGroupedQueue) CanDequeue

func (q *PersistentGroupedQueue) CanDequeue() bool

func (*PersistentGroupedQueue) CanEnqueue

func (q *PersistentGroupedQueue) CanEnqueue() bool

func (*PersistentGroupedQueue) Close

func (q *PersistentGroupedQueue) Close() error

func (*PersistentGroupedQueue) Dequeue

func (q *PersistentGroupedQueue) Dequeue() (*Item, error)

Dequeue removes and returns the next item from the queue It blocks until an item is available

func (*PersistentGroupedQueue) Enqueue

func (q *PersistentGroupedQueue) Enqueue(item *Item) error

Enqueue adds an item to the queue.

func (*PersistentGroupedQueue) FreezeDequeue

func (q *PersistentGroupedQueue) FreezeDequeue()

func (*PersistentGroupedQueue) GetElementsPerHost

func (q *PersistentGroupedQueue) GetElementsPerHost() *map[string]int

GetElementsPerHost is not thread-safe and should be called with the statsMutex locked If you real need to access elementsPerHost, you should lock the statsMutex

func (*PersistentGroupedQueue) GetStats

func (q *PersistentGroupedQueue) GetStats() QueueStats

generate and return the snapshot of the queue stats NOTE: elementsPerHost is not included in the snapshot

func (*PersistentGroupedQueue) ReadItemAt

func (q *PersistentGroupedQueue) ReadItemAt(position uint64, itemSize uint64) ([]byte, error)

func (*PersistentGroupedQueue) TempDisableHandover

func (q *PersistentGroupedQueue) TempDisableHandover(enableBack chan struct{}, syncHandover chan struct{}) bool

type QueueStats

type QueueStats struct {
	FirstEnqueueTime time.Time `json:"first_enqueue_time"`
	LastEnqueueTime  time.Time `json:"last_enqueue_time"`
	FirstDequeueTime time.Time `json:"first_dequeue_time"`
	LastDequeueTime  time.Time `json:"last_dequeue_time"`

	TotalElements              int           `json:"total_elements"`
	UniqueHosts                int           `json:"unique_hosts"`
	EnqueueCount               int           `json:"enqueue_count"`
	DequeueCount               int           `json:"dequeue_count"`
	AverageTimeBetweenEnqueues time.Duration `json:"average_time_between_enqueues"`
	AverageTimeBetweenDequeues time.Duration `json:"average_time_between_dequeues"`
	AverageElementsPerHost     float64       `json:"average_elements_per_host"`
	HandoverSuccessGetCount    uint64        `json:"handover_success_get_count"`
	// contains filtered or unexported fields
}