Move retryDelay to projection notification processor

Author: Sergiu Ghitea

driver/sql/projection_notification_processor.go

@@ -23,6 +23,8 @@ type (
 
 		logger  goengine.Logger
 		metrics Metrics
+
+		retryDelay time.Duration
 	}
 
 	// ProcessHandler is a func used to trigger a notification but with the addition of providing a Trigger func so
@@ -31,7 +33,13 @@ type (
 )
 
 // newBackgroundProcessor create a new projectionNotificationProcessor
-func newBackgroundProcessor(queueProcessors, queueBuffer int, logger goengine.Logger, metrics Metrics) (*projectionNotificationProcessor, error) {
+func newBackgroundProcessor(
+	queueProcessors,
+	queueBuffer int,
+	logger goengine.Logger,
+	metrics Metrics,
+	retryDelay time.Duration,
+) (*projectionNotificationProcessor, error) {
 	if queueProcessors <= 0 {
 		return nil, errors.New("queueProcessors must be greater then zero")
 	}
@@ -50,6 +58,7 @@ func newBackgroundProcessor(queueProcessors, queueBuffer int, logger goengine.Lo
 		queueBuffer:     queueBuffer,
 		logger:          logger,
 		metrics:         metrics,
+		retryDelay:      retryDelay,
 	}, nil
 }
 

driver/sql/projector_aggregate.go

@@ -25,8 +25,6 @@ type AggregateProjector struct {
 	db *sql.DB
 
 	logger goengine.Logger
-
-	retryDelay time.Duration
 }
 
 // NewAggregateProjector creates a new projector for a projection
@@ -63,7 +61,7 @@ func NewAggregateProjector(
 		e.String("projection", projection.Name())
 	})
 
-	processor, err := newBackgroundProcessor(10, 32, logger, metrics)
+	processor, err := newBackgroundProcessor(10, 32, logger, metrics, retryDelay)
 	if err != nil {
 		return nil, err
 	}
@@ -96,8 +94,6 @@ func NewAggregateProjector(
 		db: db,
 
 		logger: logger,
-
-		retryDelay: retryDelay,
 	}, nil
 }
 
@@ -166,7 +162,6 @@ func (a *AggregateProjector) processNotification(
 		return nil
 	case errorRetry:
 		a.logger.Debug("ProcessHandler->ErrorHandler: re-queueing notification", logFields)
-		notification.ValidAfter = time.Now().Add(a.retryDelay)
 		return queue(ctx, notification)
 	}