1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
#![recursion_limit = "256"]
//! Multi-result-set streaming inside `retry_tx` (lazy tx on implicit session).
use ydb::{ClientBuilder, Transaction, closure};
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let client = ClientBuilder::new_from_connection_string("grpc://localhost:2136/local")?
.build()
.await?;
let qc = client.query_client();
let sets = qc
// Annotate the parameter type (`tx: &mut Transaction`) so the
// IDE can complete methods on `tx`: rust-analyzer does not yet
// reliably infer `async ||` closure parameter types from the
// `AsyncFnMut` bound. The compiler infers it fine without this.
.retry_tx(closure!(async |tx: &mut Transaction| {
let mut stream = tx.query("SELECT 42 AS a; SELECT 1 AS b, 2 AS c;").await?;
// While `stream` is alive, `tx` stays mutably borrowed — a second
// concurrent query in the same transaction does not compile:
//
// tx.exec("SELECT 1").await?;
// // error[E0499]: cannot borrow `*tx` as mutable more than once
//
// The single-stream-per-transaction invariant comes for free.
let mut set_count = 0;
while let Some(result_set) = stream.next_result_set().await? {
for mut row in result_set {
let _ = row.remove_field_by_name("a");
}
set_count += 1;
}
stream.close().await?;
Ok(set_count)
}))
.await?;
println!("result sets: {sets}");
Ok(())
}