Replace chan bool with chan struct{} (#777)

Slight perf optimization that also clearly indicates that the value given to channel does not mean anything

flow/activities/flowable.go

@@ -266,7 +266,7 @@ func (a *FlowableActivity) StartFlow(ctx context.Context,
 	})
 
 	defer func() {
-		shutdown <- true
+		shutdown <- struct{}{}
 	}()
 
 	syncStartTime := time.Now()
@@ -364,7 +364,7 @@ func (a *FlowableActivity) StartNormalize(
 		return fmt.Sprintf("normalizing records from batch for job - %s", input.FlowConnectionConfigs.FlowJobName)
 	})
 	defer func() {
-		shutdown <- true
+		shutdown <- struct{}{}
 	}()
 
 	log.Info("initializing table schema...")
@@ -443,7 +443,7 @@ func (a *FlowableActivity) GetQRepPartitions(ctx context.Context,
 	})
 
 	defer func() {
-		shutdown <- true
+		shutdown <- struct{}{}
 	}()
 
 	partitions, err := srcConn.GetQRepPartitions(config, last)
@@ -574,7 +574,7 @@ func (a *FlowableActivity) replicateQRepPartition(ctx context.Context,
 	})
 
 	defer func() {
-		shutdown <- true
+		shutdown <- struct{}{}
 	}()
 
 	res, err := dstConn.SyncQRepRecords(config, partition, stream)
@@ -618,7 +618,7 @@ func (a *FlowableActivity) ConsolidateQRepPartitions(ctx context.Context, config
 	})
 
 	defer func() {
-		shutdown <- true
+		shutdown <- struct{}{}
 	}()
 
 	err = dstConn.ConsolidateQRepPartitions(config)
@@ -919,7 +919,7 @@ func (a *FlowableActivity) ReplicateXminPartition(ctx context.Context,
 	})
 
 	defer func() {
-		shutdown <- true
+		shutdown <- struct{}{}
 	}()
 
 	res, err := dstConn.SyncQRepRecords(config, partition, stream)

flow/activities/snapshot_activity.go

@@ -21,7 +21,7 @@ func (a *SnapshotActivity) CloseSlotKeepAlive(flowJobName string) error {
 	}
 
 	if s, ok := a.SnapshotConnections[flowJobName]; ok {
-		s.signal.CloneComplete <- true
+		s.signal.CloneComplete <- struct{}{}
 		s.connector.Close()
 	}
 

flow/connectors/bigquery/qrep_avro_sync.go

@@ -325,7 +325,7 @@ func (s *QRepAvroSyncMethod) writeToStage(
 		},
 	)
 	defer func() {
-		shutdown <- true
+		shutdown <- struct{}{}
 	}()
 
 	var avroFile *avro.AvroFile

flow/connectors/eventhub/eventhub.go

@@ -227,7 +227,7 @@ func (c *EventHubConnector) SyncRecords(req *model.SyncRecordsRequest) (*model.S
 		)
 	})
 	defer func() {
-		shutdown <- true
+		shutdown <- struct{}{}
 	}()
 
 	// if env var PEERDB_BETA_EVENTHUB_PUSH_ASYNC=true

flow/connectors/postgres/cdc.go

@@ -209,7 +209,7 @@ func (p *PostgresCDCSource) consumeStream(
 	})
 
 	defer func() {
-		shutdown <- true
+		shutdown <- struct{}{}
 	}()
 
 	standbyMessageTimeout := req.IdleTimeout

flow/connectors/postgres/postgres_repl_test.go

@@ -127,7 +127,7 @@ func (suite *PostgresReplicationSnapshotTestSuite) TestSimpleSlotCreation() {
 
 	log.Infof("signaling clone complete for %s after waiting for 2 seconds", flowJobName)
 	time.Sleep(2 * time.Second)
-	signal.CloneComplete <- true
+	signal.CloneComplete <- struct{}{}
 
 	log.Infof("successfully setup replication for %s", flowJobName)
 }

flow/connectors/postgres/qrep_query_executor.go

@@ -83,7 +83,7 @@ func (qe *QRepQueryExecutor) executeQueryInTx(tx pgx.Tx, cursorName string, fetc
 		})
 
 		defer func() {
-			shutdownCh <- true
+			shutdownCh <- struct{}{}
 		}()
 	}
 

flow/connectors/postgres/slot_signal.go

@@ -11,13 +11,13 @@ type SlotCreationResult struct {
 // 2. CloneComplete - which can be waited on to ensure that the clone has completed.
 type SlotSignal struct {
 	SlotCreated   chan *SlotCreationResult
-	CloneComplete chan bool
+	CloneComplete chan struct{}
 }
 
 // NewSlotSignal returns a new SlotSignal.
 func NewSlotSignal() *SlotSignal {
 	return &SlotSignal{
 		SlotCreated:   make(chan *SlotCreationResult, 1),
-		CloneComplete: make(chan bool, 1),
+		CloneComplete: make(chan struct{}, 1),
 	}
 }

flow/connectors/snowflake/qrep_avro_sync.go

@@ -318,7 +318,7 @@ func (s *SnowflakeAvroSyncMethod) putFileToStage(avroFile *avro.AvroFile, stage
 	})
 
 	defer func() {
-		shutdown <- true
+		shutdown <- struct{}{}
 	}()
 
 	if _, err := s.connector.database.Exec(putCmd); err != nil {

flow/connectors/utils/avro/avro_writer.go

@@ -135,7 +135,7 @@ func (p *peerDBOCFWriter) writeRecordsToOCFWriter(ocfWriter *goavro.OCFWriter) (
 		})
 
 		defer func() {
-			shutdown <- true
+			shutdown <- struct{}{}
 		}()
 	}
 

flow/connectors/utils/heartbeat.go

@@ -13,9 +13,9 @@ func HeartbeatRoutine(
 	ctx context.Context,
 	interval time.Duration,
 	message func() string,
-) chan bool {
+) chan struct{} {
 	counter := 1
-	shutdown := make(chan bool)
+	shutdown := make(chan struct{})
 	go func() {
 		for {
 			msg := fmt.Sprintf("heartbeat #%d: %s", counter, message())