package base

Import Path
	github.com/hibiken/asynq/internal/base (on go.dev)

Dependency Relation
	imports 13 packages, and imported by 3 packages

Involved Source Files

	  d base.go
		Package base defines foundational types and constants used in asynq package.
Package-Level Type Names (total 11)

	/* sort by: alphabet | popularity */
	 type Broker (interface)
		Broker is a message broker that supports operations to manage task queues.
		
		See rdb.RDB as a reference implementation.

		Methods (total 28)
			( Broker) AddToGroup(ctx context.Context, msg *TaskMessage, gname string) error
				Group aggregation related methods

			( Broker) AddToGroupUnique(ctx context.Context, msg *TaskMessage, groupKey string, ttl time.Duration) error
			( Broker) AggregationCheck(qname, gname string, t time.Time, gracePeriod, maxDelay time.Duration, maxSize int) (aggregationSetID string, err error)
			( Broker) Archive(ctx context.Context, msg *TaskMessage, errMsg string) error
			( Broker) CancelationPubSub() (*redis.PubSub, error)
				Cancelation related methods
				// TODO: Need to decouple from redis to support other brokers

			( Broker) ClearServerState(host string, pid int, serverID string) error
			( Broker) Close() error
			( Broker) DeleteAggregationSet(ctx context.Context, qname, gname, aggregationSetID string) error
			( Broker) DeleteExpiredCompletedTasks(qname string) error
				Task retention related method

			( Broker) Dequeue(qnames ...string) (*TaskMessage, time.Time, error)
			( Broker) Done(ctx context.Context, msg *TaskMessage) error
			( Broker) Enqueue(ctx context.Context, msg *TaskMessage) error
			( Broker) EnqueueUnique(ctx context.Context, msg *TaskMessage, ttl time.Duration) error
			( Broker) ExtendLease(qname string, ids ...string) (time.Time, error)
			( Broker) ForwardIfReady(qnames ...string) error
			( Broker) ListGroups(qname string) ([]string, error)
			( Broker) ListLeaseExpired(cutoff time.Time, qnames ...string) ([]*TaskMessage, error)
				Lease related methods

			( Broker) MarkAsComplete(ctx context.Context, msg *TaskMessage) error
			( Broker) Ping() error
			( Broker) PublishCancelation(id string) error
			( Broker) ReadAggregationSet(qname, gname, aggregationSetID string) ([]*TaskMessage, time.Time, error)
			( Broker) ReclaimStaleAggregationSets(qname string) error
			( Broker) Requeue(ctx context.Context, msg *TaskMessage) error
			( Broker) Retry(ctx context.Context, msg *TaskMessage, processAt time.Time, errMsg string, isFailure bool) error
			( Broker) Schedule(ctx context.Context, msg *TaskMessage, processAt time.Time) error
			( Broker) ScheduleUnique(ctx context.Context, msg *TaskMessage, processAt time.Time, ttl time.Duration) error
			( Broker) WriteResult(qname, id string, data []byte) (n int, err error)
			( Broker) WriteServerState(info *ServerInfo, workers []*WorkerInfo, ttl time.Duration) error
				State snapshot related methods

		Implemented By (at least one exported)
			*github.com/hibiken/asynq/internal/rdb.RDB
		Implements (at least 2)
			 Broker : github.com/prometheus/common/expfmt.Closer
			 Broker : io.Closer

	 type Cancelations (struct)
		Cancelations is a collection that holds cancel functions for all active tasks.
		
		Cancelations are safe for concurrent use by multiple goroutines.

		Methods (total 3)
			(*Cancelations) Add(id string, fn context.CancelFunc)
				Add adds a new cancel func to the collection.

			(*Cancelations) Delete(id string)
				Delete deletes a cancel func from the collection given an id.

			(*Cancelations) Get(id string) (fn context.CancelFunc, ok bool)
				Get returns a cancel func given an id.

		As Outputs Of (at least one exported)
			func NewCancelations() *Cancelations

	 type Lease (struct)
		Lease is a time bound lease for worker to process task.
		It provides a communication channel between lessor and lessee about lease expiration.

		Fields (only one)
			Clock timeutil.Clock
		Methods (total 5)
			(*Lease) Deadline() time.Time
				Deadline returns the expiration time of the lease.

			(*Lease) Done() <-chan struct{}
				Done returns a communication channel from which the lessee can read to get notified when lessor notifies about lease expiration.

			(*Lease) IsValid() bool
				IsValid returns true if the lease's expiration time is in the future or equals to the current time,
				returns false otherwise.

			(*Lease) NotifyExpiration() bool
				Sends a notification to lessee about expired lease
				Returns true if notification was sent, returns false if the lease is still valid and notification was not sent.

			(*Lease) Reset(expirationTime time.Time) bool
				Reset changes the lease to expire at the given time.
				It returns true if the lease is still valid and reset operation was successful, false if the lease had been expired.

		Implements (at least one exported)
			*Lease : database/sql/driver.Validator
		As Outputs Of (at least one exported)
			func NewLease(expirationTime time.Time) *Lease

	 type SchedulerEnqueueEvent (struct)
		SchedulerEnqueueEvent holds information about an enqueue event by a scheduler.

		Fields (total 2)
			EnqueuedAt time.Time
				Time the task was enqueued.

			TaskID string
				ID of the task that was enqueued.

		As Outputs Of (at least 2)
			func DecodeSchedulerEnqueueEvent(b []byte) (*SchedulerEnqueueEvent, error)
			func github.com/hibiken/asynq/internal/rdb.(*RDB).ListSchedulerEnqueueEvents(entryID string, pgn rdb.Pagination) ([]*SchedulerEnqueueEvent, error)
		As Inputs Of (at least 2)
			func EncodeSchedulerEnqueueEvent(event *SchedulerEnqueueEvent) ([]byte, error)
			func github.com/hibiken/asynq/internal/rdb.(*RDB).RecordSchedulerEnqueueEvent(entryID string, event *SchedulerEnqueueEvent) error

	 type SchedulerEntry (struct)
		SchedulerEntry holds information about a periodic task registered with a scheduler.

		Fields (total 7)
			ID string
				Identifier of this entry.

			Next time.Time
				Next shows the next time the task will be enqueued.

			Opts []string
				Opts is the options for the periodic task.

			Payload []byte
				Payload is the payload of the periodic task.

			Prev time.Time
				Prev shows the last time the task was enqueued.
				Zero time if task was never enqueued.

			Spec string
				Spec describes the schedule of this entry.

			Type string
				Type is the task type of the periodic task.

		As Outputs Of (at least 2)
			func DecodeSchedulerEntry(b []byte) (*SchedulerEntry, error)
			func github.com/hibiken/asynq/internal/rdb.(*RDB).ListSchedulerEntries() ([]*SchedulerEntry, error)
		As Inputs Of (at least 2)
			func EncodeSchedulerEntry(entry *SchedulerEntry) ([]byte, error)
			func github.com/hibiken/asynq/internal/rdb.(*RDB).WriteSchedulerEntries(schedulerID string, entries []*SchedulerEntry, ttl time.Duration) error

	 type ServerInfo (struct)
		ServerInfo holds information about a running server.

		Fields (total 9)
			ActiveWorkerCount int
			Concurrency int
			Host string
			PID int
			Queues map[string]int
			ServerID string
			Started time.Time
			Status string
			StrictPriority bool
		As Outputs Of (at least 2)
			func DecodeServerInfo(b []byte) (*ServerInfo, error)
			func github.com/hibiken/asynq/internal/rdb.(*RDB).ListServers() ([]*ServerInfo, error)
		As Inputs Of (at least 3)
			func EncodeServerInfo(info *ServerInfo) ([]byte, error)
			func Broker.WriteServerState(info *ServerInfo, workers []*WorkerInfo, ttl time.Duration) error
			func github.com/hibiken/asynq/internal/rdb.(*RDB).WriteServerState(info *ServerInfo, workers []*WorkerInfo, ttl time.Duration) error

	 type TaskInfo (struct)
		TaskInfo describes a task message and its metadata.

		Fields (total 4)
			Message *TaskMessage
			NextProcessAt time.Time
			Result []byte
			State TaskState
		As Outputs Of (at least 8)
			func github.com/hibiken/asynq/internal/rdb.(*RDB).GetTaskInfo(qname, id string) (*TaskInfo, error)
			func github.com/hibiken/asynq/internal/rdb.(*RDB).ListActive(qname string, pgn rdb.Pagination) ([]*TaskInfo, error)
			func github.com/hibiken/asynq/internal/rdb.(*RDB).ListAggregating(qname, gname string, pgn rdb.Pagination) ([]*TaskInfo, error)
			func github.com/hibiken/asynq/internal/rdb.(*RDB).ListArchived(qname string, pgn rdb.Pagination) ([]*TaskInfo, error)
			func github.com/hibiken/asynq/internal/rdb.(*RDB).ListCompleted(qname string, pgn rdb.Pagination) ([]*TaskInfo, error)
			func github.com/hibiken/asynq/internal/rdb.(*RDB).ListPending(qname string, pgn rdb.Pagination) ([]*TaskInfo, error)
			func github.com/hibiken/asynq/internal/rdb.(*RDB).ListRetry(qname string, pgn rdb.Pagination) ([]*TaskInfo, error)
			func github.com/hibiken/asynq/internal/rdb.(*RDB).ListScheduled(qname string, pgn rdb.Pagination) ([]*TaskInfo, error)

	 type TaskMessage (struct)
		TaskMessage is the internal representation of a task with additional metadata fields.
		Serialized data of this type gets written to redis.

		Fields (total 14)
			CompletedAt int64
				CompletedAt is the time the task was processed successfully in Unix time,
				the number of seconds elapsed since January 1, 1970 UTC.
				
				Use zero to indicate no value.

			Deadline int64
				Deadline specifies the deadline for the task in Unix time,
				the number of seconds elapsed since January 1, 1970 UTC.
				If task processing doesn't complete before the deadline, the task will be retried
				if retry count is remaining. Otherwise it will be moved to the archive.
				
				Use zero to indicate no deadline.

			ErrorMsg string
				ErrorMsg holds the error message from the last failure.

			GroupKey string
				GroupKey holds the group key used for task aggregation.
				
				Empty string indicates no aggregation is used for this task.

			ID string
				ID is a unique identifier for each task.

			LastFailedAt int64
				Time of last failure in Unix time,
				the number of seconds elapsed since January 1, 1970 UTC.
				
				Use zero to indicate no last failure

			Payload []byte
				Payload holds data needed to process the task.

			Queue string
				Queue is a name this message should be enqueued to.

			Retention int64
				Retention specifies the number of seconds the task should be retained after completion.

			Retried int
				Retried is the number of times we've retried this task so far.

			Retry int
				Retry is the max number of retry for this task.

			Timeout int64
				Timeout specifies timeout in seconds.
				If task processing doesn't complete within the timeout, the task will be retried
				if retry count is remaining. Otherwise it will be moved to the archive.
				
				Use zero to indicate no timeout.

			Type string
				Type indicates the kind of the task to be performed.

			UniqueKey string
				UniqueKey holds the redis key used for uniqueness lock for this task.
				
				Empty string indicates that no uniqueness lock was used.

		As Outputs Of (at least 7)
			func DecodeMessage(data []byte) (*TaskMessage, error)
			func Broker.Dequeue(qnames ...string) (*TaskMessage, time.Time, error)
			func Broker.ListLeaseExpired(cutoff time.Time, qnames ...string) ([]*TaskMessage, error)
			func Broker.ReadAggregationSet(qname, gname, aggregationSetID string) ([]*TaskMessage, time.Time, error)
			func github.com/hibiken/asynq/internal/rdb.(*RDB).Dequeue(qnames ...string) (msg *TaskMessage, leaseExpirationTime time.Time, err error)
			func github.com/hibiken/asynq/internal/rdb.(*RDB).ListLeaseExpired(cutoff time.Time, qnames ...string) ([]*TaskMessage, error)
			func github.com/hibiken/asynq/internal/rdb.(*RDB).ReadAggregationSet(qname, gname, setID string) ([]*TaskMessage, time.Time, error)
		As Inputs Of (at least 24)
			func EncodeMessage(msg *TaskMessage) ([]byte, error)
			func Broker.AddToGroup(ctx context.Context, msg *TaskMessage, gname string) error
			func Broker.AddToGroupUnique(ctx context.Context, msg *TaskMessage, groupKey string, ttl time.Duration) error
			func Broker.Archive(ctx context.Context, msg *TaskMessage, errMsg string) error
			func Broker.Done(ctx context.Context, msg *TaskMessage) error
			func Broker.Enqueue(ctx context.Context, msg *TaskMessage) error
			func Broker.EnqueueUnique(ctx context.Context, msg *TaskMessage, ttl time.Duration) error
			func Broker.MarkAsComplete(ctx context.Context, msg *TaskMessage) error
			func Broker.Requeue(ctx context.Context, msg *TaskMessage) error
			func Broker.Retry(ctx context.Context, msg *TaskMessage, processAt time.Time, errMsg string, isFailure bool) error
			func Broker.Schedule(ctx context.Context, msg *TaskMessage, processAt time.Time) error
			func Broker.ScheduleUnique(ctx context.Context, msg *TaskMessage, processAt time.Time, ttl time.Duration) error
			func github.com/hibiken/asynq/internal/context.New(base context.Context, msg *TaskMessage, deadline time.Time) (context.Context, context.CancelFunc)
			func github.com/hibiken/asynq/internal/rdb.(*RDB).AddToGroup(ctx context.Context, msg *TaskMessage, groupKey string) error
			func github.com/hibiken/asynq/internal/rdb.(*RDB).AddToGroupUnique(ctx context.Context, msg *TaskMessage, groupKey string, ttl time.Duration) error
			func github.com/hibiken/asynq/internal/rdb.(*RDB).Archive(ctx context.Context, msg *TaskMessage, errMsg string) error
			func github.com/hibiken/asynq/internal/rdb.(*RDB).Done(ctx context.Context, msg *TaskMessage) error
			func github.com/hibiken/asynq/internal/rdb.(*RDB).Enqueue(ctx context.Context, msg *TaskMessage) error
			func github.com/hibiken/asynq/internal/rdb.(*RDB).EnqueueUnique(ctx context.Context, msg *TaskMessage, ttl time.Duration) error
			func github.com/hibiken/asynq/internal/rdb.(*RDB).MarkAsComplete(ctx context.Context, msg *TaskMessage) error
			func github.com/hibiken/asynq/internal/rdb.(*RDB).Requeue(ctx context.Context, msg *TaskMessage) error
			func github.com/hibiken/asynq/internal/rdb.(*RDB).Retry(ctx context.Context, msg *TaskMessage, processAt time.Time, errMsg string, isFailure bool) error
			func github.com/hibiken/asynq/internal/rdb.(*RDB).Schedule(ctx context.Context, msg *TaskMessage, processAt time.Time) error
			func github.com/hibiken/asynq/internal/rdb.(*RDB).ScheduleUnique(ctx context.Context, msg *TaskMessage, processAt time.Time, ttl time.Duration) error

	 type TaskState int (basic type)
		TaskState denotes the state of a task.

		Methods (only one)
			( TaskState) String() string
		Implements (at least 2)
			 TaskState : expvar.Var
			 TaskState : fmt.Stringer
		As Outputs Of (at least one exported)
			func TaskStateFromString(s string) (TaskState, error)
		As Types Of (total 7)
			const TaskStateActive
			const TaskStateAggregating
			const TaskStateArchived
			const TaskStateCompleted
			const TaskStatePending
			const TaskStateRetry
			const TaskStateScheduled

	 type WorkerInfo (struct)
		WorkerInfo holds information about a running worker.

		Fields (total 9)
			Deadline time.Time
			Host string
			ID string
			PID int
			Payload []byte
			Queue string
			ServerID string
			Started time.Time
			Type string
		As Outputs Of (at least 2)
			func DecodeWorkerInfo(b []byte) (*WorkerInfo, error)
			func github.com/hibiken/asynq/internal/rdb.(*RDB).ListWorkers() ([]*WorkerInfo, error)
		As Inputs Of (at least 3)
			func EncodeWorkerInfo(info *WorkerInfo) ([]byte, error)
			func Broker.WriteServerState(info *ServerInfo, workers []*WorkerInfo, ttl time.Duration) error
			func github.com/hibiken/asynq/internal/rdb.(*RDB).WriteServerState(info *ServerInfo, workers []*WorkerInfo, ttl time.Duration) error

	 type Z (struct)
		Z represents sorted set member.

		Fields (total 2)
			Message *TaskMessage
			Score int64


