flow: specify when channel is read-only or write-only (#987)

Makes it clearer how a channel is intended to be used,
& is an extra layer of static type checking

Also have `HeartbeatRoutine` stop when `ctx.Done()`

flow/connectors/utils/heartbeat.go

@@ -13,19 +13,20 @@ func HeartbeatRoutine(
 	ctx context.Context,
 	interval time.Duration,
 	message func() string,
-) chan struct{} {
-	counter := 1
+) chan<- struct{} {
 	shutdown := make(chan struct{})
 	go func() {
+		counter := 0
 		for {
+			counter += 1
 			msg := fmt.Sprintf("heartbeat #%d: %s", counter, message())
 			RecordHeartbeatWithRecover(ctx, msg)
-			counter += 1
-			to := time.After(interval)
 			select {
 			case <-shutdown:
 				return
-			case <-to:
+			case <-ctx.Done():
+				return
+			case <-time.After(interval):
 			}
 		}
 	}()

flow/model/model.go

@@ -412,7 +412,7 @@ func (r *CDCRecordStream) Close() {
 	r.lastCheckpointSet = true
 }
 
-func (r *CDCRecordStream) GetRecords() chan Record {
+func (r *CDCRecordStream) GetRecords() <-chan Record {
 	return r.records
 }
 

flow/model/qrecord_stream.go

@@ -20,13 +20,13 @@ type QRecordStream struct {
 }
 
 type RecordsToStreamRequest struct {
-	records      chan Record
+	records      <-chan Record
 	TableMapping map[string]uint32
 	BatchID      int64
 }
 
 func NewRecordsToStreamRequest(
-	records chan Record,
+	records <-chan Record,
 	tableMapping map[string]uint32,
 	batchID int64,
 ) *RecordsToStreamRequest {
@@ -37,7 +37,7 @@ func NewRecordsToStreamRequest(
 	}
 }
 
-func (r *RecordsToStreamRequest) GetRecords() chan Record {
+func (r *RecordsToStreamRequest) GetRecords() <-chan Record {
 	return r.records
 }
 
@@ -80,6 +80,6 @@ func (s *QRecordStream) IsSchemaSet() bool {
 	return s.schemaSet
 }
 
-func (s *QRecordStream) SchemaChan() chan QRecordSchemaOrError {
+func (s *QRecordStream) SchemaChan() <-chan QRecordSchemaOrError {
 	return s.schema
 }