Merge pull request #6 from BranchIntl/fix-logging-interface

Remove seelog and use the new slog standard logger

Author: shashankmehra

.golangci.yml

@@ -180,7 +180,6 @@ engine := core.NewEngine(
 	core.WithPollInterval(5*time.Second),        // Polling frequency
 	core.WithShutdownTimeout(30*time.Second),    // Graceful shutdown timeout
 	core.WithJobBufferSize(100),                 // Job channel buffer
-	core.WithExitOnComplete(false),              // Exit when queues empty
 )
 ```
 
@@ -202,6 +201,10 @@ options.Exchange = "jobs"
 options.PrefetchCount = 1
 ```
 
+## Logging
+
+goworker2 uses Go's standard `log/slog` library for structured logging. By default, it uses the default slog logger. To customize logging, configure your logger before creating the engine using `slog.SetDefault(logger)`. For example: `slog.SetDefault(slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo})))`.
+
 ## Worker Functions
 
 Worker functions must match this signature:

README.md

@@ -3,12 +3,12 @@ package rabbitmq
 import (
 	"context"
 	"fmt"
+	"log/slog"
 	"time"
 
 	"github.com/BranchIntl/goworker2/core"
 	"github.com/BranchIntl/goworker2/errors"
 	"github.com/BranchIntl/goworker2/job"
-	"github.com/cihub/seelog"
 	amqp "github.com/rabbitmq/amqp091-go"
 )
 
@@ -21,7 +21,6 @@ type RabbitMQBroker struct {
 	declaredQueues map[string]bool   // Track declared queues
 	consumerQueues []string          // Queues to consume from
 	consumerTags   map[string]string // Track consumer tags
-	logger         seelog.LoggerInterface
 }
 
 // NewBroker creates a new RabbitMQ broker
@@ -104,11 +103,6 @@ func (r *RabbitMQBroker) Type() string {
 	return "rabbitmq"
 }
 
