tds_driver/client.rs
1mod auth;
2mod cancel;
3mod config;
4mod connection;
5
6mod tls;
7#[cfg(any(
8 feature = "rustls",
9 feature = "native-tls",
10 feature = "vendored-openssl"
11))]
12mod tls_stream;
13
14pub use auth::*;
15pub use cancel::CancelHandle;
16pub use config::*;
17pub(crate) use connection::*;
18
19use crate::tds::codec::RpcValue;
20use crate::tds::stream::ReceivedToken;
21use crate::{
22 result::ExecuteResult,
23 tds::{
24 codec::{self, IteratorJoin},
25 stream::{QueryStream, TokenStream},
26 },
27 BulkLoadRequest, ColumnFlag, MetaDataColumn, SqlReadBytes, ToSql,
28};
29use codec::{
30 BatchRequest, ColumnData, IsolationLevel, PacketHeader, RpcParam, RpcProcId, TokenRpcRequest,
31 TransactionManagerRequest,
32};
33use enumflags2::BitFlags;
34use futures_util::io::{AsyncRead, AsyncWrite};
35use futures_util::stream::TryStreamExt;
36use std::{borrow::Cow, fmt::Debug};
37
38/// `Client` is the main entry point to the SQL Server, providing query
39/// execution capabilities.
40///
41/// A `Client` is created using the [`Config`], defining the needed
42/// connection options and capabilities.
43///
44/// # Example
45///
46/// ```no_run
47/// # use tds_driver::{Config, AuthMethod};
48/// use tokio_util::compat::TokioAsyncWriteCompatExt;
49///
50/// # #[tokio::main]
51/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
52/// let mut config = Config::new();
53///
54/// config.host("0.0.0.0");
55/// config.port(1433);
56/// config.authentication(AuthMethod::sql_server("SA", "<Mys3cureP4ssW0rD>"));
57///
58/// let tcp = tokio::net::TcpStream::connect(config.get_addr()).await?;
59/// tcp.set_nodelay(true)?;
60/// // Client is ready to use.
61/// let client = tds_driver::Client::connect(config, tcp.compat_write()).await?;
62/// # Ok(())
63/// # }
64/// ```
65///
66/// # Cancellation safety
67///
68/// A single [`Client`] drives one connection and one request at a time. If a
69/// `query`/`execute`/`simple_query` future — or the result stream it returns —
70/// is dropped before the request has been sent in full and the response fully
71/// consumed (for example under a `tokio::time::timeout` or a `select!` branch
72/// that loses the race), the connection may be left mid-message and out of sync
73/// with the server. A cancelled *write* is detected and any further use of that
74/// connection fails cleanly; a result stream dropped mid-response cannot be
75/// recovered. In both cases the safe course is to drop the `Client` and open a
76/// new connection (a connection pool should discard the connection on error)
77/// rather than reuse it.
78///
79/// [`Config`]: struct.Config.html
80#[derive(Debug)]
81pub struct Client<S: AsyncRead + AsyncWrite + Unpin + Send> {
82 pub(crate) connection: Connection<S>,
83}
84
85impl<S: AsyncRead + AsyncWrite + Unpin + Send> Client<S> {
86 /// Uses an instance of [`Config`] to specify the connection
87 /// options required to connect to the database using an established
88 /// tcp connection
89 ///
90 /// Note: `tcp_stream` is a connected stream, so some parts of the `Config`
91 /// (such as multi-subnet failover, which selects between resolved
92 /// addresses) must be handled while establishing that stream, outside of
93 /// this constructor.
94 ///
95 /// [`Config`]: struct.Config.html
96 pub async fn connect(config: Config, tcp_stream: S) -> crate::Result<Client<S>> {
97 Ok(Client {
98 connection: Connection::connect(config, tcp_stream).await?,
99 })
100 }
101
102 /// Executes SQL statements in the SQL Server, returning the number rows
103 /// affected. Useful for `INSERT`, `UPDATE` and `DELETE` statements. The
104 /// `query` can define the parameter placement by annotating them with
105 /// `@PN`, where N is the index of the parameter, starting from `1`. If
106 /// executing multiple queries at a time, delimit them with `;` and refer to
107 /// [`ExecuteResult`] how to get results for the separate queries.
108 ///
109 /// For mapping of Rust types when writing, see the documentation for
110 /// [`ToSql`]. For reading data from the database, see the documentation for
111 /// [`FromSql`].
112 ///
113 /// This API is not quite suitable for dynamic query parameters. In these
114 /// cases using a [`Query`] object might be easier.
115 ///
116 /// # Example
117 ///
118 /// ```no_run
119 /// # use tds_driver::Config;
120 /// # use tokio_util::compat::TokioAsyncWriteCompatExt;
121 /// # use std::env;
122 /// # #[tokio::main]
123 /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
124 /// # let c_str = env::var("TDS_DRIVER_TEST_CONNECTION_STRING").unwrap_or(
125 /// # "server=tcp:localhost,1433;integratedSecurity=true;TrustServerCertificate=true".to_owned(),
126 /// # );
127 /// # let config = Config::from_ado_string(&c_str)?;
128 /// # let tcp = tokio::net::TcpStream::connect(config.get_addr()).await?;
129 /// # tcp.set_nodelay(true)?;
130 /// # let mut client = tds_driver::Client::connect(config, tcp.compat_write()).await?;
131 /// let results = client
132 /// .execute(
133 /// "INSERT INTO ##Test (id) VALUES (@P1), (@P2), (@P3)",
134 /// &[&1i32, &2i32, &3i32],
135 /// )
136 /// .await?;
137 /// # Ok(())
138 /// # }
139 /// ```
140 ///
141 /// [`ExecuteResult`]: struct.ExecuteResult.html
142 /// [`ToSql`]: trait.ToSql.html
143 /// [`FromSql`]: trait.FromSql.html
144 /// [`Query`]: struct.Query.html
145 pub async fn execute<'a>(
146 &mut self,
147 query: impl Into<Cow<'a, str>>,
148 params: &[&dyn ToSql],
149 ) -> crate::Result<ExecuteResult> {
150 self.connection.flush_stream().await?;
151 let rpc_params = Self::rpc_params(query);
152
153 let params = params.iter().map(|s| s.to_sql());
154 self.rpc_perform_query(RpcProcId::ExecuteSQL, rpc_params, params)
155 .await?;
156
157 ExecuteResult::new(&mut self.connection).await
158 }
159
160 /// Executes SQL statements in the SQL Server, returning resulting rows.
161 /// Useful for `SELECT` statements. The `query` can define the parameter
162 /// placement by annotating them with `@PN`, where N is the index of the
163 /// parameter, starting from `1`. If executing multiple queries at a time,
164 /// delimit them with `;` and refer to [`QueryStream`] on proper stream
165 /// handling.
166 ///
167 /// For mapping of Rust types when writing, see the documentation for
168 /// [`ToSql`]. For reading data from the database, see the documentation for
169 /// [`FromSql`].
170 ///
171 /// This API can be cumbersome for dynamic query parameters. In these cases,
172 /// if fighting too much with the compiler, using a [`Query`] object might be
173 /// easier.
174 ///
175 /// # Example
176 ///
177 /// ```
178 /// # use tds_driver::Config;
179 /// # use tokio_util::compat::TokioAsyncWriteCompatExt;
180 /// # use std::env;
181 /// # #[tokio::main]
182 /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
183 /// # let c_str = env::var("TDS_DRIVER_TEST_CONNECTION_STRING").unwrap_or(
184 /// # "server=tcp:localhost,1433;integratedSecurity=true;TrustServerCertificate=true".to_owned(),
185 /// # );
186 /// # let config = Config::from_ado_string(&c_str)?;
187 /// # let tcp = tokio::net::TcpStream::connect(config.get_addr()).await?;
188 /// # tcp.set_nodelay(true)?;
189 /// # let mut client = tds_driver::Client::connect(config, tcp.compat_write()).await?;
190 /// let stream = client
191 /// .query(
192 /// "SELECT @P1, @P2, @P3",
193 /// &[&1i32, &2i32, &3i32],
194 /// )
195 /// .await?;
196 /// # Ok(())
197 /// # }
198 /// ```
199 ///
200 /// [`QueryStream`]: struct.QueryStream.html
201 /// [`Query`]: struct.Query.html
202 /// [`ToSql`]: trait.ToSql.html
203 /// [`FromSql`]: trait.FromSql.html
204 pub async fn query<'a, 'b>(
205 &'a mut self,
206 query: impl Into<Cow<'b, str>>,
207 params: &'b [&'b dyn ToSql],
208 ) -> crate::Result<QueryStream<'a>>
209 where
210 'a: 'b,
211 {
212 self.connection.flush_stream().await?;
213 let rpc_params = Self::rpc_params(query);
214
215 let params = params.iter().map(|p| p.to_sql());
216 self.rpc_perform_query(RpcProcId::ExecuteSQL, rpc_params, params)
217 .await?;
218
219 let ts = TokenStream::new(&mut self.connection);
220 let mut result = QueryStream::new(ts.try_unfold());
221 result.forward_to_metadata().await?;
222
223 Ok(result)
224 }
225
226 /// Execute multiple queries, delimited with `;` and return multiple result
227 /// sets; one for each query.
228 ///
229 /// # Example
230 ///
231 /// ```
232 /// # use tds_driver::Config;
233 /// # use tokio_util::compat::TokioAsyncWriteCompatExt;
234 /// # use std::env;
235 /// # #[tokio::main]
236 /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
237 /// # let c_str = env::var("TDS_DRIVER_TEST_CONNECTION_STRING").unwrap_or(
238 /// # "server=tcp:localhost,1433;integratedSecurity=true;TrustServerCertificate=true".to_owned(),
239 /// # );
240 /// # let config = Config::from_ado_string(&c_str)?;
241 /// # let tcp = tokio::net::TcpStream::connect(config.get_addr()).await?;
242 /// # tcp.set_nodelay(true)?;
243 /// # let mut client = tds_driver::Client::connect(config, tcp.compat_write()).await?;
244 /// let row = client.simple_query("SELECT 1 AS col").await?.into_row().await?.unwrap();
245 /// assert_eq!(Some(1i32), row.get("col"));
246 /// # Ok(())
247 /// # }
248 /// ```
249 ///
250 /// # Warning
251 ///
252 /// Do not use this with any user specified input. Please resort to prepared
253 /// statements using the [`query`] method.
254 ///
255 /// [`query`]: #method.query
256 pub async fn simple_query<'a, 'b>(
257 &'a mut self,
258 query: impl Into<Cow<'b, str>>,
259 ) -> crate::Result<QueryStream<'a>>
260 where
261 'a: 'b,
262 {
263 self.connection.flush_stream().await?;
264
265 let req = BatchRequest::new(query, self.connection.context().transaction_descriptor());
266
267 let id = self.connection.context_mut().next_packet_id();
268 self.connection.send(PacketHeader::batch(id), req).await?;
269
270 let ts = TokenStream::new(&mut self.connection);
271
272 let mut result = QueryStream::new(ts.try_unfold());
273 result.forward_to_metadata().await?;
274
275 Ok(result)
276 }
277
278 /// Executes a batch of SQL statements sent verbatim, as [`simple_query`]
279 /// does, and returns the rows each affected rather than any result set.
280 /// The counterpart of [`execute`] for SQL with no parameters.
281 ///
282 /// Because it is a plain batch and not `sp_executesql`, a `SET` option or
283 /// `BEGIN TRAN` in it applies to the session, as it would from a query
284 /// tool.
285 ///
286 /// # Warning
287 ///
288 /// Do not use this with any user specified input. Please resort to prepared
289 /// statements using the [`execute`] method.
290 ///
291 /// [`simple_query`]: #method.simple_query
292 /// [`execute`]: #method.execute
293 pub async fn simple_execute<'a>(
294 &mut self,
295 query: impl Into<Cow<'a, str>>,
296 ) -> crate::Result<ExecuteResult> {
297 self.connection.flush_stream().await?;
298
299 let req = BatchRequest::new(query, self.connection.context().transaction_descriptor());
300
301 let id = self.connection.context_mut().next_packet_id();
302 self.connection.send(PacketHeader::batch(id), req).await?;
303
304 ExecuteResult::new(&mut self.connection).await
305 }
306
307 /// Execute a `BULK INSERT` statement, efficiently storing a large number of
308 /// rows to a specified table. Note: make sure the input row follows the same
309 /// schema as the table, otherwise calling `send()` will return an error.
310 ///
311 /// This is equivalent to calling `bulk_insert("table_name", &["*"])` to merge
312 /// all of a tables columns.
313 ///
314 /// # Example
315 ///
316 /// ```
317 /// # use tds_driver::{Config, IntoRow};
318 /// # use tokio_util::compat::TokioAsyncWriteCompatExt;
319 /// # use std::env;
320 /// # #[tokio::main]
321 /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
322 /// # let c_str = env::var("TDS_DRIVER_TEST_CONNECTION_STRING").unwrap_or(
323 /// # "server=tcp:localhost,1433;integratedSecurity=true;TrustServerCertificate=true".to_owned(),
324 /// # );
325 /// # let config = Config::from_ado_string(&c_str)?;
326 /// # let tcp = tokio::net::TcpStream::connect(config.get_addr()).await?;
327 /// # tcp.set_nodelay(true)?;
328 /// # let mut client = tds_driver::Client::connect(config, tcp.compat_write()).await?;
329 /// let create_table = r#"
330 /// CREATE TABLE ##bulk_test (
331 /// id INT IDENTITY PRIMARY KEY,
332 /// val INT NOT NULL
333 /// )
334 /// "#;
335 ///
336 /// client.simple_query(create_table).await?;
337 ///
338 /// // Start the bulk insert with the client.
339 /// let mut req = client.bulk_insert("##bulk_test").await?;
340 ///
341 /// for i in [0i32, 1i32, 2i32] {
342 /// let row = (i).into_row();
343 ///
344 /// // The request will handle flushing to the wire in an optimal way,
345 /// // balancing between memory usage and IO performance.
346 /// req.send(row).await?;
347 /// }
348 ///
349 /// // The request must be finalized.
350 /// let res = req.finalize().await?;
351 /// assert_eq!(3, res.total());
352 /// # Ok(())
353 /// # }
354 /// ```
355 pub async fn bulk_insert<'a>(
356 &'a mut self,
357 table: &'a str,
358 ) -> crate::Result<BulkLoadRequest<'a, S>> {
359 self.bulk_insert_columns(table, &["*"]).await
360 }
361
362 /// Execute a `BULK INSERT` statement, efficiently storing a large number of
363 /// rows to a specified table. Note: make sure the input row follows the same
364 /// schema as the column list, otherwise calling `send()` will return an error.
365 ///
366 /// # Example
367 ///
368 /// ```
369 /// # use tds_driver::{Config, IntoRow};
370 /// # use tokio_util::compat::TokioAsyncWriteCompatExt;
371 /// # use std::env;
372 /// # #[tokio::main]
373 /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
374 /// # let c_str = env::var("TDS_DRIVER_TEST_CONNECTION_STRING").unwrap_or(
375 /// # "server=tcp:localhost,1433;integratedSecurity=true;TrustServerCertificate=true".to_owned(),
376 /// # );
377 /// # let config = Config::from_ado_string(&c_str)?;
378 /// # let tcp = tokio::net::TcpStream::connect(config.get_addr()).await?;
379 /// # tcp.set_nodelay(true)?;
380 /// # let mut client = tds_driver::Client::connect(config, tcp.compat_write()).await?;
381 /// let create_table = r#"
382 /// CREATE TABLE ##bulk_test_columns (
383 /// id INT IDENTITY PRIMARY KEY,
384 /// foo INT NOT NULL,
385 /// bar FLOAT NOT NULL
386 /// )
387 /// "#;
388 ///
389 /// client.simple_query(create_table).await?;
390 ///
391 /// // Start the bulk insert with the client.
392 /// let mut req = client.bulk_insert_columns("##bulk_test_columns", &["foo", "bar"]).await?;
393 ///
394 /// for (i, j) in [(0i32, 0f64), (1i32, 1f64), (2i32, 2f64)] {
395 /// let row = (i, j).into_row();
396 ///
397 /// // The request will handle flushing to the wire in an optimal way,
398 /// // balancing between memory usage and IO performance.
399 /// req.send(row).await?;
400 /// }
401 ///
402 /// // The request must be finalized.
403 /// let res = req.finalize().await?;
404 /// assert_eq!(3, res.total());
405 /// # Ok(())
406 /// # }
407 /// ```
408 pub async fn bulk_insert_columns<'a>(
409 &'a mut self,
410 table: &'a str,
411 columns: &'a [&'a str],
412 ) -> crate::Result<BulkLoadRequest<'a, S>> {
413 let metadata = self.column_metadata(table, columns).await?;
414 self.bulk_not_cancelled().await?;
415
416 // INSERT BULK has to name the collation of char/varchar/text columns.
417 let mut collations = Vec::new();
418 if metadata.iter().any(|c| c.base.is_non_unicode_text()) {
419 let select = format!("SELECT TOP 0 {} FROM {}", columns.join(", "), table);
420 let names = async {
421 let rows = self
422 .simple_query(codec::collation_names_query(&select))
423 .await?
424 .into_first_result()
425 .await?;
426 Ok::<_, crate::Error>(
427 rows.iter()
428 .map(|r| r.get::<&str, _>(0).map(str::to_owned))
429 .collect(),
430 )
431 };
432 collations = match names.await {
433 Ok(names) => names,
434 // No sys.dm_exec_describe_first_result_set before SQL Server
435 // 2012: the server reads in the database default, as before.
436 Err(crate::Error::Server(_)) => Vec::new(),
437 Err(e) => return Err(e),
438 };
439 self.bulk_not_cancelled().await?;
440 }
441
442 // Only updateable columns are bulk targets (identity/computed columns
443 // are skipped).
444 let mut collations = collations.into_iter();
445 let (columns, collations): (Vec<_>, Vec<_>) = metadata
446 .into_iter()
447 .map(|column| {
448 let collation = collations.next().flatten();
449 let collation = collation.filter(|_| column.base.is_non_unicode_text());
450 (column, collation)
451 })
452 .filter(|(column, _)| column.base.flags.contains(ColumnFlag::Updateable))
453 .unzip();
454
455 // now start bulk upload
456 self.connection.flush_stream().await?;
457 let col_data = columns
458 .iter()
459 .zip(&collations)
460 .map(|(c, collation)| codec::insert_bulk_column(c, collation.as_deref()))
461 .join(", ");
462 let query = format!("INSERT BULK {} ({})", table, col_data);
463
464 let req = BatchRequest::new(query, self.connection.context().transaction_descriptor());
465 let id = self.connection.context_mut().next_packet_id();
466
467 self.connection.send(PacketHeader::batch(id), req).await?;
468
469 let ts = TokenStream::new(&mut self.connection);
470 ts.flush_done().await?;
471 self.bulk_not_cancelled().await?;
472
473 BulkLoadRequest::new(&mut self.connection, columns)
474 }
475
476 /// A cancel that stopped one of the requests a bulk insert makes before
477 /// the rows leaves the server expecting none, so the insert stops there.
478 async fn bulk_not_cancelled(&mut self) -> crate::Result<()> {
479 if !self.connection.is_cancelled() {
480 return Ok(());
481 }
482 self.connection.cancel_request().await?;
483 Err(crate::Error::BulkInput("cancelled".into()))
484 }
485
486 /// Retrieve the column metadata for a set of columns of a table, including
487 /// the column names, types (with their size, precision and scale) and flags
488 /// such as nullability and whether a column is an identity column.
489 ///
490 /// Pass `&["*"]` as `columns` to return the metadata for every column of the
491 /// table.
492 ///
493 /// ```no_run
494 /// # use tds_driver::Config;
495 /// # use tokio_util::compat::TokioAsyncWriteCompatExt;
496 /// # use std::env;
497 /// # #[tokio::main]
498 /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
499 /// # let c_str = env::var("TDS_DRIVER_TEST_CONNECTION_STRING").unwrap_or(
500 /// # "server=tcp:localhost,1433;integratedSecurity=true;TrustServerCertificate=true".to_owned(),
501 /// # );
502 /// # let config = Config::from_ado_string(&c_str)?;
503 /// # let tcp = tokio::net::TcpStream::connect(config.get_addr()).await?;
504 /// # tcp.set_nodelay(true)?;
505 /// # let mut client = tds_driver::Client::connect(config, tcp.compat_write()).await?;
506 /// let meta = client.column_metadata("some_table", &["*"]).await?;
507 /// assert!(meta[0].base().is_identity());
508 /// # Ok(())
509 /// # }
510 /// ```
511 pub async fn column_metadata(
512 &mut self,
513 table: &str,
514 columns: &[&str],
515 ) -> crate::Result<Vec<MetaDataColumn<'static>>> {
516 self.connection.flush_stream().await?;
517
518 // Ask the server for the column layout without returning any rows.
519 let columns = columns.join(", ");
520 let query = format!("SELECT TOP 0 {columns} FROM {table}");
521
522 let req = BatchRequest::new(query, self.connection.context().transaction_descriptor());
523 let id = self.connection.context_mut().next_packet_id();
524 self.connection.send(PacketHeader::batch(id), req).await?;
525
526 let token_stream = TokenStream::new(&mut self.connection).try_unfold();
527
528 let columns = token_stream
529 .try_fold(None, |mut columns, token| async move {
530 if let ReceivedToken::NewResultset(metadata) = token {
531 columns = Some(metadata.columns.clone());
532 };
533
534 Ok(columns)
535 })
536 .await?;
537
538 let columns = columns.ok_or_else(|| {
539 crate::Error::Protocol("expecting column metadata from query but not found".into())
540 })?;
541
542 // Own the column names so the returned metadata is not tied to the
543 // lifetime of the token stream.
544 Ok(columns
545 .into_iter()
546 .map(|c| MetaDataColumn {
547 base: c.base,
548 col_name: std::borrow::Cow::Owned(c.col_name.into_owned()),
549 })
550 .collect())
551 }
552
553 /// Sends a TDS Attention signal to the server (packet type `0x06`,
554 /// MS-TDS section 2.2.1.6) to cancel the request that is currently in
555 /// flight on this connection, and drains the acknowledging token stream so
556 /// the connection can be reused for further queries.
557 ///
558 /// The server responds to the Attention signal by aborting the running
559 /// batch or RPC and returning a `DONE` token with the `DONE_ATTN` status
560 /// bit set. This method waits for that acknowledgement before returning,
561 /// discarding any remaining rows or tokens from the cancelled request.
562 /// If a [`CancelHandle`] has already cancelled the request, it only waits
563 /// for the acknowledgement, and returns at once if that has been read.
564 ///
565 /// # Query cancellation and futures
566 ///
567 /// Dropping a [`query`], [`execute`] or [`simple_query`] future (for
568 /// example when a `tokio::time::timeout` elapses or a `select!` branch is
569 /// cancelled) stops the client from polling the stream, but it does *not*
570 /// tell the server to stop working on the request. To actually cancel the
571 /// in-flight work on the server, keep the [`Client`] and call
572 /// `cancel_query` on it. Because `cancel_query` borrows the client
573 /// mutably, it can only be issued once the borrowing result stream has
574 /// been dropped — typically from a separate task holding the client, or
575 /// after a cancelled/timed-out future has released its borrow. To cancel
576 /// without dropping it, use a [`CancelHandle`].
577 ///
578 /// [`query`]: #method.query
579 /// [`execute`]: #method.execute
580 /// [`simple_query`]: #method.simple_query
581 pub async fn cancel_query(&mut self) -> crate::Result<()> {
582 self.connection.cancel_request().await
583 }
584
585 /// A handle that cancels the request in flight on this connection from
586 /// outside the future awaiting it. See [`CancelHandle`].
587 pub fn cancel_handle(&self) -> CancelHandle {
588 self.connection.cancel_handle()
589 }
590
591 /// Whether the connection is inside a transaction, as far as the server
592 /// has said.
593 ///
594 /// `true` from the ENVCHANGE token that begins one, whether through
595 /// [`begin_transaction`] or a `BEGIN TRAN` in a batch, until the one that
596 /// ends it. The server also sends that one when it rolls a transaction
597 /// back on its own -- a deadlock, an error under `SET XACT_ABORT ON`, a
598 /// `ROLLBACK` in a procedure -- so this is how a client learns that the
599 /// transaction it thinks it is in is gone, and that its next statement
600 /// would run in autocommit.
601 ///
602 /// [`begin_transaction`]: #method.begin_transaction
603 pub fn in_transaction(&self) -> bool {
604 self.connection.context().transaction_descriptor() != [0; 8]
605 }
606
607 /// Closes this database connection explicitly.
608 pub async fn close(self) -> crate::Result<()> {
609 self.connection.close().await
610 }
611
612 /// Begins a new transaction using a Transaction Manager request
613 /// (`TM_BEGIN_XACT`, MS-TDS 2.2.6.9) instead of a `BEGIN TRAN` T-SQL
614 /// batch.
615 ///
616 /// On success the server replies with a `BeginTransaction` environment
617 /// change token whose descriptor is stored in the connection context and
618 /// automatically attached to subsequent requests, scoping them to the
619 /// transaction. Commit the work with [`commit_transaction`] or discard it
620 /// with [`rollback_transaction`].
621 ///
622 /// The transaction is READ COMMITTED, SQL Server's default, whatever
623 /// level an earlier transaction or `SET TRANSACTION ISOLATION LEVEL` left
624 /// on the session; the session stays READ COMMITTED afterwards. Use
625 /// [`begin_transaction_with_isolation`] for another level.
626 ///
627 /// [`commit_transaction`]: #method.commit_transaction
628 /// [`rollback_transaction`]: #method.rollback_transaction
629 /// [`begin_transaction_with_isolation`]: #method.begin_transaction_with_isolation
630 pub async fn begin_transaction(&mut self) -> crate::Result<()> {
631 // Not `Unspecified`, which keeps the session's current level (MS-TDS
632 // 2.2.6.9). SqlClient does the same: `BeginSqlTransaction` turns
633 // `Unspecified` into `ReadCommitted`.
634 self.begin_transaction_with_isolation(IsolationLevel::ReadCommitted)
635 .await
636 }
637
638 /// Begins a new transaction with an explicit isolation level using a
639 /// Transaction Manager request (`TM_BEGIN_XACT`, MS-TDS 2.2.6.9).
640 ///
641 /// The level is a session setting: it stays in effect after the
642 /// transaction ends, for statements outside transactions and for later
643 /// transactions begun with [`IsolationLevel::Unspecified`], which keeps
644 /// the current level. [`begin_transaction`] resets it to READ COMMITTED.
645 ///
646 /// See [`begin_transaction`] for details on transaction scoping.
647 ///
648 /// [`begin_transaction`]: #method.begin_transaction
649 pub async fn begin_transaction_with_isolation(
650 &mut self,
651 isolation_level: IsolationLevel,
652 ) -> crate::Result<()> {
653 let req = TransactionManagerRequest::begin(
654 self.connection.context().transaction_descriptor(),
655 isolation_level,
656 "",
657 );
658
659 self.send_transaction_manager_request(req).await
660 }
661
662 /// Commits the active transaction using a Transaction Manager request
663 /// (`TM_COMMIT_XACT`, MS-TDS 2.2.6.9).
664 ///
665 /// After a successful commit the connection is no longer scoped to a
666 /// transaction.
667 pub async fn commit_transaction(&mut self) -> crate::Result<()> {
668 let req = TransactionManagerRequest::commit(
669 self.connection.context().transaction_descriptor(),
670 "",
671 );
672
673 self.send_transaction_manager_request(req).await
674 }
675
676 /// Rolls back the active transaction using a Transaction Manager request
677 /// (`TM_ROLLBACK_XACT`, MS-TDS 2.2.6.9).
678 ///
679 /// After a successful rollback the connection is no longer scoped to a
680 /// transaction.
681 pub async fn rollback_transaction(&mut self) -> crate::Result<()> {
682 let req = TransactionManagerRequest::rollback(
683 self.connection.context().transaction_descriptor(),
684 "",
685 );
686
687 self.send_transaction_manager_request(req).await
688 }
689
690 /// Creates a named savepoint in the active transaction using a Transaction
691 /// Manager request (`TM_SAVE_XACT`, MS-TDS 2.2.6.9).
692 ///
693 /// The savepoint can later be targeted by a T-SQL `ROLLBACK TRANSACTION
694 /// <name>` to undo work performed after it while keeping the surrounding
695 /// transaction open.
696 pub async fn save_transaction<'a>(
697 &mut self,
698 name: impl Into<Cow<'a, str>>,
699 ) -> crate::Result<()> {
700 let req = TransactionManagerRequest::save(
701 self.connection.context().transaction_descriptor(),
702 name,
703 );
704
705 self.send_transaction_manager_request(req).await
706 }
707
708 async fn send_transaction_manager_request(
709 &mut self,
710 req: TransactionManagerRequest<'_>,
711 ) -> crate::Result<()> {
712 self.connection.flush_stream().await?;
713
714 let id = self.connection.context_mut().next_packet_id();
715 self.connection
716 .send(PacketHeader::transaction_manager(id), req)
717 .await?;
718
719 // The server responds with a DONE token (plus an ENVCHANGE token that
720 // the token stream applies to the connection context, updating the
721 // active transaction descriptor).
722 TokenStream::new(&mut self.connection).flush_done().await?;
723
724 Ok(())
725 }
726
727 pub(crate) fn rpc_params<'a>(query: impl Into<Cow<'a, str>>) -> Vec<RpcParam<'a>> {
728 vec![
729 RpcParam {
730 name: Cow::Borrowed("stmt"),
731 flags: BitFlags::empty(),
732 value: RpcValue::Scalar(ColumnData::String(Some(query.into()))),
733 },
734 RpcParam {
735 name: Cow::Borrowed("params"),
736 flags: BitFlags::empty(),
737 value: RpcValue::Scalar(ColumnData::I32(Some(0))),
738 },
739 ]
740 }
741
742 pub(crate) async fn rpc_perform_query<'a, 'b>(
743 &'a mut self,
744 proc_id: RpcProcId,
745 mut rpc_params: Vec<RpcParam<'b>>,
746 params: impl Iterator<Item = ColumnData<'b>>,
747 ) -> crate::Result<()>
748 where
749 'a: 'b,
750 {
751 let params: Vec<_> = params.collect();
752
753 // The server's error 8003 for this names its 2100-parameter limit,
754 // not the 2098 left to the caller, and comes after the whole send.
755 if params.len() > crate::Query::MAX_PARAMETERS {
756 return Err(crate::Error::Conversion(
757 format!(
758 "{} parameters bound; a query can bind at most {} (Query::MAX_PARAMETERS)",
759 params.len(),
760 crate::Query::MAX_PARAMETERS
761 )
762 .into(),
763 ));
764 }
765
766 let mut param_str = String::new();
767
768 for (i, param) in params.into_iter().enumerate() {
769 if i > 0 {
770 param_str.push(',')
771 }
772 param_str.push_str(&format!("@P{} ", i + 1));
773 param_str.push_str(¶m.type_name());
774
775 rpc_params.push(RpcParam {
776 name: Cow::Owned(format!("@P{}", i + 1)),
777 flags: BitFlags::empty(),
778 value: RpcValue::Scalar(param),
779 });
780 }
781
782 if let Some(params) = rpc_params.iter_mut().find(|x| x.name == "params") {
783 params.value = RpcValue::Scalar(ColumnData::String(Some(param_str.into())));
784 }
785
786 let req = TokenRpcRequest::new(
787 proc_id,
788 rpc_params,
789 self.connection.context().transaction_descriptor(),
790 );
791
792 let id = self.connection.context_mut().next_packet_id();
793 self.connection.send(PacketHeader::rpc(id), req).await?;
794
795 Ok(())
796 }
797
798 /// Sends a named-procedure RPC request with the given parameters. The caller
799 /// is responsible for flushing the connection beforehand and for consuming
800 /// the resulting token stream.
801 pub(crate) async fn rpc_run_command<'a, 'b>(
802 &'a mut self,
803 command_name: Cow<'b, str>,
804 rpc_params: Vec<RpcParam<'b>>,
805 ) -> crate::Result<()>
806 where
807 'a: 'b,
808 {
809 let req = TokenRpcRequest::new(
810 command_name,
811 rpc_params,
812 self.connection.context().transaction_descriptor(),
813 );
814
815 let id = self.connection.context_mut().next_packet_id();
816 self.connection.send(PacketHeader::rpc(id), req).await?;
817
818 Ok(())
819 }
820
821 /// Runs a batch query solely to retrieve its column metadata. Used to
822 /// resolve the column layout of a table-valued parameter type.
823 pub(crate) async fn query_run_for_metadata<'b>(
824 &mut self,
825 query: String,
826 ) -> crate::Result<Option<Vec<MetaDataColumn<'b>>>> {
827 self.connection.flush_stream().await?;
828
829 let req = BatchRequest::new(query, self.connection.context().transaction_descriptor());
830
831 let id = self.connection.context_mut().next_packet_id();
832 self.connection.send(PacketHeader::batch(id), req).await?;
833
834 let token_stream = TokenStream::new(&mut self.connection).try_unfold();
835
836 let columns = token_stream
837 .try_fold(None, |mut columns, token| async move {
838 if let ReceivedToken::NewResultset(metadata) = token {
839 columns = Some(metadata.columns.clone());
840 };
841
842 Ok(columns)
843 })
844 .await?;
845
846 Ok(columns)
847 }
848}