Start on cdc stream adapter

flow/pua/stream_adapter.go

@@ -31,3 +31,26 @@ func AttachToStream(ls *lua.LState, lfn *lua.LFunction, stream *model.QRecordStr
 	}()
 	return output
 }
+
+func AttachToCdcStream(ls *lua.LState, lfn *lua.LFunction, stream *model.CDCStream[model.RecordItems]) *model.CDCStream[model.RecordItems] {
+	// TODO no buffering on output stream
+	outstream := model.NewCDCStream[model.RecordItems]()
+	go func() {
+		// TODO change to latch, fix Close race
+		if stream.WaitAndCheckEmpty() {
+			outstream.SignalAsEmpty()
+		} else {
+			outstream.SignalAsNotEmpty()
+			for record := range stream.GetRecords() {
+				// TODO call lfn
+				// what to do about errors?
+				outstream.AddRecord(record)
+			}
+		}
+		<-stream.GetRecords() // TODO needed because empty signal comes before Close
+		outstream.SchemaDeltas = stream.SchemaDeltas
+		outstream.UpdateLatestCheckpoint(stream.GetLastCheckpoint())
+		outstream.Close()
+	}()
+	return outstream
+}