fixed global config issue

config/config.go

@@ -79,12 +79,6 @@ func (cp *configProvider) GetConfig() *Config {
 	return &cp.applicationConfig
 }
 
-// SetConfigProvider sets the global configuration provider.
-// This is useful for injecting custom configurations during tests.
-func SetConfigProvider(provider ConfigProvider) {
-	globalConfigProvider = provider
-}
-
 // GetConfigProvider returns the global configuration provider.
 func GetConfigProvider() ConfigProvider {
 	if globalConfigProvider == nil {

config/mongodb.go

@@ -4,6 +4,8 @@ import "go.mongodb.org/mongo-driver/mongo"
 
 // MongoDBConfig ...
 type MongoDBConfig struct {
-	Client   *mongo.Client
-	Database string
+	Client      *mongo.Client
+	URI         string
+	MaxPoolSize uint64 // Maximum number of connections in the pool
+	DbName      string
 }

example/consumer/main.go

@@ -2,16 +2,20 @@ package main
 
 import (
 	"context"
+	"encoding/json"
 	"github.com/sirupsen/logrus"
 	"github.com/surendratiwari3/paota/config"
 	"github.com/surendratiwari3/paota/schema"
 	"github.com/surendratiwari3/paota/workerpool"
-
 	//"github.com/surendratiwari3/paota/example/task"
 	"github.com/surendratiwari3/paota/logger"
 	"os"
 )
 
+type printWorker struct {
+	workerPool workerpool.Pool
+}
+
 func main() {
 	logrusLog := logrus.StandardLogger()
 	logrusLog.SetFormatter(&logrus.JSONFormatter{})
@@ -24,7 +28,7 @@ func main() {
 		//Store:         "null",
 		TaskQueueName: "paota_task_queue",
 		AMQP: &config.AMQPConfig{
-			Url:                "amqp://localhost:5672/",
+			Url:                "amqp://guest:guest@localhost:5672/",
 			Exchange:           "paota_task_exchange",
 			ExchangeType:       "direct",
 			BindingKey:         "paota_task_binding_key",
@@ -47,10 +51,13 @@ func main() {
 		logger.ApplicationLogger.Info("workerPool is nil")
 		os.Exit(0)
 	}
+
+	printWorker := printWorker{workerPool: newWorkerPool}
+
 	logger.ApplicationLogger.Info("newWorkerPool created successfully")
 	// Register tasks
 	regTasks := map[string]interface{}{
-		"Print": Print,
+		"Print": printWorker.Print,
 	}
 	err = newWorkerPool.RegisterTasks(regTasks)
 	if err != nil {
@@ -65,7 +72,46 @@ func main() {
 	}
 }
 
-func Print(arg *schema.Signature) error {
+func (wp printWorker) Publish() {
+	// UserRecord represents the structure of user records.
+	type UserRecord struct {
+		ID    string `json:"id"`
+		Name  string `json:"name"`
+		Email string `json:"email"`
+		// Add other fields as needed
+	}
+	// Replace this with the received user record
+	user := UserRecord{
+		ID:    "1",
+		Name:  "John Doe",
+		Email: "john.doe@example.com",
+	}
+
+	// Convert the struct to a JSON string
+	userJSON, err := json.Marshal(user)
+	if err != nil {
+		//
+	}
+
+	printJob := &schema.Signature{
+		Name: "Print",
+		Args: []schema.Arg{
+			{
+				Type:  "string",
+				Value: string(userJSON),
+			},
+		},
+		RetryCount:                  10,
+		RoutingKey:                  "consumer_publisher",
+		IgnoreWhenTaskNotRegistered: true,
+	}
+
+	wp.workerPool.SendTaskWithContext(context.Background(), printJob)
+
+}
+
+func (wp printWorker) Print(arg *schema.Signature) error {
 	logger.ApplicationLogger.Info("success")
+	wp.Publish()
 	return nil
 }

example/consumer_publisher/main.go

@@ -0,0 +1,124 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"github.com/sirupsen/logrus"
+	"github.com/surendratiwari3/paota/config"
+	"github.com/surendratiwari3/paota/schema"
+	"github.com/surendratiwari3/paota/workerpool"
+	"sync"
+
+	//"github.com/surendratiwari3/paota/example/task"
+	"github.com/surendratiwari3/paota/logger"
+	"os"
+)
+
+type printWorker struct {
+	workerPool workerpool.Pool
+}
+
+func main() {
+	logrusLog := logrus.StandardLogger()
+	logrusLog.SetFormatter(&logrus.JSONFormatter{})
+	logrusLog.SetReportCaller(true)
+
+	logger.ApplicationLogger = logrusLog
+
+	cnf := config.Config{
+		Broker: "amqp",
+		//Store:         "null",
+		TaskQueueName: "paota_task_queue",
+		AMQP: &config.AMQPConfig{
+			Url:                "amqp://guest:guest@localhost:5672/",
+			Exchange:           "paota_task_exchange",
+			ExchangeType:       "direct",
+			BindingKey:         "consumer_publisher",
+			PrefetchCount:      100,
+			ConnectionPoolSize: 10,
+			DelayedQueue:       "delay_consumer_publisher",
+		},
+	}
+
+	newWorkerPool, err := workerpool.NewWorkerPoolWithConfig(context.Background(), 10, "testWorker", cnf)
+	if err != nil {
+		logger.ApplicationLogger.Error("workerPool is not created", err)
+		os.Exit(0)
+	} else if newWorkerPool == nil {
+		logger.ApplicationLogger.Info("workerPool is nil")
+		os.Exit(0)
+	}
+
+	printWorker := printWorker{workerPool: newWorkerPool}
+
+	logger.ApplicationLogger.Info("newWorkerPool created successfully")
+	// Register tasks
+	regTasks := map[string]interface{}{
+		"Print": printWorker.Print,
+	}
+	err = newWorkerPool.RegisterTasks(regTasks)
+	if err != nil {
+		logger.ApplicationLogger.Info("error while registering task")
+		return
+	}
+	logger.ApplicationLogger.Info("Worker is also started")
+
+	err = newWorkerPool.Start()
+	if err != nil {
+		logger.ApplicationLogger.Error("error while starting worker")
+	}
+}
+
+func (wp printWorker) Publish() {
+	// UserRecord represents the structure of user records.
+	type UserRecord struct {
+		ID    string `json:"id"`
+		Name  string `json:"name"`
+		Email string `json:"email"`
+		// Add other fields as needed
+	}
+	// Replace this with the received user record
+	user := UserRecord{
+		ID:    "1",
+		Name:  "John Doe",
+		Email: "john.doe@example.com",
+	}
+
+	// Convert the struct to a JSON string
+	userJSON, err := json.Marshal(user)
+	if err != nil {
+		//
+	}
+
+	printJob := &schema.Signature{
+		Name: "Print",
+		Args: []schema.Arg{
+			{
+				Type:  "string",
+				Value: string(userJSON),
+			},
+		},
+		RetryCount:                  10,
+		RoutingKey:                  "consumer_publisher_",
+		IgnoreWhenTaskNotRegistered: true,
+	}
+
+	waitGrp := sync.WaitGroup{}
+	waitGrp.Add(1)
+		go func() {
+			for i := 0; i < 100; i++ {
+				wp.workerPool.SendTaskWithContext(context.Background(), printJob)
+			}
+			waitGrp.Done()
+		}()
+	}
+
+	waitGrp.Wait()
+
+}
+
+func (wp printWorker) Print(arg *schema.Signature) error {
+	logger.ApplicationLogger.Info("success")
+	wp.Publish()
+	return nil
+}

internal/store/backend_interface.go → internal/backend/backend_interface.go

@@ -1,13 +1,13 @@
-package store
+package backend
 
 import (
 	"github.com/surendratiwari3/paota/schema"
 )
 
 // Backend - a common interface for all result Store
 type Backend interface {
-	// InsertTask Insert task state to result backend
-	InsertTask(signature schema.Signature) error
+	// StoreTask Insert task state to result backend
+	StoreTask(signature schema.Signature) error
 	// DeleteTask Delete task state from result backend
 	DeleteTask(taskId string) error
 	// GetTask Get task state from result backend

internal/broker/amqp/broker.go

@@ -64,8 +64,8 @@ func (b *AMQPBroker) isDirectExchange() bool {
 // NewAMQPBroker creates a new instance of the AMQP broker
 // It opens connections to RabbitMQ, declares an exchange, opens a channel,
 // declares and binds the queue, and enables publish notifications
-func NewAMQPBroker() (broker.Broker, error) {
-	cfg := config.GetConfigProvider().GetConfig()
+func NewAMQPBroker(configProvider config.ConfigProvider) (broker.Broker, error) {
+	cfg := configProvider.GetConfig()
 	amqpErrorChannel := make(chan *amqp.Error, 1)
 	stopChannel := make(chan struct{})
 	doneStopChannel := make(chan struct{})