Struct Client 

pub struct Client<S: ConnectionState> { /* private fields */ }

SQL Server client with type-state connection management.

The generic parameter S represents the current connection state, ensuring at compile time that certain operations are only available in appropriate states.

Implementations

impl Client<Disconnected>

pub async fn connect(config: Config) -> Result<Client<Ready>>

Connect to SQL Server.

This establishes a connection, performs TLS negotiation (if required), and authenticates with the server.

Example

let client = Client::connect(config).await?;

impl<S: ConnectionState> Client<S>

pub fn procedure(&mut self, proc_name: &str) -> Result<ProcedureBuilder<'_, S>>

Start building a stored procedure call with full control over parameters.

Returns a crate::procedure::ProcedureBuilder that allows adding named input and output parameters before executing the call.

The procedure name is validated to prevent SQL injection. It may be schema-qualified (e.g., "dbo.MyProc").

Example

let result = client.procedure("dbo.CalculateSum")?
    .input("@a", &10i32)
    .input("@b", &20i32)
    .output_int("@result")
    .execute().await?;

let sum = result.get_output("@result").unwrap();

pub async fn call_procedure( &mut self, proc_name: &str, params: &[&(dyn ToSql + Sync)], ) -> Result<ProcedureResult>

Execute a stored procedure with positional input parameters.

This is a convenience method for the common case of calling a procedure with input-only parameters. For output parameters or named parameters, use procedure() instead.

Example

let result = client.call_procedure("dbo.GetUser", &[&1i32]).await?;
assert_eq!(result.return_value, 0);

if let Some(rs) = result.first_result_set() {
    println!("columns: {:?}", rs.columns());
}

pub async fn bulk_insert( &mut self, builder: &BulkInsertBuilder, ) -> Result<BulkWriter<'_, S>>

Start a bulk insert operation for the specified table.

Sends the INSERT BULK statement to the server and returns a crate::bulk::BulkWriter for streaming rows. The writer holds a mutable borrow on the client, preventing other operations while the bulk insert is in progress.

Example

use mssql_client::{BulkInsertBuilder, BulkColumn};

let builder = BulkInsertBuilder::new("dbo.Users")
    .with_typed_columns(vec![
        BulkColumn::new("id", "INT", 0)?,
        BulkColumn::new("name", "NVARCHAR(100)", 1)?,
    ]);

let mut writer = client.bulk_insert(&builder).await?;
writer.send_row(&[&1i32, &"Alice"])?;
writer.send_row(&[&2i32, &"Bob"])?;
let result = writer.finish().await?;
println!("Inserted {} rows", result.rows_affected);

pub async fn bulk_insert_without_schema_discovery( &mut self, builder: &BulkInsertBuilder, ) -> Result<BulkWriter<'_, S>>

Start a bulk insert without querying the server for column metadata.

Unlike bulk_insert(), this method does not send SELECT TOP 0 * FROM table to discover column types. Instead, the column metadata is constructed from the BulkColumn types provided on the builder. This saves a round-trip when the schema is known.

Caveats

The caller must ensure BulkColumn entries match the target table’s column definitions exactly. Mismatched types, lengths, precision/scale, or column ordering will cause the server to reject the BulkLoad packet.

For most use cases, prefer bulk_insert() — the extra round-trip is usually negligible and the server-supplied metadata is guaranteed correct.

pub async fn query_named<'a>( &'a mut self, sql: &str, params: &[NamedParam], ) -> Result<QueryStream<'a>>

Execute a query with named parameters and return a streaming result set.

This method accepts NamedParam values, making it compatible with the ToParams trait and the #[derive(ToParams)] macro.

Example

use mssql_client::{NamedParam, ToParams};

// With derive macro:
#[derive(ToParams)]
struct UserQuery { name: String }

let q = UserQuery { name: "Alice".into() };
let rows = client.query_named(
    "SELECT * FROM users WHERE name = @name",
    &q.to_params()?,
).await?;

// Or manually:
let params = vec![NamedParam::from_value("name", &"Alice")?];
let rows = client.query_named(
    "SELECT * FROM users WHERE name = @name",
    &params,
).await?;

pub async fn execute_named( &mut self, sql: &str, params: &[NamedParam], ) -> Result<u64>

Execute a statement with named parameters.

Returns the number of affected rows. This is the named-parameter counterpart of execute(), compatible with the ToParams trait.

Example

use mssql_client::NamedParam;

let params = vec![
    NamedParam::from_value("name", &"Alice")?,
    NamedParam::from_value("email", &"alice@example.com")?,
];
let rows_affected = client.execute_named(
    "INSERT INTO users (name, email) VALUES (@name, @email)",
    &params,
).await?;

impl Client<Ready>

pub fn mark_needs_reset(&mut self)

Mark this connection as needing a reset on next use.

