Merge pull request #1 from violetpay-org/refactor/structure

Refactor/structure

Author: asheswook

config/config.go

@@ -0,0 +1,108 @@
+package queuemanagerconfig
+
+import (
+	"github.com/IBM/sarama"
+	"time"
+)
+
+type RedisOpts func(*RedisConfig)
+type RedisConfig struct {
+	QueueName string
+	Retry     int
+	TTL       int
+	Addrs     []string
+}
+
+func SetRedisQueueName(queueName string) RedisOpts {
+	return func(c *RedisConfig) {
+		c.QueueName = queueName
+	}
+}
+
+func AddRedisBroker(broker string) RedisOpts {
+	return func(c *RedisConfig) {
+		c.Addrs = append(c.Addrs, broker)
+	}
+}
+
+func AddRedisBrokers(brokers []string) RedisOpts {
+	return func(c *RedisConfig) {
+		c.Addrs = append(c.Addrs, brokers...)
+	}
+}
+
+func AddRedisTTL(ttl int) RedisOpts {
+	return func(c *RedisConfig) {
+		c.TTL = ttl
+	}
+}
+
+func AddRedisRetry(retry int) RedisOpts {
+	return func(c *RedisConfig) {
+		c.Retry = retry
+	}
+}
+
+func NewRedisConfig(opts ...RedisOpts) *RedisConfig {
+	conf := RedisConfig{
+		QueueName: "",
+		Retry:     0,
+		TTL:       0,
+		Addrs:     []string{},
+	}
+
+	for _, opt := range opts {
+		opt(&conf)
+	}
+
+	return &conf
+}
+
+var (
+	TestKafkaBrokers   = []string{}
+	TestKafkaQueueName = "test_queue_2"
+)
+
+type KafkaOpts func(*KafkaConfig)
+type KafkaConfig struct {
+	Conf    *sarama.Config
+	Brokers []string
+	Topic   string
+}
+
+func AddKafkaBroker(broker string) KafkaOpts {
+	return func(c *KafkaConfig) {
+		c.Brokers = append(c.Brokers, broker)
+	}
+}
+
+func AddKafkaBrokers(brokers []string) KafkaOpts {
+	return func(c *KafkaConfig) {
+		c.Brokers = append(c.Brokers, brokers...)
+	}
+}
+
+func SetKafkaTopic(topic string) KafkaOpts {
+	return func(c *KafkaConfig) {
+		c.Topic = topic
+	}
+}
+
+func NewKafkaConfig(opts ...KafkaOpts) *KafkaConfig {
+	conf := KafkaConfig{
+		Conf:    sarama.NewConfig(),
+		Brokers: []string{"kafka.vp-datacenter-1.violetpay.net:9092", "kafka.vp-datacenter-1.violetpay.net:9093"},
+	}
+
+	for _, opt := range opts {
+		opt(&conf)
+	}
+
+	conf.Conf.Producer.Partitioner = sarama.NewRoundRobinPartitioner
+	conf.Conf.Producer.Return.Successes = true
+	conf.Conf.Producer.Flush.Frequency = 500 * time.Millisecond
+	conf.Conf.Producer.RequiredAcks = sarama.WaitForAll
+
+	// conf.Conf.Producer.Flush.Bytes = 1024 * 1024 * 10 // 10 MB
+	return &conf
+}

services/queue_name.go renamed to config/name.go

@@ -1,7 +1,7 @@
-package qmanservices
+package queuemanagerconfig
 
 import (
-	qmanErr "github.com/violetpay-org/point3-quman/errors"
+	"github.com/violetpay-org/queuemanager/internal/queueerror"
 )
 
 // QueueName is an enumeration of the different queues that the service can use.
@@ -11,11 +11,11 @@ type QueueName int
 
 var (
 	// map of int to string
-	queue_names map[int]string = make(map[int]string)
+	queueNames map[int]string = make(map[int]string)
 )
 
 func (q QueueName) String() string {
-	name, _ := queue_names[int(q)]
+	name, _ := queueNames[int(q)]
 	return name
 }
 
