refactor: Extract a function for sending points
@@ -321,7 +321,19 @@ async fn send_series_filters(
tx.send(series_frame_response_header).await.unwrap();
}
- match batch.values {
+ if let Err(e) = send_points(tx.clone(), batch.values).await {
+ tx.send(Err(e)).await.unwrap();
+ }
+ }
+
+ Ok(())
+}
+
+async fn send_points(
+ mut tx: mpsc::Sender<Result<ReadResponse, Status>>,
+ read_values: ReadValues,
+) -> Result<(), Status> {
+ match read_values {
ReadValues::F64(values) => {
let (timestamps, values) = values.into_iter().map(|p| (p.time, p.value)).unzip();
let data_frame_response = Ok(ReadResponse {
@@ -346,7 +358,6 @@ async fn send_series_filters(
tx.send(data_frame_response).await.unwrap();
}
}
- }
Ok(())
}