Called by the connection pool when a connection is returned. The next SQL batch or RPC will include the RESETCONNECTION flag in the TDS packet header, causing SQL Server to reset connection state (temp tables, SET options, transaction isolation level, etc.) before executing the command.

This is more efficient than calling sp_reset_connection as a separate command because it’s handled at the TDS protocol level.

pub fn needs_reset(&self) -> bool

Check if this connection needs a reset.

Returns true if mark_needs_reset() was called and the reset hasn’t been performed yet.

pub async fn query<'a>( &'a mut self, sql: &str, params: &[&(dyn ToSql + Sync)], ) -> Result<QueryStream<'a>>

Execute a query and return a streaming result set.

Per ADR-007, results are streamed by default for memory efficiency. Use .collect_all() on the stream if you need all rows in memory.

Example

use futures::StreamExt;

// Streaming (memory-efficient)
let mut stream = client.query("SELECT * FROM users WHERE id = @p1", &[&1]).await?;
while let Some(row) = stream.next().await {
    let row = row?;
    process(&row);
}

// Buffered (loads all into memory)
let rows: Vec<Row> = client
    .query("SELECT * FROM small_table", &[])
    .await?
    .collect_all()
    .await?;

pub async fn query_with_timeout<'a>( &'a mut self, sql: &str, params: &[&(dyn ToSql + Sync)], timeout_duration: Duration, ) -> Result<QueryStream<'a>>

Execute a query with a specific timeout.

This overrides the default command_timeout from the connection configuration for this specific query. If the query does not complete within the specified duration, an error is returned.

Arguments

• sql - The SQL query to execute
• params - Query parameters
• timeout_duration - Maximum time to wait for the query to complete

Example

use std::time::Duration;

// Execute with a 5-second timeout
let rows = client
    .query_with_timeout(
        "SELECT * FROM large_table",
        &[],
        Duration::from_secs(5),
    )
    .await?;

pub async fn query_multiple<'a>( &'a mut self, sql: &str, params: &[&(dyn ToSql + Sync)], ) -> Result<MultiResultStream<'a>>

Execute a batch that may return multiple result sets.

This is useful for stored procedures or SQL batches that contain multiple SELECT statements.

Example

// Execute a batch with multiple SELECTs
let mut results = client.query_multiple(
    "SELECT 1 AS a; SELECT 2 AS b, 3 AS c;",
    &[]
).await?;

// Process first result set
while let Some(row) = results.next_row().await? {
    println!("Result 1: {:?}", row);
}

// Move to second result set
if results.next_result().await? {
    while let Some(row) = results.next_row().await? {
        println!("Result 2: {:?}", row);
    }
}

pub async fn execute( &mut self, sql: &str, params: &[&(dyn ToSql + Sync)], ) -> Result<u64>

Execute a query that doesn’t return rows.

Returns the number of affected rows.

pub async fn execute_with_timeout( &mut self, sql: &str, params: &[&(dyn ToSql + Sync)], timeout_duration: Duration, ) -> Result<u64>

Execute a statement with a specific timeout.

This overrides the default command_timeout from the connection configuration for this specific statement. If the statement does not complete within the specified duration, an error is returned.

Arguments

• sql - The SQL statement to execute
• params - Statement parameters
• timeout_duration - Maximum time to wait for the statement to complete

Example

use std::time::Duration;

// Execute with a 10-second timeout
let rows_affected = client
    .execute_with_timeout(
        "UPDATE large_table SET status = @p1",
        &[&"processed"],
        Duration::from_secs(10),
    )
    .await?;

pub async fn begin_transaction(self) -> Result<Client<InTransaction>>

Begin a transaction.

This transitions the client from Ready to InTransaction state. Per MS-TDS spec, the server returns a transaction descriptor in the BeginTransaction EnvChange token that must be included in subsequent ALL_HEADERS sections.

pub async fn begin_transaction_with_isolation( self, isolation_level: IsolationLevel, ) -> Result<Client<InTransaction>>

Begin a transaction with a specific isolation level.

This transitions the client from Ready to InTransaction state with the specified isolation level.

Example

use mssql_client::IsolationLevel;

let tx = client.begin_transaction_with_isolation(IsolationLevel::Serializable).await?;
// All operations in this transaction use SERIALIZABLE isolation
tx.commit().await?;

pub async fn simple_query(&mut self, sql: &str) -> Result<()>

Execute a simple query without parameters.

This is useful for DDL statements and simple queries where you don’t need to retrieve the affected row count.

pub async fn close(self) -> Result<()>

Close the connection gracefully.

pub fn database(&self) -> Option<&str>

Get the current database name.

pub fn host(&self) -> &str

Get the server host.

pub fn port(&self) -> u16

Get the server port.

pub fn is_in_transaction(&self) -> bool

Check if the connection is currently in a transaction.

This returns true if a transaction was started via raw SQL (BEGIN TRANSACTION) and has not yet been committed or rolled back.