Package-Level Functions (total 39)

	 func ActiveKey(qname string) string
		ActiveKey returns a redis key for the active tasks.

	 func AggregationSetKey(qname, gname, setID string) string
		AggregationSetKey returns a redis key used for an aggregation set.

	 func AllAggregationSets(qname string) string
		AllAggregationSets returns a redis key used to store all aggregation sets (set of tasks staged to be aggregated)
		in a given queue.

	 func AllGroups(qname string) string
		AllGroups return a redis key used to store all group keys used in a given queue.

	 func ArchivedKey(qname string) string
		ArchivedKey returns a redis key for the archived tasks.

	 func CompletedKey(qname string) string
	 func DecodeMessage(data []byte) (*TaskMessage, error)
		DecodeMessage unmarshals the given bytes and returns a decoded task message.

	 func DecodeSchedulerEnqueueEvent(b []byte) (*SchedulerEnqueueEvent, error)
		DecodeSchedulerEnqueueEvent unmarshals the given bytes
		and returns a decoded SchedulerEnqueueEvent.

	 func DecodeSchedulerEntry(b []byte) (*SchedulerEntry, error)
		DecodeSchedulerEntry unmarshals the given bytes and returns a decoded SchedulerEntry.

	 func DecodeServerInfo(b []byte) (*ServerInfo, error)
		DecodeServerInfo decodes the given bytes into ServerInfo.

	 func DecodeWorkerInfo(b []byte) (*WorkerInfo, error)
		DecodeWorkerInfo decodes the given bytes into WorkerInfo.

	 func EncodeMessage(msg *TaskMessage) ([]byte, error)
		EncodeMessage marshals the given task message and returns an encoded bytes.

	 func EncodeSchedulerEnqueueEvent(event *SchedulerEnqueueEvent) ([]byte, error)
		EncodeSchedulerEnqueueEvent marshals the given event
		and returns an encoded bytes.

	 func EncodeSchedulerEntry(entry *SchedulerEntry) ([]byte, error)
		EncodeSchedulerEntry marshals the given entry and returns an encoded bytes.

	 func EncodeServerInfo(info *ServerInfo) ([]byte, error)
		EncodeServerInfo marshals the given ServerInfo and returns the encoded bytes.

	 func EncodeWorkerInfo(info *WorkerInfo) ([]byte, error)
		EncodeWorkerInfo marshals the given WorkerInfo and returns the encoded bytes.

	 func FailedKey(qname string, t time.Time) string
		FailedKey returns a redis key for failure count for the given day for the queue.

	 func FailedTotalKey(qname string) string
		FailedTotalKey returns a redis key for total failure count for the given queue.

	 func GroupKey(qname, gkey string) string
		GroupKey returns a redis key used to group tasks belong in the same group.

	 func GroupKeyPrefix(qname string) string
		GroupKeyPrefix returns a prefix for group key.

	 func LeaseKey(qname string) string
		LeaseKey returns a redis key for the lease.

	 func NewCancelations() *Cancelations
		NewCancelations returns a Cancelations instance.

	 func NewLease(expirationTime time.Time) *Lease
	 func PausedKey(qname string) string
		PausedKey returns a redis key to indicate that the given queue is paused.

	 func PendingKey(qname string) string
		PendingKey returns a redis key for the given queue name.

	 func ProcessedKey(qname string, t time.Time) string
		ProcessedKey returns a redis key for processed count for the given day for the queue.

	 func ProcessedTotalKey(qname string) string
		ProcessedTotalKey returns a redis key for total processed count for the given queue.

	 func QueueKeyPrefix(qname string) string
		QueueKeyPrefix returns a prefix for all keys in the given queue.

	 func RetryKey(qname string) string
		RetryKey returns a redis key for the retry tasks.

	 func ScheduledKey(qname string) string
		ScheduledKey returns a redis key for the scheduled tasks.

	 func SchedulerEntriesKey(schedulerID string) string
		SchedulerEntriesKey returns a redis key for the scheduler entries given scheduler ID.

	 func SchedulerHistoryKey(entryID string) string
		SchedulerHistoryKey returns a redis key for the scheduler's history for the given entry.

	 func ServerInfoKey(hostname string, pid int, serverID string) string
		ServerInfoKey returns a redis key for process info.

	 func TaskKey(qname, id string) string
		TaskKey returns a redis key for the given task message.

	 func TaskKeyPrefix(qname string) string
		TaskKeyPrefix returns a prefix for task key.

	 func TaskStateFromString(s string) (TaskState, error)
	 func UniqueKey(qname, tasktype string, payload []byte) string
		UniqueKey returns a redis key with the given type, payload, and queue name.

	 func ValidateQueueName(qname string) error
		ValidateQueueName validates a given qname to be used as a queue name.
		Returns nil if valid, otherwise returns non-nil error.

	 func WorkersKey(hostname string, pid int, serverID string) string
		WorkersKey returns a redis key for the workers given hostname, pid, and server ID.


