google_cloud_bigquery/write/
append_future.rs

1// Copyright 2026 Google LLC
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     https://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use super::append_response::AppendResponse;
16use super::error::{AppendError, AppendResult};
17use std::future::Future;
18use std::pin::Pin;
19use std::task::{Context, Poll};
20use tokio::sync::oneshot;
21
22/// A future that resolves to the result of an async append operation.
23///
24/// This future represents a write request that has already been queued by the
25/// client library to send over the network. Awaiting this future yields the server's acknowledgment
26/// or an error if the write fails.
27#[derive(Debug)]
28pub struct AppendFuture {
29    rx: oneshot::Receiver<AppendResult<AppendResponse>>,
30}
31
32impl AppendFuture {
33    #[allow(dead_code)]
34    pub(crate) fn new(rx: oneshot::Receiver<AppendResult<AppendResponse>>) -> Self {
35        Self { rx }
36    }
37}
38
39impl Future for AppendFuture {
40    type Output = AppendResult<AppendResponse>;
41
42    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
43        let result = std::task::ready!(Pin::new(&mut self.rx).poll(cx));
44        match result {
45            Ok(res) => Poll::Ready(res),
46            Err(_) => Poll::Ready(Err(AppendError::UnexpectedEndOfStream)),
47        }
48    }
49}
50
51#[cfg(test)]
52mod tests {
53    use super::*;
54    use crate::model::TableSchema;
55
56    #[tokio::test]
57    async fn happy_path() {
58        let (tx, rx) = oneshot::channel();
59        let _ = tx.send(Ok(AppendResponse {
60            offset: None,
61            updated_schema: Some(TableSchema::default()),
62        }));
63        let future = AppendFuture::new(rx);
64        let resp = future.await.expect("should succeed");
65        assert_eq!(resp.offset, None);
66        assert_eq!(resp.updated_schema, Some(TableSchema::default()));
67    }
68
69    #[tokio::test]
70    async fn dropped_sender() {
71        let (tx, rx) = oneshot::channel::<AppendResult<AppendResponse>>();
72        // Drop the sender immediately
73        drop(tx);
74
75        let future = AppendFuture::new(rx);
76        let err = future
77            .await
78            .expect_err("should return unexpected end of stream");
79        assert!(matches!(err, AppendError::UnexpectedEndOfStream));
80    }
81
82    #[tokio::test]
83    async fn channel_returns_error() {
84        let (tx, rx) = oneshot::channel();
85        let _ = tx.send(Err(AppendError::UnexpectedEndOfStream));
86        let future = AppendFuture::new(rx);
87        let err = future.await.expect_err("should return error from task");
88        assert!(matches!(err, AppendError::UnexpectedEndOfStream));
89    }
90}
google_cloud_bigquery/write/append_future.rs

google_cloud_bigquery/write/
append_future.rs
