package main import ( "context" "fmt" "log" "time" "github.com/saravanasai/goqueue" "github.com/saravanasai/goqueue/config" ) // Define your job type EmailJob struct { To string `json:"to"` Subject string `json:"subject"` } // Implement the Job interface func (e EmailJob) Process(ctx context.Context) error { fmt.Printf("Sending email to %s: %s\n", e.To, e.Subject) return nil } // Register job type for serialization func init() { goqueue.RegisterJob("EmailJob", func() goqueue.Job { return &EmailJob{} }) } func main() { // Create a queue with in-memory backend cfg := config.NewInMemoryConfig() q, err := goqueue.NewQueueWithDefaults("email-queue", cfg) if err != nil { log.Fatal(err) } // Start worker pool with 2 concurrent workers ctx := context.Background() q.StartWorkers(ctx, 2) // Dispatch a job job := EmailJob{ To: "user@example.com", Subject: "Welcome to GoQueue!", } if err := q.Dispatch(job); err != nil { log.Printf("Failed to dispatch job: %v", err) } // Wait to see results (in production, workers would run continuously) time.Sleep(2 * time.Second) // Graceful shutdown shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() q.Shutdown(shutdownCtx) }

package main import ( "context" "fmt" "log" "github.com/saravanasai/goqueue" "github.com/saravanasai/goqueue/config" ) func main() { // Create a queue with PostgreSQL backend cfg := config.NewPostgresConfig("postgresql://user:password@localhost:5432/dbname") // Configure metrics callback (optional) cfg = cfg.WithMetricsCallback(func(metrics config.JobMetrics) { fmt.Printf("Job: %s, Queue: %s, Duration: %v, Error: %v\n", metrics.JobID, metrics.QueueName, metrics.Duration, metrics.Error) }) // Create the queue q, err := goqueue.NewQueueWithDefaults("emails", cfg) if err != nil { log.Fatalf("Failed to create queue: %v", err) } // Dispatch jobs and start workers as normal... }

cfg := config.NewInMemoryConfig() queue, err := goqueue.NewQueueWithDefaults("emails", cfg)

// PostgreSQL cfg := config.NewPostgresConfig("postgresql://user:password@localhost:5432/dbname") queue, err := goqueue.NewQueueWithDefaults("emails", cfg) // MySQL cfg := config.NewMySQLConfig("root:password@tcp(localhost:3306)/dbname") queue, err := goqueue.NewQueueWithDefaults("emails", cfg)

cfg := config.NewRedisConfig( "localhost:6379", // Redis server address "", // Password (if any) 0, // Database number ) queue, err := goqueue.NewQueueWithDefaults("emails", cfg)

cfg := config.NewSQSConfig( "https://sqs.us-west-2.amazonaws.com/123456789012/my-queue", // Queue URL "us-west-2", // AWS Region "xxx", // Access key "xxxx", // Secret key ) queue, err := goqueue.NewQueueWithDefaults("notifications", cfg)

// Basic configuration cfg := config.NewRedisConfig("localhost:6379", "", 0) // Worker options cfg = cfg.WithMaxWorkers(5) // Number of worker goroutines cfg = cfg.WithConcurrencyLimit(10) // Maximum concurrent jobs // Retry behavior cfg = cfg.WithMaxRetryAttempts(3) // Max retry attempts for failed jobs cfg = cfg.WithRetryDelay(5 * time.Second) // Base delay between retries cfg = cfg.WithExponentialBackoff(true) // Use increasing delays between retries // Metrics collection cfg = cfg.WithMetricsCallback(func(metrics config.JobMetrics) { log.Printf("Job %s completed in %v with error: %v", metrics.JobID, metrics.Duration, metrics.Error) })

// Schedule a job to run 5 minutes from now delay := 5 * time.Minute if err := q.DispatchWithDelay(job, delay); err != nil { log.Printf("Failed to schedule job: %v", err) }

// Available built-in middleware loggingMiddleware := middleware.LoggingMiddleware(logger) // Structured logging skipCondition := middleware.ConditionalSkipMiddleware(func(jobCtx *job.JobContext) bool { // Skip processing based on job attributes return jobCtx.EnqueuedAt.Before(time.Now().Add(-24 * time.Hour)) }) // Custom middleware rateLimiter := func(next middleware.HandlerFunc) middleware.HandlerFunc { return func(ctx context.Context, jobCtx *job.JobContext) error { // Implement rate limiting logic here fmt.Println("Rate limiting job execution") return next(ctx, jobCtx) } } // Apply middleware through configuration (order matters - first added, first executed) cfg = cfg.WithMiddleware(loggingMiddleware) cfg = cfg.WithMiddleware(skipCondition) cfg = cfg.WithMiddleware(rateLimiter) // Or add multiple middlewares at once cfg = cfg.WithMiddlewares(loggingMiddleware, skipCondition, rateLimiter)

// Custom handler for failed jobs type CustomDLQ struct{} func (d *CustomDLQ) Push(ctx context.Context, jobCtx *job.JobContext, err error) error { // Store job data and error in database fmt.Printf("Job %s failed after %d attempts: %v\n", jobCtx.JobID, jobCtx.Attempts, err) // Send alerts or notifications // alertService.Send(fmt.Sprintf("Critical job failure: %s", jobCtx.JobID)) return nil } // Register custom DLQ cfg = cfg.WithDLQAdapter(&CustomDLQ{})

// Register metrics callback cfg = cfg.WithMetricsCallback(func(metrics config.JobMetrics) { // Record metrics in your monitoring system (Prometheus, etc.) fmt.Printf("Job: %s, Queue: %s, Duration: %v, Error: %v\n", metrics.JobID, metrics.QueueName, metrics.Duration, metrics.Error) // Track success/failure rates if metrics.Error != nil { failureCounter.Inc() } else { successCounter.Inc() } })