func (q *Queue) RegisterWorker(ctx context.Context, w worker.Worker) {
	ctx, cancelFunc := context.WithCancel(ctx)
	c := make(chan os.Signal, 1)
	signal.Notify(c, os.Interrupt)
	go func() {
		for range c {
			cancelFunc()
			q.wg.Wait()
			q.workers = nil
		}
	}()
	q.workers[w.ID()] = append(q.workers[w.ID()], w)
	q.wg.Add(1)
	go func() {
		var jobC JobChan
		var ok bool
		for {
			if q.IsWorkerRegCompleted == false {
				time.Sleep(5 * time.Second)
				continue
			}
			select {
			case <-ctx.Done():
				q.wg.Done()
				return
			case jobC, ok = <-q.mq.notifier[q.mq.popIndex]:
				if ok == false && q.IsMultiQueue {
					q.mq.mux.Lock()
					q.Shrink()
					q.mq.mux.Unlock()
					continue
				}
				logger.Log.Debug("job is poped, queue stats", zap.Any("popIndex", q.mq.popIndex), zap.Any("queueSize", len(q.mq.notifier[q.mq.popIndex])))
				md, _ := metadata.FromIncomingContext(jobC.ctx)
				ctx = metadata.NewOutgoingContext(ctx, md)
				ctx = metadata.NewIncomingContext(ctx, md)
				lenWorker := len(q.workers[jobC.workerName])
				if lenWorker == 0 {
					continue
				}
				worker := q.workers[jobC.workerName][rand.Intn(lenWorker)]
				if jobC.jobData.JobID == 0 {
					repoJob := job.ToRepoJob(&jobC.jobData)
					jobID, err := q.repo.CreateJob(ctx, dbgen.CreateJobParams{Type: repoJob.Type, Status: repoJob.Status, Data: repoJob.Data,
						Comments: repoJob.Comments, StartTime: repoJob.StartTime, EndTime: repoJob.EndTime})
					if err != nil {
						logger.Log.Error("Unable to push job to db: %s, requeueing the job", zap.Error(err))
						q.PushJob(ctx, jobC.jobData, jobC.workerName)
						continue
					}
					jobC.jobId = jobID
					jobC.jobData.JobID = jobID
				}
				processing(ctx, q, worker, jobC)
			default:
				time.Sleep(q.PollRate)
			}
		}
	}()
}
RegisterWorker contains the main loop for all Workers.
package workerqueue
import (
	"context"
	"database/sql"
	"math/rand"
	"optisam-backend/common/optisam/logger"
	"optisam-backend/common/optisam/workerqueue/job"
	repoInterface "optisam-backend/common/optisam/workerqueue/repository"
	repo "optisam-backend/common/optisam/workerqueue/repository/postgres"
	dbgen "optisam-backend/common/optisam/workerqueue/repository/postgres/db"
	"optisam-backend/common/optisam/workerqueue/worker"
	"os"
	"os/signal"
	"sync"
	"time"
	"go.uber.org/zap"
	metadata "google.golang.org/grpc/metadata"
)
type JobChan struct {
	jobId      int32
	workerName string
	jobData    job.Job
	ctx        context.Context
}
type mlQueue struct {
	notifier  []chan JobChan
	pushIndex int
	popIndex  int
	total     int
	mux       sync.Mutex
}
type Queue struct {
	IsWorkerRegCompleted bool
	IsMultiQueue bool
	ID string
	repo repoInterface.Workerqueue
	mq mlQueue
	queueSize int
	retries int
	baseDelay time.Duration
	workers map[string][]worker.Worker
	wg *sync.WaitGroup
	PollRate time.Duration
}
func NewQueue(ctx context.Context, queueID string, db *sql.DB, conf QueueConfig) (*Queue, error) {
	q := &Queue{ID: queueID}
	q.repo = repo.NewRepository(db)
	q.PollRate = time.Duration(100 * time.Millisecond)  
	q.queueSize = 10000                                 
	q.retries = 3                                       
	q.IsMultiQueue = conf.IsMultiQueue
	q.baseDelay = time.Duration(3 * time.Second)  
	if conf.PollingRate > 0 {
		q.PollRate = conf.PollingRate
	}
	if conf.Qsize > 0 {
		q.queueSize = conf.Qsize
	}
	if conf.BaseDelay > 0 {
		q.baseDelay = conf.BaseDelay
	}
	if conf.Retries >= 0 {
		q.retries = conf.Retries
	}
	temp := mlQueue{}
	temp.notifier = make([]chan JobChan, 1)
	temp.notifier[0] = make(chan JobChan, q.queueSize)
	temp.total = 1
	q.mq = temp
	m := make(map[string][]worker.Worker)
	q.workers = m
	var wg sync.WaitGroup
	q.wg = &wg
	err := q.ResumePendingJobs(ctx)
	if err != nil {
		logger.Log.Error("Unable to resume jobs from bucket: %s", zap.Error(err))
	}
	return q, nil
}
func (q *Queue) Close(ctx context.Context) {
	q.wg.Wait()
	for i := 0; i < q.mq.total; i++ {
		close(q.mq.notifier[i])
	}
	q.workers = nil
}
func (q *Queue) GetRetries() int32 {
	return int32(q.retries)
}
func (q *Queue) GetLength() int32 {
	return int32(len(q.mq.notifier[q.mq.pushIndex]))
}
func (q *Queue) GetCapacity() int32 {
	return int32(cap(q.mq.notifier[q.mq.pushIndex]))
}