-// SetLogger sets the logger for the broker
-func (r *RabbitMQBroker) SetLogger(logger seelog.LoggerInterface) {
-	r.logger = logger
-}
-
 // Capabilities returns RabbitMQ broker capabilities
 func (r *RabbitMQBroker) Capabilities() core.BrokerCapabilities {
 	return core.BrokerCapabilities{
@@ -331,16 +325,9 @@ func (r *RabbitMQBroker) ensureQueue(name string) error {
 	return nil
 }
 
-// logError logs an error message if logger is available
-func (r *RabbitMQBroker) logError(format string, args ...interface{}) {
-	if r.logger != nil {
-		r.logger.Errorf(format, args...)
-	}
-}
-
 // Start implements the Poller interface for push-based consumption
 func (r *RabbitMQBroker) Start(ctx context.Context, jobChan chan<- job.Job) error {
-	r.logger.Infof("Starting RabbitMQ consumer for queues: %v", r.consumerQueues)
+	slog.Info("Starting RabbitMQ consumer", "queues", r.consumerQueues)
 
 	for _, queue := range r.consumerQueues {
 		if err := r.ensureQueue(queue); err != nil {
@@ -370,7 +357,7 @@ func (r *RabbitMQBroker) Start(ctx context.Context, jobChan chan<- job.Job) erro
 
 	// Keep running until context is cancelled
 	<-ctx.Done()
-	r.logger.Info("RabbitMQ consumer stopped")
+	slog.Info("RabbitMQ consumer stopped")
 	close(jobChan)
 	return nil
 }
@@ -383,7 +370,7 @@ func (r *RabbitMQBroker) handleDeliveries(ctx context.Context, queue string, del
 			return
 		case delivery, ok := <-deliveries:
 			if !ok {
-				r.logger.Warnf("Delivery channel closed for queue %s", queue)
+				slog.Warn("Delivery channel closed", "queue", queue)
 				return
 			}
 
@@ -394,11 +381,11 @@ func (r *RabbitMQBroker) handleDeliveries(ctx context.Context, queue string, del
 				case <-ctx.Done():
 					// Put job back on queue
 					if err := delivery.Nack(false, true); err != nil {
-						r.logger.Errorf("Failed to nack job during shutdown: %v", err)
+						slog.Error("Failed to nack job during shutdown", "error", err)
 					}
 					return
 				case jobChan <- job:
-					r.logger.Debugf("Job sent to workers: %s", job.GetClass())
+					slog.Debug("Job sent to workers", "class", job.GetClass())
 				}
 			}
 		}
@@ -422,9 +409,9 @@ func (r *RabbitMQBroker) convertDeliveryToJob(delivery amqp.Delivery, queue stri
 	if err != nil {
 		// Reject message if we can't deserialize it
 		if nackErr := delivery.Nack(false, false); nackErr != nil {
-			r.logError("Failed to nack message after deserialization error: %v", nackErr)
+			slog.Error("Failed to nack message after deserialization error", "error", nackErr)
 		}
-		r.logError("Failed to deserialize job: %v", err)
+		slog.Error("Failed to deserialize job", "error", err)
 		return nil
 	}
 

brokers/rabbitmq/broker.go

@@ -3,13 +3,13 @@ package redis
 import (
 	"context"
 	"fmt"
+	"log/slog"
 	"time"
 
 	"github.com/BranchIntl/goworker2/core"
 	"github.com/BranchIntl/goworker2/errors"
 	redisUtils "github.com/BranchIntl/goworker2/internal/redis"
 	"github.com/BranchIntl/goworker2/job"
-	"github.com/cihub/seelog"
 	"github.com/gomodule/redigo/redis"
 )
 
@@ -19,7 +19,6 @@ type RedisBroker struct {
 	namespace  string
 	options    Options
 	serializer core.Serializer
-	logger     seelog.LoggerInterface
 }
 
 // NewBroker creates a new Redis broker
@@ -61,7 +60,7 @@ func (r *RedisBroker) Close() error {
 	return nil
 }
 
-// Health checks the Redis connection health
+// Health checks Redis connection health
 func (r *RedisBroker) Health() error {
 	if r.pool == nil {
 		return errors.ErrNotConnected
@@ -83,11 +82,6 @@ func (r *RedisBroker) Type() string {
 	return "redis"
 }
 
-// SetLogger sets the logger for the broker
-func (r *RedisBroker) SetLogger(logger seelog.LoggerInterface) {
-	r.logger = logger
-}
-
 // Capabilities returns Redis broker capabilities
 func (r *RedisBroker) Capabilities() core.BrokerCapabilities {
 	return core.BrokerCapabilities{
@@ -123,7 +117,7 @@ func (r *RedisBroker) Enqueue(ctx context.Context, j job.Job) error {
 
 	// Add queue to set of known queues (best effort)
 	if _, err := conn.Do("SADD", r.queuesKey(), j.GetQueue()); err != nil {
-		r.logError("Failed to track queue %s: %v", j.GetQueue(), err)
+		slog.Error("Failed to track queue %s: %v", j.GetQueue(), err)
 	}
 
 	return nil
@@ -213,7 +207,7 @@ func (r *RedisBroker) DeleteQueue(ctx context.Context, name string) error {
 
 	// Remove from set of queues (best effort)
 	if _, err := conn.Do("SREM", r.queuesKey(), name); err != nil {
-		r.logError("Failed to remove queue %s from set: %v", name, err)
+		slog.Error("Failed to remove queue %s from set: %v", name, err)
 	}
 
 	return nil
@@ -264,9 +258,3 @@ func (r *RedisBroker) queueKey(queue string) string {
 func (r *RedisBroker) queuesKey() string {
 	return fmt.Sprintf("%squeues", r.namespace)
 }
-
-func (r *RedisBroker) logError(format string, args ...interface{}) {
-	if r.logger != nil {
-		r.logger.Errorf(format, args...)
-	}
-}

brokers/redis/broker.go

@@ -3,6 +3,7 @@ package core
 import (
 	"context"
 	"fmt"
+	"log/slog"
 	"os"
 	"os/signal"
 	"sync"
@@ -11,7 +12,6 @@ import (
 
 	"github.com/BranchIntl/goworker2/errors"
 	"github.com/BranchIntl/goworker2/job"
-	"github.com/cihub/seelog"
 )
 
 // Engine is the main orchestration engine
@@ -27,8 +27,6 @@ type Engine struct {
 	ctx    context.Context
 	cancel context.CancelFunc
 	wg     sync.WaitGroup
-
-	logger seelog.LoggerInterface
 }
 
 // NewEngine creates a new engine with dependency injection
@@ -44,25 +42,19 @@ func NewEngine(
 		opt(config)
 	}
 
-	logger, _ := seelog.LoggerFromWriterWithMinLevel(config.LogOutput, config.LogLevel)
-
 	return &Engine{
 		broker:     broker,
 		stats:      stats,
 		registry:   registry,
 		serializer: serializer,
 		config:     config,
-		logger:     logger,
 	}
 }
 
 // Start begins processing jobs
 func (e *Engine) Start(ctx context.Context) error {
 	e.ctx, e.cancel = context.WithCancel(ctx)
 
-	// Set logger on broker
-	e.broker.SetLogger(e.logger)
-
 	// If broker implements SetConsumerQueues (for push-based consumption), set the queues
 	if setQueues, ok := e.broker.(interface{ SetConsumerQueues([]string) }); ok {
 		setQueues.SetConsumerQueues(e.config.Queues)
@@ -94,7 +86,6 @@ func (e *Engine) Start(ctx context.Context) error {
 			e.stats,
 			e.config.Queues,
 			e.config.PollInterval,
-			e.logger,
 		)
 	}
 
@@ -106,7 +97,6 @@ func (e *Engine) Start(ctx context.Context) error {
 		e.config.Concurrency,
 		e.config.Queues,
 		jobChan,
-		e.logger,
 		e.broker,
 	)
 
@@ -115,18 +105,18 @@ func (e *Engine) Start(ctx context.Context) error {
 	go func() {
 		defer e.wg.Done()
 		if err := poller.Start(e.ctx, jobChan); err != nil {
-			e.logger.Errorf("Poller error: %v", err)
+			slog.Error("Poller error", "error", err)
 		}
 	}()
 
 	go func() {
 		defer e.wg.Done()
 		if err := e.workerPool.Start(e.ctx); err != nil {
-			e.logger.Errorf("Worker pool error: %v", err)
+			slog.Error("Worker pool error", "error", err)
 		}
 	}()
 
-	e.logger.Info("Engine started")
+	slog.Info("Engine started")
 	return nil
 }
 
@@ -145,18 +135,18 @@ func (e *Engine) Stop() error {
 
 	select {
 	case <-done:
-		e.logger.Info("Engine stopped gracefully")
+		slog.Info("Engine stopped gracefully")
 	case <-time.After(e.config.ShutdownTimeout):
-		e.logger.Warn("Engine shutdown timeout exceeded")
+		slog.Warn("Engine shutdown timeout exceeded")
 	}
 
 	// Close connections
 	if err := e.broker.Close(); err != nil {
-		e.logger.Errorf("Error closing broker: %v", err)
+		slog.Error("Error closing broker", "error", err)
 	}
 
 	if err := e.stats.Close(); err != nil {
-		e.logger.Errorf("Error closing statistics: %v", err)
+		slog.Error("Error closing statistics", "error", err)
 	}
 
 	return nil
@@ -209,9 +199,9 @@ func (e *Engine) Run(ctx context.Context) error {
 	// Wait for either context cancellation or signal
 	select {
 	case <-ctx.Done():
-		e.logger.Info("Context cancelled, shutting down...")
+		slog.Info("Context cancelled, shutting down...")
 	case sig := <-sigChan:
-		e.logger.Infof("Received signal %v, shutting down...", sig)
+		slog.Info("Received signal, shutting down...", "signal", sig)
 	}
 
 	// Graceful shutdown

core/engine.go

@@ -289,7 +289,6 @@ func TestEngine_Integration_MultipleQueues(t *testing.T) {
 		WithQueues([]string{"high", "medium", "low"}),
 		WithConcurrency(2),
 		WithPollInterval(50*time.Millisecond),
-		WithStrictQueues(true), // Process in queue order
 	)
 
 	// Register workers

core/engine_test.go

@@ -2,11 +2,12 @@ package core
 
 import (
 	"context"
+	"log/slog"
+	"os"
 	"testing"
 	"time"
 
 	"github.com/BranchIntl/goworker2/job"
-	"github.com/cihub/seelog"
 )
 
 // TestSetup provides common test dependencies
@@ -15,17 +16,21 @@ type TestSetup struct {
 	Stats      *MockStatistics
 	Registry   *MockRegistry
 	Serializer *MockSerializer
-	Logger     seelog.LoggerInterface
 }
 
 // NewTestSetup creates a standard test setup with all mocks
 func NewTestSetup() *TestSetup {
+	// Set up a discard logger for tests to avoid noise
+	logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{
+		Level: slog.LevelError, // Only show errors in tests
+	}))
+	slog.SetDefault(logger)
+
 	return &TestSetup{
 		Broker:     NewMockBroker(),
 		Stats:      NewMockStatistics(),
 		Registry:   NewMockRegistry(),
 		Serializer: NewMockSerializer(),
-		Logger:     seelog.Disabled,
 	}
 }
 
@@ -133,7 +138,7 @@ func (b *WorkerBuilder) WithQueues(queues []string) *WorkerBuilder {
 
 // Build creates the worker
 func (b *WorkerBuilder) Build() *Worker {
-	return NewWorker(b.id, b.queues, b.setup.Registry, b.setup.Stats, b.setup.Logger, b.setup.Broker)
+	return NewWorker(b.id, b.queues, b.setup.Registry, b.setup.Stats, b.setup.Broker)
 }
 
 // PollerBuilder helps create pollers for testing
@@ -168,7 +173,7 @@ func (b *PollerBuilder) WithInterval(interval time.Duration) *PollerBuilder {
 
 // Build creates the poller
 func (b *PollerBuilder) Build() *StandardPoller {
-	return NewStandardPoller(b.setup.Broker, b.setup.Stats, b.queues, b.interval, b.setup.Logger)
+	return NewStandardPoller(b.setup.Broker, b.setup.Stats, b.queues, b.interval)
 }
 
 // WorkerPoolBuilder helps create worker pools for testing
@@ -204,7 +209,7 @@ func (b *WorkerPoolBuilder) WithQueues(queues []string) *WorkerPoolBuilder {
 // Build creates the worker pool
 func (b *WorkerPoolBuilder) Build() *WorkerPool {
 	return NewWorkerPool(b.setup.Registry, b.setup.Stats, b.setup.Serializer,
-		b.concurrency, b.queues, b.jobChan, b.setup.Logger, b.setup.Broker)
+		b.concurrency, b.queues, b.jobChan, b.setup.Broker)
 }
 
 // ErrorTestCase represents a common error test scenario