Package-Level Variables (only one)

	  var DefaultQueue string
		DefaultQueue is the redis key for the default queue.


Package-Level Constants (total 14)

	const AllQueues = "asynq:queues" // SET
		Global Redis keys.

	const AllSchedulers = "asynq:schedulers" // ZSET
		Global Redis keys.

	const AllServers = "asynq:servers" // ZSET
		Global Redis keys.

	const AllWorkers = "asynq:workers" // ZSET
		Global Redis keys.

	const CancelChannel = "asynq:cancel" // PubSub channel
		Global Redis keys.

	const DefaultQueueName = "default"
		DefaultQueueName is the queue name used if none are specified by user.

	const TaskStateActive TaskState = 1
	const TaskStateAggregating TaskState = 7 // describes a state where task is waiting in a group to be aggregated
	const TaskStateArchived TaskState = 5
	const TaskStateCompleted TaskState = 6
	const TaskStatePending TaskState = 2
	const TaskStateRetry TaskState = 4
	const TaskStateScheduled TaskState = 3
	const Version = "0.24.1"
		Version of asynq library and CLI.

The pages are generated with Golds v0.8.2. (GOOS=linux GOARCH=amd64)
Golds is a Go 101 project developed by Tapir Liu.
PR and bug reports are welcome and can be submitted to the issue list.
Please follow @zigo_101 (reachable from the left QR code) to get the latest news of Golds.