fix: Wait for specified time duration before shutdown

CHANGELOG.md

@@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- Fixed processor to wait for specified time duration before forcefully shutdown workers.
+
+
 ## [0.13.0] - 2020-10-13
 
 ### Added

processor.go

@@ -88,22 +88,23 @@ func newProcessor(params processorParams) *processor {
 		orderedQueues = sortByPriority(queues)
 	}
 	return &processor{
-		logger:         params.logger,
+		logger:          params.logger,
-		broker:         params.broker,
+		broker:          params.broker,
-		queueConfig:    queues,
+		queueConfig:     queues,
-		orderedQueues:  orderedQueues,
+		orderedQueues:   orderedQueues,
-		retryDelayFunc: params.retryDelayFunc,
+		retryDelayFunc:  params.retryDelayFunc,
-		syncRequestCh:  params.syncCh,
+		syncRequestCh:   params.syncCh,
-		cancelations:   params.cancelations,
+		cancelations:    params.cancelations,
-		errLogLimiter:  rate.NewLimiter(rate.Every(3*time.Second), 1),
+		errLogLimiter:   rate.NewLimiter(rate.Every(3*time.Second), 1),
-		sema:           make(chan struct{}, params.concurrency),
+		sema:            make(chan struct{}, params.concurrency),
-		done:           make(chan struct{}),
+		done:            make(chan struct{}),
-		quit:           make(chan struct{}),
+		quit:            make(chan struct{}),
-		abort:          make(chan struct{}),
+		abort:           make(chan struct{}),
-		errHandler:     params.errHandler,
+		errHandler:      params.errHandler,
-		handler:        HandlerFunc(func(ctx context.Context, t *Task) error { return fmt.Errorf("handler not set") }),
+		handler:         HandlerFunc(func(ctx context.Context, t *Task) error { return fmt.Errorf("handler not set") }),
-		starting:       params.starting,
+		shutdownTimeout: params.shutdownTimeout,
-		finished:       params.finished,
+		starting:        params.starting,
+		finished:        params.finished,
 	}
 }