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