Move done channel to NotificationQueuer

Author: Christian Wygoda

driver/sql/projection_notification_processor.go

@@ -84,9 +84,7 @@ func (b *ProjectionNotificationProcessor) Execute(ctx context.Context, handler P
 
 // Start starts the background processes that will call the ProcessHandler based on the notification queued by Exec
 func (b *ProjectionNotificationProcessor) Start(ctx context.Context, handler ProcessHandler) func() {
-	b.done = make(chan struct{})
-
-	b.notificationQueue.Open(b.done)
+	b.done = b.notificationQueue.Open()
 
 	var wg sync.WaitGroup
 	wg.Add(b.queueProcessors)

driver/sql/projection_notification_processor_test.go

@@ -68,14 +68,19 @@ func TestStartProcessor(t *testing.T) {
 
 			e.Queue(gomock.Eq(ctx), gomock.Eq(notification)).Times(queueCallCount)
 			e.ReQueue(gomock.Eq(ctx), gomock.Eq(notification)).Times(reQueueCallCount)
-			e.Open(gomock.Any()).AnyTimes()
+			done := make(chan struct{})
+			e.Open().DoAndReturn(func() chan struct{} {
+				return done
+			}).AnyTimes()
 			channel := make(chan *sql.ProjectionNotification, 1)
 			channel <- notification
 			e.Channel().Return(channel).AnyTimes()
 			e.PutBack(gomock.Eq(notification)).Do(func(notification *sql.ProjectionNotification) {
 				channel <- notification
 			}).AnyTimes()
-			e.Close()
+			e.Close().Do(func() {
+				close(channel)
+			})
 
 			bufferSize := 1
 			queueProcessorsCount := 1

driver/sql/projection_notification_queue.go

@@ -18,7 +18,7 @@ type (
 		Channel() chan *ProjectionNotification
 		Close()
 		Empty() bool
-		Open(chan struct{})
+		Open() chan struct{}
 		PutBack(*ProjectionNotification)
 		Queue(context.Context, *ProjectionNotification) error
 		ReQueue(context.Context, *ProjectionNotification) error
@@ -90,7 +90,9 @@ func (nq *NotificationQueue) ReQueue(ctx context.Context, notification *Projecti
 }
 
 // Open enables the queue for business
-func (nq *NotificationQueue) Open(done chan struct{}) {
-	nq.done = done
+func (nq *NotificationQueue) Open() chan struct{} {
+	nq.done = make(chan struct{})
 	nq.queue = make(chan *ProjectionNotification, nq.queueBuffer)
+
+	return nq.done
 }