filequeue

package
v1.30.3 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jan 4, 2026 License: GPL-3.0 Imports: 19 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrQueueEmpty        = errors.New("queue is empty")
	ErrQueueItemNotFound = errors.New("queue item not found")
)

Errors for the queue

View Source 
var (
	ErrQueueFileEmpty        = fmt.Errorf("queue file is empty")
	ErrQueueFileItemNotFound = fmt.Errorf("queue file item not found")
)

Errors for the queue file

Functions

func New 

func New(baseDir string) execution.QueueStore

Types

type DualQueue 

type DualQueue struct {
	dirlock.DirLock // Embed DirLock to ensure thread-safe access to the queue files
	// contains filtered or unexported fields
}

DualQueue represents a queue for storing dag-runs with two priorities: high and low. It uses two queue files to store the items.

func NewDualQueue 

func NewDualQueue(baseDir, name string) *DualQueue

NewDualQueue creates a new queue with the specified base directory and name It initializes the queue files for high and low priority

func (*DualQueue) Dequeue 

func (q *DualQueue) Dequeue(ctx context.Context) (execution.QueuedItemData, error)

Dequeue retrieves a dag-run from the queue and removes it. It checks the high-priority queue first, then the low-priority queue

func (*DualQueue) DequeueByDAGRunID 

func (q *DualQueue) DequeueByDAGRunID(ctx context.Context, dagRun execution.DAGRunRef) ([]execution.QueuedItemData, error)

DequeueByDAGRunID retrieves a dag-run from the queue by its dag-run ID

func (*DualQueue) Enqueue 

func (q *DualQueue) Enqueue(ctx context.Context, priority execution.QueuePriority, dagRun execution.DAGRunRef) error

Enqueue adds a dag-run to the queue with the specified priority

func (*DualQueue) FindByDAGRunID 

func (q *DualQueue) FindByDAGRunID(ctx context.Context, dagRunID string) (execution.QueuedItemData, error)

FindByDAGRunID retrieves a dag-run from the queue by its dag-run ID without removing it. It returns the first found item in the queue files. If the item is not found in any of the queue files, it returns ErrQueueItemNotFound.

func (*DualQueue) Len 

func (q *DualQueue) Len(ctx context.Context) (int, error)

Len returns the total number of items in the queue

func (*DualQueue) List 

func (q *DualQueue) List(ctx context.Context) ([]execution.QueuedItemData, error)

List returns all items in the queue

type ItemData 

type ItemData struct {
	FileName string              `json:"fileName"`
	DAGRun   execution.DAGRunRef `json:"dagRun"`
	QueuedAt time.Time           `json:"queuedAt"`
}

ItemData represents the data stored in the queue file

type Job 

type Job struct {
	ItemData
	// contains filtered or unexported fields
}

Job implements execution.QueuedItemData for a job stored in a file.

func (*Job) Data 

func (j *Job) Data() (*execution.DAGRunRef, error)

func (*Job) ID 

func (j *Job) ID() string

type QueueFile 

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

QueueFile is a simple queue implementation using files It stores the queued items in JSON files in a specified directory The timestamp is in UTC and the milliseconds are added to the filename Since this relies on the file system, it is not thread-safe and should be accessed by a single process at a time.

func NewQueueFile 

func NewQueueFile(baseDir, prefix string) *QueueFile

NewQueueFile constructs a QueueFile configured to use baseDir for storage and prefix to identify queue item files. The returned QueueFile includes a compiled regular expression that matches queue filenames of the form item_<prefix><YYYYMMDD_HHMMSS>_<nanoseconds>Z_<dagRunID>.json.

func (*QueueFile) FindByDAGRunID 

func (q *QueueFile) FindByDAGRunID(ctx context.Context, dagRunID string) (*QueuedFile, error)

FindByDAGRunID finds a job by its dag-run ID without removing it from the queue.

func (*QueueFile) Len 

func (q *QueueFile) Len(ctx context.Context) (int, error)

Len returns the number of items in the queue

func (*QueueFile) List 

func (q *QueueFile) List(ctx context.Context) ([]*QueuedFile, error)

func (*QueueFile) Pop 

func (q *QueueFile) Pop(ctx context.Context) (execution.QueuedItemData, error)

func (*QueueFile) PopByDAGRunID 

func (q *QueueFile) PopByDAGRunID(ctx context.Context, dagRun execution.DAGRunRef) ([]execution.QueuedItemData, error)

PopByDAGRunID removes jobs from the queue by dag-run ID

func (*QueueFile) Push 

func (q *QueueFile) Push(ctx context.Context, dagRun execution.DAGRunRef) error

Push adds a job to the queue Since it's a prototype, it just create a json file with the job ID and dag-run reference

type QueuedFile added in v1.26.0 

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

func NewQueuedFile added in v1.26.0 

func NewQueuedFile(file string) *QueuedFile

NewQueuedFile creates a QueuedFile for the given queue file path. The returned QueuedFile's id is derived from the base filename (the filename without its extension) and its file field is set to the provided path. It does not read or validate the file contents.

func (*QueuedFile) Data added in v1.26.0 

func (j *QueuedFile) Data() (*execution.DAGRunRef, error)

Data implements execution.QueuedItemData.

func (*QueuedFile) ExtractJob added in v1.26.0 

func (j *QueuedFile) ExtractJob() (*Job, error)

ExtractJob loads and returns the underlying QueuedItemData.

func (*QueuedFile) ID added in v1.26.0 

func (j *QueuedFile) ID() string

ID implements execution.QueuedItemData.

type Store 

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

Store implements models.QueueStore. It provides a dead-simple queue implementation using files. Since implementing a queue is not trivial, this implementation provides as a prototype for a more complex queue implementation.

func (*Store) All 

func (s *Store) All(ctx context.Context) ([]execution.QueuedItemData, error)

All implements models.QueueStore.

func (*Store) BaseDir 

func (s *Store) BaseDir() string

BaseDir returns the base directory of the queue store

func (*Store) DequeueByDAGRunID 

func (s *Store) DequeueByDAGRunID(ctx context.Context, name string, dagRun execution.DAGRunRef) ([]execution.QueuedItemData, error)

DequeueByDAGRunID implements models.QueueStore.

func (*Store) DequeueByName 

func (s *Store) DequeueByName(ctx context.Context, name string) (execution.QueuedItemData, error)

DequeueByName implements models.QueueStore.

func (*Store) Enqueue 

func (s *Store) Enqueue(ctx context.Context, name string, p execution.QueuePriority, dagRun execution.DAGRunRef) error

Enqueue implements models.QueueStore.

func (*Store) Len 

func (s *Store) Len(ctx context.Context, name string) (int, error)

Len implements models.QueueStore.

func (*Store) List 

func (s *Store) List(ctx context.Context, name string) ([]execution.QueuedItemData, error)

List implements models.QueueStore.

func (*Store) ListByDAGName added in v1.22.0 

func (s *Store) ListByDAGName(ctx context.Context, name, dagName string) ([]execution.QueuedItemData, error)

func (*Store) QueueList added in v1.24.8 

func (s *Store) QueueList(ctx context.Context) ([]string, error)

QueueList lists all queue names that have at least one item in the queue.

func (*Store) QueueWatcher added in v1.24.8 

func (s *Store) QueueWatcher(_ context.Context) execution.QueueWatcher

QueueWatcher implements execution.QueueStore.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.