refactor: hq source package select statements (#579)

Author: willmhowes

internal/pkg/source/hq/consumer.go

@@ -44,28 +44,23 @@ func (s *HQ) consumer() {
 	go s.consumerSender(ctx, &wg, urlBuffer)
 
 	// Wait for shutdown signal
-	for {
-		select {
-		case <-s.ctx.Done():
-			logger.Debug("received done signal")
+	<-s.ctx.Done()
+	logger.Debug("received done signal")
 
-			// Cancel the context to stop all goroutines.
-			cancel()
+	// Cancel the context to stop all goroutines.
+	cancel()
 
-			logger.Debug("waiting for goroutines to finish")
+	logger.Debug("waiting for goroutines to finish")
 
-			// Wait for all goroutines to finish
-			wg.Wait()
+	// Wait for all goroutines to finish
+	wg.Wait()
 
-			// Close the urlBuffer to signal consumerSenders to finish
-			close(urlBuffer)
+	// Close the urlBuffer to signal consumerSenders to finish
+	close(urlBuffer)
 
-			s.wg.Done()
+	s.wg.Done()
 
-			logger.Debug("closed")
-			return
-		}
-	}
+	logger.Debug("closed")
 }
 
 func (s *HQ) consumerFetcher(ctx context.Context, wg *sync.WaitGroup, urlBuffer chan<- *gocrawlhq.URL, batchSize int) {
@@ -182,11 +177,9 @@ func (s *HQ) consumerSender(ctx context.Context, wg *sync.WaitGroup, urlBuffer <
 			err = reactor.ReceiveInsert(newItem)
 			if err != nil {
 				if err == reactor.ErrReactorFrozen {
-					select {
-					case <-ctx.Done():
-						logger.Debug("closed while sending to frozen reactor")
-						return
-					}
+					<-ctx.Done()
+					logger.Debug("closed while sending to frozen reactor")
+					return
 				}
 				panic(err)
 			}

internal/pkg/source/hq/finisher.go

@@ -40,28 +40,23 @@ func (s *HQ) finisher() {
 	go s.finisherDispatcher(ctx, &wg, batchCh)
 
 	// Wait for the context to be canceled.
-	for {
-		select {
-		case <-s.ctx.Done():
-			logger.Debug("received done signal")
+	<-s.ctx.Done()
+	logger.Debug("received done signal")
 
-			// Cancel the context to stop all goroutines.
-			cancel()
+	// Cancel the context to stop all goroutines.
+	cancel()
 
-			logger.Debug("waiting for goroutines to finish")
+	logger.Debug("waiting for goroutines to finish")
 
-			// Wait for the finisher and dispatcher to finish.
-			wg.Wait()
+	// Wait for the finisher and dispatcher to finish.
+	wg.Wait()
 
-			// Close the batch channel to signal the dispatcher to finish.
-			close(batchCh)
+	// Close the batch channel to signal the dispatcher to finish.
+	close(batchCh)
 
-			s.wg.Done()
+	s.wg.Done()
 
-			logger.Debug("closed")
-			return
-		}
-	}
+	logger.Debug("closed")
 }
 
 // finisherReceiver reads URLs from finishCh, accumulates them into batches, and sends the batches to batchCh.

internal/pkg/source/hq/producer.go

@@ -39,28 +39,23 @@ func (s *HQ) producer() {
 	go s.producerDispatcher(ctx, &wg, batchCh)
 
 	// Wait for the context to be canceled.
-	for {
-		select {
-		case <-s.ctx.Done():
-			logger.Debug("received done signal")
+	<-s.ctx.Done()
+	logger.Debug("received done signal")
 
-			// Cancel the context to stop all goroutines.
-			cancel()
+	// Cancel the context to stop all goroutines.
+	cancel()
 
-			logger.Debug("waiting for goroutines to finish")
+	logger.Debug("waiting for goroutines to finish")
 
-			// Wait for the producer and dispatcher to finish.
-			wg.Wait()
+	// Wait for the producer and dispatcher to finish.
+	wg.Wait()
 
-			// Close the batch channel to signal the dispatcher to finish.
-			close(batchCh)
+	// Close the batch channel to signal the dispatcher to finish.
+	close(batchCh)
 
-			s.wg.Done()
+	s.wg.Done()
 
-			logger.Debug("closed")
-			return
-		}
-	}
+	logger.Debug("closed")
 }
 
 // producerReceiver reads URLs from produceCh, accumulates them into batches, and sends the batches to batchCh.