Merge pull request #5 from surendratiwari3/enhancement-issue-3

Enhancement issue 3

.circleci/config.yml

@@ -16,7 +16,7 @@ jobs:
           command: export PATH=$PATH:$GOPATH/bin
       - run:
           name: Generate Mocks
-          command: mockery --all --output=mocks
+          command: mockery --inpackage --all
       - run:
           name: Run Unit Tests
           command: |

.gitignore

@@ -25,4 +25,6 @@ go.work
 
 #ignore editor folder
 .idea/
-example/main
+example/main
+
+**/mock_*.go

adapter/amqp_adapter_test.go

@@ -5,13 +5,12 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/surendratiwari3/paota/config"
 	"github.com/surendratiwari3/paota/errors"
-	"github.com/surendratiwari3/paota/mocks"
 	"testing"
 )
 
 func TestNewAMQPAdapterWithNilConfig(t *testing.T) {
 	// Create a mock ConfigProvider with a nil config
-	mockConfigProvider := new(mocks.ConfigProvider)
+	mockConfigProvider := new(config.MockConfigProvider)
 	mockConfigProvider.On("GetConfig").Return(nil)
 
 	// Set the mock ConfigProvider for testing
@@ -29,7 +28,7 @@ func TestNewAMQPAdapterWithNilConfig(t *testing.T) {
 
 func TestNewAMQPAdapterWithValidConfig(t *testing.T) {
 	// Create a mock ConfigProvider with a valid config
-	mockConfigProvider := new(mocks.ConfigProvider)
+	mockConfigProvider := new(config.MockConfigProvider)
 	mockConfig := &config.Config{AMQP: &config.AMQPConfig{}}
 	mockConfigProvider.On("GetConfig").Return(mockConfig)
 
@@ -74,7 +73,7 @@ func TestCreateConnection_ConnectionRefused(t *testing.T) {
 
 func TestCreateConnection_NilAMQPConfig(t *testing.T) {
 	// Create a mock AMQPAdapter with a valid config
-	mockConfigProvider := new(mocks.ConfigProvider)
+	mockConfigProvider := new(config.MockConfigProvider)
 	mockConfigProvider.On("GetConfig").Return(&config.Config{}, nil)
 
 	config.SetConfigProvider(mockConfigProvider)
@@ -91,7 +90,7 @@ func TestCreateConnection_NilAMQPConfig(t *testing.T) {
 
 func TestCreateConnectionPool_ConnectionError(t *testing.T) {
 	// Create a mock AMQPAdapter with a valid config
-	mockConfigProvider := new(mocks.ConfigProvider)
+	mockConfigProvider := new(config.MockConfigProvider)
 	mockConfigProvider.On("GetConfig").Return(&config.Config{
 		AMQP: &config.AMQPConfig{
 			Url:                "amqp://localhost:5672",
@@ -108,7 +107,7 @@ func TestCreateConnectionPool_ConnectionError(t *testing.T) {
 
 func TestCreateConnectionPool_InvalidConnectionPoolSize(t *testing.T) {
 	// Create a mock AMQPAdapter with a valid config
-	mockConfigProvider := new(mocks.ConfigProvider)
+	mockConfigProvider := new(config.MockConfigProvider)
 	mockConfigProvider.On("GetConfig").Return(&config.Config{
 		AMQP: &config.AMQPConfig{
 			Url:                "amqp://localhost:5672",

api/handlers/task_handler.go

@@ -0,0 +1,61 @@
+package handlers
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"github.com/go-playground/validator/v10"
+	"github.com/google/uuid"
+	"github.com/labstack/echo/v4"
+	"github.com/surendratiwari3/paota/errors"
+	"github.com/surendratiwari3/paota/logger"
+	"github.com/surendratiwari3/paota/schema"
+	"github.com/surendratiwari3/paota/workerpool"
+	"io/ioutil"
+	"net/http"
+)
+
+// AddTaskHandlerV1 handles the request to add a task for API version 1
+func AddTaskHandlerV1(c echo.Context) error {
+	// Parse request body to get the Signature
+	body, err := ioutil.ReadAll(c.Request().Body)
+	if err != nil {
+		return c.String(http.StatusInternalServerError, "Failed to read request body")
+	}
+
+	taskName := c.Param("task_name")
+
+	var signature schema.Signature
+	if err := json.Unmarshal(body, &signature); err != nil {
+		return c.String(http.StatusBadRequest, "Invalid JSON payload")
+	}
+	signature.Name = taskName
+	signature.UUID = uuid.NewString()
+	validate := validator.New(validator.WithRequiredStructEnabled())
+
+	// Validate the Signature
+	if err := validate.Struct(signature); err != nil {
+		return c.String(http.StatusBadRequest, fmt.Sprintf("Invalid Signature: %s", err.Error()))
+	}
+
+	// Add the task based on the received Signature and task_name
+	err = EnqueueTask(&signature)
+	if err != nil {
+		return c.String(http.StatusInternalServerError, fmt.Sprintf("Failed to add task: %s", err.Error()))
+	}
+
+	return c.String(http.StatusOK, "Task added successfully")
+}
+
+func EnqueueTask(job *schema.Signature) error {
+	newWorkerPool, err := workerpool.NewWorkerPool(context.Background(), 10, "testWorker")
+	if err != nil {
+		logger.ApplicationLogger.Error("workerPool is not created", err)
+		return err
+	} else if newWorkerPool == nil {
+		logger.ApplicationLogger.Info("workerPool is nil")
+		return errors.ErrInvalidConfig
+	}
+	_, err = newWorkerPool.SendTaskWithContext(context.Background(), job)
+	return err
+}

api/main.go

@@ -0,0 +1,43 @@
+package api
+
+import (
+	"github.com/labstack/echo/v4"
+	"github.com/sirupsen/logrus"
+	"github.com/surendratiwari3/paota/api/handlers"
+	"github.com/surendratiwari3/paota/config"
+	"github.com/surendratiwari3/paota/logger"
+)
+
+func main() {
+	logrusLog := logrus.StandardLogger()
+	logrusLog.SetFormatter(&logrus.JSONFormatter{})
+	logger.ApplicationLogger = logrusLog
+
+	cnf := config.Config{
+		Broker: "amqp",
+		//Store:         "null",
+		TaskQueueName: "paota_task_queue",
+		AMQP: &config.AMQPConfig{
+			Url:                "amqp://guest:guest@localhost:55005/",
+			Exchange:           "paota_task_exchange",
+			ExchangeType:       "direct",
+			BindingKey:         "paota_task_binding_key",
+			PrefetchCount:      100,
+			ConnectionPoolSize: 10,
+		},
+	}
+	err := config.GetConfigProvider().SetApplicationConfig(cnf)
+	if err != nil {
+		logger.ApplicationLogger.Error("config error", err)
+		return
+	}
+
+	e := echo.New()
+
+	v1 := e.Group("/v1/paota")
+
+	v1.POST("/tasks/:task_name", handlers.AddTaskHandlerV1)
+
+	// Start the server
+	e.Start(":8090")
+}

bench/amqpnostorage/consumer/benchmark_test.go

@@ -5,7 +5,7 @@ import (
 	"github.com/sirupsen/logrus"
 	"github.com/surendratiwari3/paota/config"
 	"github.com/surendratiwari3/paota/logger"
-	"github.com/surendratiwari3/paota/task"
+	"github.com/surendratiwari3/paota/schema"
 	"github.com/surendratiwari3/paota/workerpool"
 	"os"
 	"testing"
@@ -68,6 +68,6 @@ func Consumer() {
 	newWorkerPool.Stop()
 }
 
-func ReturnNil(arg *task.Signature) error {
+func ReturnNil(arg *schema.Signature) error {
 	return nil
 }

bench/amqpnostorage/publisher/benchmark_test.go

@@ -6,7 +6,7 @@ import (
 	"github.com/sirupsen/logrus"
 	"github.com/surendratiwari3/paota/config"
 	"github.com/surendratiwari3/paota/logger"
-	"github.com/surendratiwari3/paota/task"
+	"github.com/surendratiwari3/paota/schema"
 	"github.com/surendratiwari3/paota/workerpool"
 	"os"
 	"testing"
@@ -82,9 +82,9 @@ func Publisher() {
 		//
 	}
 
-	returnNil := &task.Signature{
+	returnNil := &schema.Signature{
 		Name: "returnNil",
-		Args: []task.Arg{
+		Args: []schema.Arg{
 			{
 				Type:  "string",
 				Value: string(userJSON),
@@ -98,6 +98,6 @@ func Publisher() {
 	}
 }
 
-func ReturnNil(arg *task.Signature) error {
+func ReturnNil(arg *schema.Signature) error {
 	return nil
 }