Note: This only tracks transactions started via raw SQL. Transactions started via the type-state API (begin_transaction()) result in a Client<InTransaction> which is a different type.

Example

client.execute("BEGIN TRANSACTION", &[]).await?;
assert!(client.is_in_transaction());

client.execute("COMMIT", &[]).await?;
assert!(!client.is_in_transaction());

pub fn is_in_flight(&self) -> bool

Check if a request is in-flight (sent but response not fully read).

Used by the connection pool to detect dirty connections that were interrupted mid-query (e.g., by tokio::select! or a timeout). A connection with an in-flight request has unread data in the TCP buffer and must be discarded rather than returned to the pool.

pub fn cancel_handle(&self) -> CancelHandle

Get a handle for cancelling the current query.

The cancel handle can be cloned and sent to other tasks, enabling cancellation of long-running queries from a separate async context.

Example

use std::time::Duration;

let cancel_handle = client.cancel_handle();

// Spawn a task to cancel after 10 seconds
let handle = tokio::spawn(async move {
    tokio::time::sleep(Duration::from_secs(10)).await;
    let _ = cancel_handle.cancel().await;
});

// This query will be cancelled if it runs longer than 10 seconds
let result = client.query("SELECT * FROM very_large_table", &[]).await;

impl Client<InTransaction>

Drop Behavior

Client<InTransaction> has no automatic rollback on drop. If the client is dropped without calling commit() or rollback(), the transaction remains open on the server until the TCP connection closes (at which point SQL Server automatically rolls back).

This is because Drop is synchronous and cannot perform the async I/O needed to send a ROLLBACK TRANSACTION command.

Consequences of dropping without commit/rollback

• Direct connections: The transaction leaks until the OS TCP timeout (potentially 30+ minutes), holding locks on any modified rows.
• Pooled connections: The pool detects the active transaction descriptor and discards the connection rather than returning it to the idle pool (see PooledConnection::drop in mssql-driver-pool).

Best practice

Always ensure commit() or rollback() is called. Use helper patterns for error paths:

let tx = client.begin_transaction().await?;
match do_work(&tx).await {
    Ok(_) => { tx.commit().await?; }
    Err(e) => { tx.rollback().await?; return Err(e); }
}

pub async fn query<'a>( &'a mut self, sql: &str, params: &[&(dyn ToSql + Sync)], ) -> Result<QueryStream<'a>>

Execute a query within the transaction and return a streaming result set.

See Client<Ready>::query for usage examples.

pub async fn execute( &mut self, sql: &str, params: &[&(dyn ToSql + Sync)], ) -> Result<u64>

Execute a statement within the transaction.

Returns the number of affected rows.

pub async fn query_with_timeout<'a>( &'a mut self, sql: &str, params: &[&(dyn ToSql + Sync)], timeout_duration: Duration, ) -> Result<QueryStream<'a>>

Execute a query within the transaction with a specific timeout.

See Client<Ready>::query_with_timeout for details.

pub async fn execute_with_timeout( &mut self, sql: &str, params: &[&(dyn ToSql + Sync)], timeout_duration: Duration, ) -> Result<u64>

Execute a statement within the transaction with a specific timeout.

See Client<Ready>::execute_with_timeout for details.

pub async fn commit(self) -> Result<Client<Ready>>

Commit the transaction.

This transitions the client back to Ready state.

pub async fn rollback(self) -> Result<Client<Ready>>

Rollback the transaction.

This transitions the client back to Ready state.

pub async fn save_point(&mut self, name: &str) -> Result<SavePoint>

Create a savepoint and return a handle for later rollback.

The returned SavePoint handle contains the validated savepoint name. Use it with rollback_to() to partially undo transaction work.

Example

let tx = client.begin_transaction().await?;
tx.execute("INSERT INTO orders ...").await?;
let sp = tx.save_point("before_items").await?;
tx.execute("INSERT INTO items ...").await?;
// Oops, rollback just the items
tx.rollback_to(&sp).await?;
tx.commit().await?;

pub async fn rollback_to(&mut self, savepoint: &SavePoint) -> Result<()>

Rollback to a savepoint.

This rolls back all changes made after the savepoint was created, but keeps the transaction active. The savepoint remains valid and can be rolled back to again.

Example

let sp = tx.save_point("checkpoint").await?;
// ... do some work ...
tx.rollback_to(&sp).await?;  // Undo changes since checkpoint
// Transaction is still active, savepoint is still valid

pub async fn release_savepoint(&mut self, savepoint: SavePoint) -> Result<()>

Release a savepoint (optional cleanup).

Note: SQL Server doesn’t have explicit savepoint release, but this method is provided for API completeness. The savepoint is automatically released when the transaction commits or rolls back.

pub fn cancel_handle(&self) -> CancelHandle

Get a handle for cancelling the current query within the transaction.

See Client<Ready>::cancel_handle for usage examples.

Trait Implementations

impl<S: ConnectionState> Debug for Client<S>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.