objectstorage

Documentation

Overview

Package objectstorage provides configuration and utilities for object storage operations.

Package objectstorage provides an object storage-based implementation of worker.Publisher and worker.Consumer using Go Cloud Development Kit (CDK) blob storage abstraction.

Index

• func GenerateTaskID() string
• func GenerateTaskKey(taskID string) string
• func ParseTimestampFromKey(key string) (int64, error)
• type Consumer
  • func NewConsumer(ctx context.Context, bucket *blob.Bucket, messageFormat worker.MessageFormat, ...) (*Consumer, error)
  • func (c *Consumer) Consume(handler worker.TaskHandlerFunc) error
• type Option
  • func WithIntervalExpFactor(factor float64) Option
  • func WithMaxInterval(interval time.Duration) Option
  • func WithMaxItemsToReadAtOnce(maxItems int) Option
  • func WithMinInterval(interval time.Duration) Option
  • func WithPrefix(prefix string) Option
  • func WithProcessingTimeout(timeout time.Duration) Option
  • func WithRetryAttempts(attempts int) Option
• type Publisher
  • func NewPublisher(bucket *blob.Bucket, messageFormat worker.MessageFormat, opts ...Option) (*Publisher, error)
  • func (p *Publisher) Publish(task *worker.Task) error
  • func (p *Publisher) PublishWithRetry(task *worker.Task) error

Functions

func GenerateTaskID

func GenerateTaskID() string

GenerateTaskID creates a new UUID for task identification.

func GenerateTaskKey

func GenerateTaskKey(taskID string) string

GenerateTaskKey creates a timestamp-based key for task storage. Format: {timestamp_nanos}_{task_id} This ensures lexicographic ordering for FIFO processing and collision resistance.

func ParseTimestampFromKey

func ParseTimestampFromKey(key string) (int64, error)

ParseTimestampFromKey extracts the timestamp from a task key. Returns the timestamp in nanoseconds and any parsing error.

Types

type Consumer

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

Consumer implements worker.Consumer using object storage via Go CDK blob.Bucket.

func NewConsumer

func NewConsumer(
	ctx context.Context,
	bucket *blob.Bucket,
	messageFormat worker.MessageFormat,
	opts ...Option,
) (*Consumer, error)

NewConsumer creates a new object storage consumer that implements worker.Consumer. It accepts a context, Go CDK blob.Bucket, message format, and optional configuration.

func (*Consumer) Consume

func (c *Consumer) Consume(handler worker.TaskHandlerFunc) error

Consume implements worker.Consumer.Consume by continuously polling for objects and processing them in FIFO order based on timestamp.

type Option

type Option func(*config)

Option is a function that configures the object storage config.

func WithIntervalExpFactor

func WithIntervalExpFactor(factor float64) Option

WithIntervalExpFactor sets the exponential factor to multiply the interval by when no tasks are found.

func WithMaxInterval

func WithMaxInterval(interval time.Duration) Option

WithMaxInterval sets the maximum duration for exponential backoff polling interval.

func WithMaxItemsToReadAtOnce

func WithMaxItemsToReadAtOnce(maxItems int) Option

WithMaxItemsToReadAtOnce sets the maximum number of items to read in one batch.

func WithMinInterval

func WithMinInterval(interval time.Duration) Option

WithMinInterval sets the minimum duration for exponential backoff polling interval.

func WithPrefix

func WithPrefix(prefix string) Option

WithPrefix sets the object key prefix.

func WithProcessingTimeout

func WithProcessingTimeout(timeout time.Duration) Option

WithProcessingTimeout sets the processing timeout for task batches.

func WithRetryAttempts

func WithRetryAttempts(attempts int) Option

WithRetryAttempts sets the number of retry attempts for failed operations.

type Publisher

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

Publisher implements worker.Publisher using object storage via Go CDK blob.Bucket.

func NewPublisher

func NewPublisher(bucket *blob.Bucket, messageFormat worker.MessageFormat, opts ...Option) (*Publisher, error)

NewPublisher creates a new object storage publisher that implements worker.Publisher. It accepts a Go CDK blob.Bucket for storage abstraction and optional configuration.

func (*Publisher) Publish

func (p *Publisher) Publish(task *worker.Task) error

Publish implements worker.Publisher.Publish by serializing the task and uploading it as an object with a timestamp-based key for FIFO ordering.

func (*Publisher) PublishWithRetry

func (p *Publisher) PublishWithRetry(task *worker.Task) error

PublishWithRetry publishes a task with retry logic based on the configuration.