func (q *Queue) Shrink() {
	if q.GetIthLength(q.mq.popIndex) == 0 && q.mq.pushIndex > 0 {
		q.mq.notifier = q.mq.notifier[q.mq.popIndex+1:]
		q.mq.total--
		q.mq.pushIndex--
		logger.Log.Error("WorkerQueue Stats", zap.Any("popindex", q.mq.popIndex), zap.Any("queueSize", len(q.mq.notifier[q.mq.popIndex])), zap.Any("totalChannelList", q.mq.total))
	}
}
Shrink shrinks the queue
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]))
}
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)
			}
		}
	}()
}
func processing(ctx context.Context, q *Queue, worker worker.Worker, jobC JobChan) {
	for {
		if jobC.jobData.Status == job.JobStatusFAILED {
			err := q.repo.UpdateJobStatusFailed(ctx, dbgen.UpdateJobStatusFailedParams{JobID: jobC.jobId, Status: "FAILED", EndTime: sql.NullTime{Time: time.Now(), Valid: true}, Comments: jobC.jobData.Comments})
			if err != nil {
				logger.Log.Error("Error update status to failed for job: %s  requeued", zap.Error(err))
				q.PushJob(ctx, jobC.jobData, jobC.jobData.Type.String)
			}
			break
		}
		err := worker.DoWork(ctx, &jobC.jobData)
		if err != nil {
			if jobC.jobData.RetryCount.Int32 < int32(q.retries) && int32(q.retries) != 0 {
				q.repo.UpdateJobStatusRetry(ctx, dbgen.UpdateJobStatusRetryParams{
					JobID:  jobC.jobId,
					Status: "RETRY",
				})
				jobC.jobData.RetryCount.Int32++
				logger.Log.Error("Error In Retries ", zap.Error(err), zap.Int32("jobID", jobC.jobId), zap.Int32("retryCount", jobC.jobData.RetryCount.Int32))
				time.Sleep(q.baseDelay * time.Millisecond)
			} else if jobC.jobData.Type.String == "DEFER_WORKER" {
				q.PushJob(ctx, jobC.jobData, jobC.jobData.Type.String)
			} else {
				logger.Log.Error("Retries execceded for ", zap.Int32("jobId", jobC.jobData.JobID))
				err = q.repo.UpdateJobStatusFailed(ctx, dbgen.UpdateJobStatusFailedParams{JobID: jobC.jobId, Status: "FAILED", EndTime: sql.NullTime{Time: time.Now(), Valid: true}, Comments: sql.NullString{String: err.Error(), Valid: true}, RetryCount: sql.NullInt32{Int32: jobC.jobData.RetryCount.Int32, Valid: true}})
				if err != nil {
					logger.Log.Error("Error update status to failed for job: %s", zap.Error(err))
					q.PushJob(ctx, jobC.jobData, jobC.jobData.Type.String)
				}
				break
			}
		} else {
			err = q.repo.UpdateJobStatusCompleted(ctx, dbgen.UpdateJobStatusCompletedParams{JobID: jobC.jobId, Status: "COMPLETED", EndTime: sql.NullTime{Time: time.Now(), Valid: true}})
			if err != nil {
				logger.Log.Error("Failed to Update job", zap.Error(err))
			}
			break
		}
	}
}
func (q *Queue) PushJob(ctx context.Context, j job.Job, workerName string) (int32, error) {
	q.mq.mux.Lock()
	if q.GetLength() == q.GetCapacity() && q.IsMultiQueue {
		q.Grow()
	}
	q.mq.notifier[q.mq.pushIndex] <- JobChan{j.JobID, workerName, j, ctx}
	logger.Log.Debug("queue info", zap.Any("queueno", q.mq.pushIndex), zap.Any("queueLen", q.GetLength()))
	q.mq.mux.Unlock()
	return 0, nil
}
func (q *Queue) GetIthLength(i int) int32 {
	return int32(len(q.mq.notifier[i]))
}