tokio_postgres/client.rs
1use crate::codec::{BackendMessages, FrontendMessage};
2use crate::config::{SslMode, SslNegotiation};
3use crate::connection::{Request, RequestMessages};
4use crate::copy_out::CopyOutStream;
5#[cfg(feature = "runtime")]
6use crate::keepalive::KeepaliveConfig;
7use crate::query::RowStream;
8use crate::simple_query::SimpleQueryStream;
9#[cfg(feature = "runtime")]
10use crate::tls::MakeTlsConnect;
11use crate::tls::TlsConnect;
12use crate::types::{Oid, ToSql, Type};
13#[cfg(feature = "runtime")]
14use crate::Socket;
15use crate::{
16 copy_in, copy_out, prepare, query, simple_query, slice_iter, CancelToken, CopyInSink, Error,
17 Row, SimpleQueryMessage, Statement, ToStatement, Transaction, TransactionBuilder,
18};
19use bytes::{Buf, BytesMut};
20use fallible_iterator::FallibleIterator;
21use futures_channel::mpsc;
22use futures_util::{StreamExt, TryStreamExt};
23use parking_lot::Mutex;
24use postgres_protocol::message::backend::Message;
25use postgres_protocol::message::frontend;
26use postgres_types::{BorrowToSql, FromSqlOwned};
27use std::collections::HashMap;
28use std::fmt;
29use std::future;
30#[cfg(feature = "runtime")]
31use std::net::IpAddr;
32#[cfg(feature = "runtime")]
33use std::path::PathBuf;
34use std::pin::pin;
35use std::sync::Arc;
36use std::task::{ready, Context, Poll};
37#[cfg(feature = "runtime")]
38use std::time::Duration;
39use tokio::io::{AsyncRead, AsyncWrite};
40
41pub struct Responses {
42 receiver: mpsc::Receiver<BackendMessages>,
43 cur: BackendMessages,
44}
45
46impl Responses {
47 pub fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Message, Error>> {
48 loop {
49 match self.cur.next().map_err(Error::parse)? {
50 Some(Message::ErrorResponse(body)) => return Poll::Ready(Err(Error::db(body))),
51 Some(message) => return Poll::Ready(Ok(message)),
52 None => {}
53 }
54
55 match ready!(self.receiver.poll_next_unpin(cx)) {
56 Some(messages) => self.cur = messages,
57 None => return Poll::Ready(Err(Error::closed())),
58 }
59 }
60 }
61
62 pub async fn next(&mut self) -> Result<Message, Error> {
63 future::poll_fn(|cx| self.poll_next(cx)).await
64 }
65}
66
67/// A cache of type info and prepared statements for fetching type info
68/// (corresponding to the queries in the [prepare](prepare) module).
69#[derive(Default)]
70struct CachedTypeInfo {
71 /// A statement for basic information for a type from its
72 /// OID. Corresponds to [TYPEINFO_QUERY](prepare::TYPEINFO_QUERY) (or its
73 /// fallback).
74 typeinfo: Option<Statement>,
75 /// A statement for getting information for a composite type from its OID.
76 /// Corresponds to [TYPEINFO_QUERY](prepare::TYPEINFO_COMPOSITE_QUERY).
77 typeinfo_composite: Option<Statement>,
78 /// A statement for getting information for a composite type from its OID.
79 /// Corresponds to [TYPEINFO_QUERY](prepare::TYPEINFO_COMPOSITE_QUERY) (or
80 /// its fallback).
81 typeinfo_enum: Option<Statement>,
82
83 /// Cache of types already looked up.
84 types: HashMap<Oid, Type>,
85}
86
87pub struct InnerClient {
88 sender: mpsc::UnboundedSender<Request>,
89 cached_typeinfo: Mutex<CachedTypeInfo>,
90
91 /// A buffer to use when writing out postgres commands.
92 buffer: Mutex<BytesMut>,
93}
94
95impl InnerClient {
96 pub fn send(&self, messages: RequestMessages) -> Result<Responses, Error> {
97 let (sender, receiver) = mpsc::channel(1);
98 let request = Request { messages, sender };
99 self.sender
100 .unbounded_send(request)
101 .map_err(|_| Error::closed())?;
102
103 Ok(Responses {
104 receiver,
105 cur: BackendMessages::empty(),
106 })
107 }
108
109 pub fn typeinfo(&self) -> Option<Statement> {
110 self.cached_typeinfo.lock().typeinfo.clone()
111 }
112
113 pub fn set_typeinfo(&self, statement: &Statement) {
114 self.cached_typeinfo.lock().typeinfo = Some(statement.clone());
115 }
116
117 pub fn typeinfo_composite(&self) -> Option<Statement> {
118 self.cached_typeinfo.lock().typeinfo_composite.clone()
119 }
120
121 pub fn set_typeinfo_composite(&self, statement: &Statement) {
122 self.cached_typeinfo.lock().typeinfo_composite = Some(statement.clone());
123 }
124
125 pub fn typeinfo_enum(&self) -> Option<Statement> {
126 self.cached_typeinfo.lock().typeinfo_enum.clone()
127 }
128
129 pub fn set_typeinfo_enum(&self, statement: &Statement) {
130 self.cached_typeinfo.lock().typeinfo_enum = Some(statement.clone());
131 }
132
133 pub fn type_(&self, oid: Oid) -> Option<Type> {
134 self.cached_typeinfo.lock().types.get(&oid).cloned()
135 }
136
137 pub fn set_type(&self, oid: Oid, type_: &Type) {
138 self.cached_typeinfo.lock().types.insert(oid, type_.clone());
139 }
140
141 pub fn clear_type_cache(&self) {
142 self.cached_typeinfo.lock().types.clear();
143 }
144
145 /// Call the given function with a buffer to be used when writing out
146 /// postgres commands.
147 pub fn with_buf<F, R>(&self, f: F) -> R
148 where
149 F: FnOnce(&mut BytesMut) -> R,
150 {
151 let mut buffer = self.buffer.lock();
152 let r = f(&mut buffer);
153 buffer.clear();
154 r
155 }
156}
157
158#[cfg(feature = "runtime")]
159#[derive(Clone)]
160pub(crate) struct SocketConfig {
161 pub addr: Addr,
162 pub hostname: Option<String>,
163 pub port: u16,
164 pub connect_timeout: Option<Duration>,
165 pub tcp_user_timeout: Option<Duration>,
166 pub keepalive: Option<KeepaliveConfig>,
167}
168
169#[cfg(feature = "runtime")]
170#[derive(Clone)]
171pub(crate) enum Addr {
172 Tcp(IpAddr),
173 #[cfg(unix)]
174 Unix(PathBuf),
175}
176
177/// An asynchronous PostgreSQL client.
178///
179/// The client is one half of what is returned when a connection is established. Users interact with the database
180/// through this client object.
181pub struct Client {
182 inner: Arc<InnerClient>,
183 #[cfg(feature = "runtime")]
184 socket_config: Option<SocketConfig>,
185 ssl_mode: SslMode,
186 ssl_negotiation: SslNegotiation,
187 process_id: i32,
188 secret_key: i32,
189}
190
191impl Client {
192 pub(crate) fn new(
193 sender: mpsc::UnboundedSender<Request>,
194 ssl_mode: SslMode,
195 ssl_negotiation: SslNegotiation,
196 process_id: i32,
197 secret_key: i32,
198 ) -> Client {
199 Client {
200 inner: Arc::new(InnerClient {
201 sender,
202 cached_typeinfo: Default::default(),
203 buffer: Default::default(),
204 }),
205 #[cfg(feature = "runtime")]
206 socket_config: None,
207 ssl_mode,
208 ssl_negotiation,
209 process_id,
210 secret_key,
211 }
212 }
213
214 pub(crate) fn inner(&self) -> &Arc<InnerClient> {
215 &self.inner
216 }
217
218 #[cfg(feature = "runtime")]
219 pub(crate) fn set_socket_config(&mut self, socket_config: SocketConfig) {
220 self.socket_config = Some(socket_config);
221 }
222
223 /// Creates a new prepared statement.
224 ///
225 /// Prepared statements can be executed repeatedly, and may contain query parameters (indicated by `$1`, `$2`, etc),
226 /// which are set when executed. Prepared statements can only be used with the connection that created them.
227 pub async fn prepare(&self, query: &str) -> Result<Statement, Error> {
228 self.prepare_typed(query, &[]).await
229 }
230
231 /// Like `prepare`, but allows the types of query parameters to be explicitly specified.
232 ///
233 /// The list of types may be smaller than the number of parameters - the types of the remaining parameters will be
234 /// inferred. For example, `client.prepare_typed(query, &[])` is equivalent to `client.prepare(query)`.
235 pub async fn prepare_typed(
236 &self,
237 query: &str,
238 parameter_types: &[Type],
239 ) -> Result<Statement, Error> {
240 prepare::prepare(&self.inner, query, parameter_types).await
241 }
242
243 /// Executes a statement, returning a vector of the resulting rows.
244 ///
245 /// A statement may contain parameters, specified by `$n`, where `n` is the index of the parameter of the list
246 /// provided, 1-indexed.
247 ///
248 /// The `statement` argument can either be a `Statement`, or a raw query string. If the same statement will be
249 /// repeatedly executed (perhaps with different query parameters), consider preparing the statement up front
250 /// with the `prepare` method.
251 pub async fn query<T>(
252 &self,
253 statement: &T,
254 params: &[&(dyn ToSql + Sync)],
255 ) -> Result<Vec<Row>, Error>
256 where
257 T: ?Sized + ToStatement,
258 {
259 self.query_raw(statement, slice_iter(params))
260 .await?
261 .try_collect()
262 .await
263 }
264
265 /// Returns a vector of scalars.
266 pub async fn query_scalar<R: FromSqlOwned, T>(
267 &self,
268 statement: &T,
269 params: &[&(dyn ToSql + Sync)],
270 ) -> Result<Vec<R>, Error>
271 where
272 T: ?Sized + ToStatement + fmt::Debug,
273 {
274 let rows: Vec<Row> = self
275 .query_raw(statement, slice_iter(params))
276 .await?
277 .try_collect()
278 .await?;
279
280 if let Some(row) = rows.first() {
281 if row.len() != 1 {
282 return Err(Error::column_count());
283 }
284 };
285
286 rows.into_iter().map(|r| r.try_get(0)).collect()
287 }
288
289 /// Executes a statement which returns a single row, returning it.
290 ///
291 /// Returns an error if the query does not return exactly one row.
292 ///
293 /// A statement may contain parameters, specified by `$n`, where `n` is the index of the parameter of the list
294 /// provided, 1-indexed.
295 ///
296 /// The `statement` argument can either be a `Statement`, or a raw query string. If the same statement will be
297 /// repeatedly executed (perhaps with different query parameters), consider preparing the statement up front
298 /// with the `prepare` method.
299 pub async fn query_one<T>(
300 &self,
301 statement: &T,
302 params: &[&(dyn ToSql + Sync)],
303 ) -> Result<Row, Error>
304 where
305 T: ?Sized + ToStatement,
306 {
307 self.query_opt(statement, params)
308 .await
309 .and_then(|res| res.ok_or_else(Error::row_count))
310 }
311
312 /// Like [`Client::query_one`] but returns one scalar.
313 pub async fn query_one_scalar<R: FromSqlOwned, T>(
314 &self,
315 statement: &T,
316 params: &[&(dyn ToSql + Sync)],
317 ) -> Result<R, Error>
318 where
319 T: ?Sized + ToStatement + fmt::Debug,
320 {
321 let row = self.query_one(statement, params).await?;
322
323 if row.len() != 1 {
324 return Err(Error::column_count());
325 }
326
327 row.try_get(0)
328 }
329
330 /// Executes a statements which returns zero or one rows, returning it.
331 ///
332 /// Returns an error if the query returns more than one row.
333 ///
334 /// A statement may contain parameters, specified by `$n`, where `n` is the index of the parameter of the list
335 /// provided, 1-indexed.
336 ///
337 /// The `statement` argument can either be a `Statement`, or a raw query string. If the same statement will be
338 /// repeatedly executed (perhaps with different query parameters), consider preparing the statement up front
339 /// with the `prepare` method.
340 pub async fn query_opt<T>(
341 &self,
342 statement: &T,
343 params: &[&(dyn ToSql + Sync)],
344 ) -> Result<Option<Row>, Error>
345 where
346 T: ?Sized + ToStatement,
347 {
348 let mut stream = pin!(self.query_raw(statement, slice_iter(params)).await?);
349
350 let mut first = None;
351
352 // Originally this was two calls to `try_next().await?`,
353 // once for the first element, and second to error if more than one.
354 //
355 // However, this new form with only one .await in a loop generates
356 // slightly smaller codegen/stack usage for the resulting future.
357 while let Some(row) = stream.try_next().await? {
358 if first.is_some() {
359 return Err(Error::row_count());
360 }
361
362 first = Some(row);
363 }
364
365 Ok(first)
366 }
367
368 /// Like [`Client::query_opt`] but returns an optional scalar.
369 pub async fn query_opt_scalar<R: FromSqlOwned, T>(
370 &self,
371 statement: &T,
372 params: &[&(dyn ToSql + Sync)],
373 ) -> Result<Option<R>, Error>
374 where
375 T: ?Sized + ToStatement + fmt::Debug,
376 {
377 let row = self.query_opt(statement, params).await?;
378
379 if let Some(row) = &row {
380 if row.len() != 1 {
381 return Err(Error::column_count());
382 }
383 }
384
385 row.map(|x| x.try_get::<_, R>(0)).transpose()
386 }
387
388 /// The maximally flexible version of [`query`].
389 ///
390 /// A statement may contain parameters, specified by `$n`, where `n` is the index of the parameter of the list
391 /// provided, 1-indexed.
392 ///
393 /// The `statement` argument can either be a `Statement`, or a raw query string. If the same statement will be
394 /// repeatedly executed (perhaps with different query parameters), consider preparing the statement up front
395 /// with the `prepare` method.
396 ///
397 /// [`query`]: #method.query
398 ///
399 /// # Examples
400 ///
401 /// ```no_run
402 /// # async fn async_main(client: &tokio_postgres::Client) -> Result<(), tokio_postgres::Error> {
403 /// use std::pin::pin;
404 /// use futures_util::TryStreamExt;
405 ///
406 /// let params: Vec<String> = vec![
407 /// "first param".into(),
408 /// "second param".into(),
409 /// ];
410 /// let mut it = pin!(client.query_raw(
411 /// "SELECT foo FROM bar WHERE biz = $1 AND baz = $2",
412 /// params,
413 /// ).await?);
414 ///
415 /// while let Some(row) = it.try_next().await? {
416 /// let foo: i32 = row.get("foo");
417 /// println!("foo: {}", foo);
418 /// }
419 /// # Ok(())
420 /// # }
421 /// ```
422 pub async fn query_raw<T, P, I>(&self, statement: &T, params: I) -> Result<RowStream, Error>
423 where
424 T: ?Sized + ToStatement,
425 P: BorrowToSql,
426 I: IntoIterator<Item = P>,
427 I::IntoIter: ExactSizeIterator,
428 {
429 let statement = statement.__convert().into_statement(&self.inner).await?;
430 query::query(&self.inner, statement, params).await
431 }
432
433 /// Like `query`, but requires the types of query parameters to be explicitly specified.
434 ///
435 /// Compared to `query`, this method allows performing queries without three round trips (for
436 /// prepare, execute, and close) by requiring the caller to specify parameter values along with
437 /// their Postgres type. Thus, this is suitable in environments where prepared statements aren't
438 /// supported (such as Cloudflare Workers with Hyperdrive).
439 ///
440 /// A statement may contain parameters, specified by `$n`, where `n` is the index of the
441 /// parameter of the list provided, 1-indexed.
442 pub async fn query_typed(
443 &self,
444 query: &str,
445 params: &[(&(dyn ToSql + Sync), Type)],
446 ) -> Result<Vec<Row>, Error> {
447 self.query_typed_raw(query, params.iter().map(|(v, t)| (*v, t.clone())))
448 .await?
449 .try_collect()
450 .await
451 }
452
453 /// The maximally flexible version of [`query_typed`].
454 ///
455 /// Compared to `query`, this method allows performing queries without three round trips (for
456 /// prepare, execute, and close) by requiring the caller to specify parameter values along with
457 /// their Postgres type. Thus, this is suitable in environments where prepared statements aren't
458 /// supported (such as Cloudflare Workers with Hyperdrive).
459 ///
460 /// A statement may contain parameters, specified by `$n`, where `n` is the index of the
461 /// parameter of the list provided, 1-indexed.
462 ///
463 /// [`query_typed`]: #method.query_typed
464 ///
465 /// # Examples
466 ///
467 /// ```no_run
468 /// # async fn async_main(client: &tokio_postgres::Client) -> Result<(), tokio_postgres::Error> {
469 /// use std::pin::pin;
470 /// use futures_util::{TryStreamExt};
471 /// use tokio_postgres::types::Type;
472 ///
473 /// let params: Vec<(String, Type)> = vec![
474 /// ("first param".into(), Type::TEXT),
475 /// ("second param".into(), Type::TEXT),
476 /// ];
477 /// let mut it = pin!(client.query_typed_raw(
478 /// "SELECT foo FROM bar WHERE biz = $1 AND baz = $2",
479 /// params,
480 /// ).await?);
481 ///
482 /// while let Some(row) = it.try_next().await? {
483 /// let foo: i32 = row.get("foo");
484 /// println!("foo: {}", foo);
485 /// }
486 /// # Ok(())
487 /// # }
488 /// ```
489 pub async fn query_typed_raw<P, I>(&self, query: &str, params: I) -> Result<RowStream, Error>
490 where
491 P: BorrowToSql,
492 I: IntoIterator<Item = (P, Type)>,
493 {
494 query::query_typed(&self.inner, query, params).await
495 }
496
497 /// Executes a statement, returning the number of rows modified.
498 ///
499 /// A statement may contain parameters, specified by `$n`, where `n` is the index of the parameter of the list
500 /// provided, 1-indexed.
501 ///
502 /// The `statement` argument can either be a `Statement`, or a raw query string. If the same statement will be
503 /// repeatedly executed (perhaps with different query parameters), consider preparing the statement up front
504 /// with the `prepare` method.
505 ///
506 /// If the statement does not modify any rows (e.g. `SELECT`), 0 is returned.
507 pub async fn execute<T>(
508 &self,
509 statement: &T,
510 params: &[&(dyn ToSql + Sync)],
511 ) -> Result<u64, Error>
512 where
513 T: ?Sized + ToStatement,
514 {
515 self.execute_raw(statement, slice_iter(params)).await
516 }
517
518 /// The maximally flexible version of [`execute`].
519 ///
520 /// A statement may contain parameters, specified by `$n`, where `n` is the index of the parameter of the list
521 /// provided, 1-indexed.
522 ///
523 /// The `statement` argument can either be a `Statement`, or a raw query string. If the same statement will be
524 /// repeatedly executed (perhaps with different query parameters), consider preparing the statement up front
525 /// with the `prepare` method.
526 ///
527 /// [`execute`]: #method.execute
528 pub async fn execute_raw<T, P, I>(&self, statement: &T, params: I) -> Result<u64, Error>
529 where
530 T: ?Sized + ToStatement,
531 P: BorrowToSql,
532 I: IntoIterator<Item = P>,
533 I::IntoIter: ExactSizeIterator,
534 {
535 let statement = statement.__convert().into_statement(&self.inner).await?;
536 query::execute(self.inner(), statement, params).await
537 }
538
539 /// Executes a `COPY FROM STDIN` statement, returning a sink used to write the copy data.
540 ///
541 /// PostgreSQL does not support parameters in `COPY` statements, so this method does not take any. The copy *must*
542 /// be explicitly completed via the `Sink::close` or `finish` methods. If it is not, the copy will be aborted.
543 pub async fn copy_in<T, U>(&self, statement: &T) -> Result<CopyInSink<U>, Error>
544 where
545 T: ?Sized + ToStatement,
546 U: Buf + 'static + Send,
547 {
548 let statement = statement.__convert().into_statement(&self.inner).await?;
549 copy_in::copy_in(self.inner(), statement).await
550 }
551
552 /// Executes a `COPY TO STDOUT` statement, returning a stream of the resulting data.
553 ///
554 /// PostgreSQL does not support parameters in `COPY` statements, so this method does not take any.
555 pub async fn copy_out<T>(&self, statement: &T) -> Result<CopyOutStream, Error>
556 where
557 T: ?Sized + ToStatement,
558 {
559 let statement = statement.__convert().into_statement(&self.inner).await?;
560 copy_out::copy_out(self.inner(), statement).await
561 }
562
563 /// Executes a sequence of SQL statements using the simple query protocol, returning the resulting rows.
564 ///
565 /// Statements should be separated by semicolons. If an error occurs, execution of the sequence will stop at that
566 /// point. The simple query protocol returns the values in rows as strings rather than in their binary encodings,
567 /// so the associated row type doesn't work with the `FromSql` trait. Rather than simply returning a list of the
568 /// rows, this method returns a list of an enum which indicates either the completion of one of the commands,
569 /// or a row of data. This preserves the framing between the separate statements in the request.
570 ///
571 /// # Warning
572 ///
573 /// Prepared statements should be use for any query which contains user-specified data, as they provided the
574 /// functionality to safely embed that data in the request. Do not form statements via string concatenation and pass
575 /// them to this method!
576 pub async fn simple_query(&self, query: &str) -> Result<Vec<SimpleQueryMessage>, Error> {
577 self.simple_query_raw(query).await?.try_collect().await
578 }
579
580 /// Executes a sequence of SQL statements using the simple query protocol, returning the resulting rows as a stream.
581 ///
582 /// Statements should be separated by semicolons. If an error occurs, execution of the sequence will stop at that
583 /// point. The simple query protocol returns the values in rows as strings rather than in their binary encodings,
584 /// so the associated row type doesn't work with the `FromSql` trait. Rather than simply returning a list of the
585 /// rows, this method returns a list of an enum which indicates either the completion of one of the commands,
586 /// or a row of data. This preserves the framing between the separate statements in the request.
587 ///
588 /// # Warning
589 ///
590 /// Prepared statements should be use for any query which contains user-specified data, as they provided the
591 /// functionality to safely embed that data in the request. Do not form statements via string concatenation and pass
592 /// them to this method!
593 pub async fn simple_query_raw(&self, query: &str) -> Result<SimpleQueryStream, Error> {
594 simple_query::simple_query(self.inner(), query).await
595 }
596
597 /// Executes a sequence of SQL statements using the simple query protocol.
598 ///
599 /// Statements should be separated by semicolons. If an error occurs, execution of the sequence will stop at that
600 /// point. This is intended for use when, for example, initializing a database schema.
601 ///
602 /// # Warning
603 ///
604 /// Prepared statements should be use for any query which contains user-specified data, as they provided the
605 /// functionality to safely embed that data in the request. Do not form statements via string concatenation and pass
606 /// them to this method!
607 pub async fn batch_execute(&self, query: &str) -> Result<(), Error> {
608 simple_query::batch_execute(self.inner(), query).await
609 }
610
611 /// Check that the connection is alive and wait for the confirmation.
612 pub async fn check_connection(&self) -> Result<(), Error> {
613 // sync is a very quick message to test the connection health.
614 query::sync(self.inner()).await
615 }
616
617 /// Begins a new database transaction.
618 ///
619 /// The transaction will roll back by default - use the `commit` method to commit it.
620 pub async fn transaction(&mut self) -> Result<Transaction<'_>, Error> {
621 self.build_transaction().start().await
622 }
623
624 /// Returns a builder for a transaction with custom settings.
625 ///
626 /// Unlike the `transaction` method, the builder can be used to control the transaction's isolation level and other
627 /// attributes.
628 pub fn build_transaction(&mut self) -> TransactionBuilder<'_> {
629 TransactionBuilder::new(self)
630 }
631
632 /// Constructs a cancellation token that can later be used to request cancellation of a query running on the
633 /// connection associated with this client.
634 pub fn cancel_token(&self) -> CancelToken {
635 CancelToken {
636 #[cfg(feature = "runtime")]
637 socket_config: self.socket_config.clone(),
638 ssl_mode: self.ssl_mode,
639 ssl_negotiation: self.ssl_negotiation,
640 process_id: self.process_id,
641 secret_key: self.secret_key,
642 }
643 }
644
645 /// Attempts to cancel an in-progress query.
646 ///
647 /// The server provides no information about whether a cancellation attempt was successful or not. An error will
648 /// only be returned if the client was unable to connect to the database.
649 ///
650 /// Requires the `runtime` Cargo feature (enabled by default).
651 #[cfg(feature = "runtime")]
652 #[deprecated(since = "0.6.0", note = "use Client::cancel_token() instead")]
653 pub async fn cancel_query<T>(&self, tls: T) -> Result<(), Error>
654 where
655 T: MakeTlsConnect<Socket>,
656 {
657 self.cancel_token().cancel_query(tls).await
658 }
659
660 /// Like `cancel_query`, but uses a stream which is already connected to the server rather than opening a new
661 /// connection itself.
662 #[deprecated(since = "0.6.0", note = "use Client::cancel_token() instead")]
663 pub async fn cancel_query_raw<S, T>(&self, stream: S, tls: T) -> Result<(), Error>
664 where
665 S: AsyncRead + AsyncWrite + Unpin,
666 T: TlsConnect<S>,
667 {
668 self.cancel_token().cancel_query_raw(stream, tls).await
669 }
670
671 /// Clears the client's type information cache.
672 ///
673 /// When user-defined types are used in a query, the client loads their definitions from the database and caches
674 /// them for the lifetime of the client. If those definitions are changed in the database, this method can be used
675 /// to flush the local cache and allow the new, updated definitions to be loaded.
676 pub fn clear_type_cache(&self) {
677 self.inner().clear_type_cache();
678 }
679
680 /// Determines if the connection to the server has already closed.
681 ///
682 /// In that case, all future queries will fail.
683 pub fn is_closed(&self) -> bool {
684 self.inner.sender.is_closed()
685 }
686
687 #[doc(hidden)]
688 pub fn __private_api_rollback(&self, name: Option<&str>) {
689 let buf = self.inner().with_buf(|buf| {
690 if let Some(name) = name {
691 frontend::query(&format!("ROLLBACK TO {}", name), buf).unwrap();
692 } else {
693 frontend::query("ROLLBACK", buf).unwrap();
694 }
695 buf.split().freeze()
696 });
697 let _ = self
698 .inner()
699 .send(RequestMessages::Single(FrontendMessage::Raw(buf)));
700 }
701
702 #[doc(hidden)]
703 pub fn __private_api_close(&mut self) {
704 self.inner.sender.close_channel()
705 }
706}
707
708impl fmt::Debug for Client {
709 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
710 f.debug_struct("Client").finish()
711 }
712}