@@ -24,10 +24,10 @@ func (q QueueName) GetQueueName() string {
 }
 
 func RegisterQueueName(name string, index int) (QueueName, error) {
-	if _, ok := queue_names[index]; !ok {
-		queue_names[index] = name
+	if _, ok := queueNames[index]; !ok {
+		queueNames[index] = name
 		return QueueName(index), nil
 	}
 
-	return QueueName(0), qmanErr.ErrDuplicateQueueName()
+	return QueueName(0), queueerror.ErrDuplicateQueueName()
 }

point3-quman.go renamed to factory.go

@@ -1,11 +1,11 @@
-package quman
+package queuemanager
 
 import (
 	"context"
+	"github.com/violetpay-org/queuemanager/config"
+	"github.com/violetpay-org/queuemanager/internal/queueerror"
+	innerqueue "github.com/violetpay-org/queuemanager/queue"
 	"sync"
-
-	qmanErr "github.com/violetpay-org/point3-quman/errors"
-	qmanservices "github.com/violetpay-org/point3-quman/services"
 )
 
 // Main package for Quman (Queue Manager) service
@@ -23,44 +23,44 @@ func NewQueueFactory(wg *sync.WaitGroup, ctx *context.Context) *MainQueueFactory
 	}
 }
 
-func (f *MainQueueFactory) RegisterQueueName(name string, index int) (qmanservices.QueueName, error) {
-	return qmanservices.RegisterQueueName(name, index)
+func (f *MainQueueFactory) RegisterQueueName(name string, index int) (queuemanagerconfig.QueueName, error) {
+	return queuemanagerconfig.RegisterQueueName(name, index)
 }
 
 func (f *MainQueueFactory) GetWaitGroup() (*sync.WaitGroup, error) {
 	if f.wg == nil {
-		return nil, qmanErr.ErrQueueFactoryWaitGroupNil()
+		return nil, queueerror.ErrQueueFactoryWaitGroupNil()
 	}
 
 	return f.wg, nil
 }
 
-func (f *MainQueueFactory) GetQueueService(queueName qmanservices.QueueName) (qmanservices.IQueueService, error) {
+func (f *MainQueueFactory) GetQueueService(queueName queuemanagerconfig.QueueName) (innerqueue.Service, error) {
 	queue := f.queueSet[queueName]
 
 	if queue == nil {
-		return nil, qmanErr.ErrQueueNotFound(queueName.GetQueueName())
+		return nil, queueerror.ErrQueueNotFound(queueName.GetQueueName())
 	}
 
 	return queue.QueueService, nil
 }
 
-func (f *MainQueueFactory) RunQueue(queueName qmanservices.QueueName) (*sync.WaitGroup, error) {
+func (f *MainQueueFactory) RunQueue(queueName queuemanagerconfig.QueueName) (*sync.WaitGroup, error) {
 	if f.wg == nil {
-		return nil, qmanErr.ErrQueueFactoryWaitGroupNil()
+		return nil, queueerror.ErrQueueFactoryWaitGroupNil()
 	}
 
 	queue := f.queueSet[queueName]
 
 	if queue == nil {
-		return f.wg, qmanErr.ErrQueueNotFound(queueName.GetQueueName())
+		return f.wg, queueerror.ErrQueueNotFound(queueName.GetQueueName())
 	}
 
 	queueOperator := queue.LowLevelQueueOperator
 	queueCallbacks := queue.ConsumeCallback
 
 	if (queueOperator == nil) || (queueCallbacks == nil) {
-		return f.wg, qmanErr.ErrQueueNotPrepared(queueName.GetQueueName())
+		return f.wg, queueerror.ErrQueueNotPrepared(queueName.GetQueueName())
 	}
 
 	err := queueOperator.StartQueue(
@@ -72,32 +72,32 @@ func (f *MainQueueFactory) RunQueue(queueName qmanservices.QueueName) (*sync.Wai
 	return f.wg, err
 }
 
-func (f *MainQueueFactory) StopQueue(queueName qmanservices.QueueName) (*sync.WaitGroup, error) {
+func (f *MainQueueFactory) StopQueue(queueName queuemanagerconfig.QueueName) (*sync.WaitGroup, error) {
 	if f.wg == nil {
-		return nil, qmanErr.ErrQueueFactoryWaitGroupNil()
+		return nil, queueerror.ErrQueueFactoryWaitGroupNil()
 	}
 
 	queue := f.queueSet[queueName]
 
 	if queue == nil {
-		return f.wg, qmanErr.ErrQueueNotFound(queueName.GetQueueName())
+		return f.wg, queueerror.ErrQueueNotFound(queueName.GetQueueName())
 	}
 
 	queueOperator := queue.LowLevelQueueOperator
 	queueCallbacks := queue.StopCallback
 
 	if queueOperator == nil || queueCallbacks == nil {
-		return f.wg, qmanErr.ErrQueueNotPrepared(queueName.GetQueueName())
+		return f.wg, queueerror.ErrQueueNotPrepared(queueName.GetQueueName())
 	}
 
 	err := queueOperator.StopQueue(queueCallbacks)
 
 	return f.wg, err
 }
 
-func (f *MainQueueFactory) AddQueue(queueName qmanservices.QueueName, queue *Queue) error {
+func (f *MainQueueFactory) AddQueue(queueName queuemanagerconfig.QueueName, queue *Queue) error {
 	if f.queueSet[queueName] != nil {
-		return qmanErr.ErrDuplicateQueue()
+		return queueerror.ErrDuplicateQueue()
 	}
 
 	f.queueSet[queueName] = queue
@@ -108,7 +108,7 @@ func (f *MainQueueFactory) AddAllQueues(queueSet QueueSet) error {
 	// Check for duplicates
 	for queueName := range queueSet {
 		if f.queueSet[queueName] != nil {
-			return qmanErr.ErrDuplicateQueue()
+			return queueerror.ErrDuplicateQueue()
 		}
 	}
 
@@ -120,7 +120,7 @@ func (f *MainQueueFactory) AddAllQueues(queueSet QueueSet) error {
 	return nil
 }
 
-func (f *MainQueueFactory) UpsertQueue(queueName qmanservices.QueueName, queue *Queue) error {
+func (f *MainQueueFactory) UpsertQueue(queueName queuemanagerconfig.QueueName, queue *Queue) error {
 
 	_, err := f.GetQueueService(queueName)
 
@@ -145,7 +145,7 @@ func (f *MainQueueFactory) UpsertQueue(queueName qmanservices.QueueName, queue *
 
 func (f *MainQueueFactory) RunAllQueues() (*sync.WaitGroup, error) {
 	var erroredQueue interface{}
-	queuesExecuted := []qmanservices.QueueName{}
+	queuesExecuted := []queuemanagerconfig.QueueName{}
 
 	for queueName := range f.queueSet {
 		_, err := f.RunQueue(queueName)
@@ -163,7 +163,7 @@ func (f *MainQueueFactory) RunAllQueues() (*sync.WaitGroup, error) {
 			_, _ = f.StopQueue(queueName)
 		}
 
-		return f.wg, qmanErr.ErrQueueNotPrepared(erroredQueue.(qmanservices.QueueName).GetQueueName())
+		return f.wg, queueerror.ErrQueueNotPrepared(erroredQueue.(queuemanagerconfig.QueueName).GetQueueName())
 	}
 
 	return f.wg, nil

go.mod

@@ -1,4 +1,4 @@
-module github.com/violetpay-org/point3-quman
+module github.com/violetpay-org/queuemanager
 
 go 1.19
 

hub.go

@@ -0,0 +1,39 @@
+package queuemanager
+
+import (
+	"github.com/violetpay-org/queuemanager/config"
+	"github.com/violetpay-org/queuemanager/internal/queue/kafka"
+	"github.com/violetpay-org/queuemanager/internal/queue/redis"
+	"github.com/violetpay-org/queuemanager/item"
+)
+
+// NewRedisHub is a function that returns a new Hub.
+func NewRedisHub(
+	messageSerializer queueitem.RedisSerializer,
+	config *queuemanagerconfig.RedisConfig,
+	logger func(string),
+) *redis.Hub {
+	return redis.NewHub(
+		messageSerializer,
+		config,
+		logger,
+	)
+}
+
+// NewKafkaHub is a function that returns a new Hub.
+// maxConsumerCount is the maximum number of consumers that can be created.
+func NewKafkaHub(
+	maxConsumerCount int,
+	messageSerializer queueitem.KafkaSerializer,
+	publishOnly bool,
+	config *queuemanagerconfig.KafkaConfig,
+	logger func(string),
+) *kafka.Hub {
+	return kafka.NewHub(
+		maxConsumerCount,
+		messageSerializer,
+		publishOnly,
+		config,
+		logger,
+	)
+}