dnim/rust-postgres
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Docs

Browse Source
This commit is contained in:
Steven Fackler
2019-01-07 20:43:52 -08:00
parent 1f6d9ddc06
commit e372cdc6f9
3 changed files with 206 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
tokio-postgres/src/config.rs
View File

@@ -22,10 +22,13 @@ use crate::proto::HandshakeFuture;
use crate::{Connect, MakeTlsMode, Socket};
use crate::{Error, Handshake, TlsMode};
/// Properties required of a database.
#[cfg(feature = "runtime")]
#[derive(Debug, Copy, Clone, PartialEq)]
pub enum TargetSessionAttrs {
/// No special permissions are required.
Any,
/// The database must be writable.
ReadWrite,
#[doc(hidden)]
__NonExhaustive,
@@ -60,6 +63,77 @@ pub(crate) struct Inner {
pub(crate) target_session_attrs: TargetSessionAttrs,
}
/// Connection configuration.
///
/// Configuration can be parsed from libpq-style connection strings. These strings come in two formats:
///
/// # Key-Value
///
/// This format consists of space-separated key-value pairs. Values which are either the empty string or contain
/// whitespace should be wrapped in `'`. `'` and `\` characters should be backslash-escaped.
///
/// ## Keys
///
/// * `user` - The username to authenticate with. Required.
/// * `password` - The password to authenticate with.
/// * `dbname` - The name of the database to connect to. Defaults to the username.
/// * `options` - Command line options used to configure the server.
/// * `application_name` - Sets the `application_name` parameter on the server.
/// * `host` - The host to connect to. On Unix platforms, if the host starts with a `/` character it is treated as the
/// path to the directory containing Unix domain sockets. Otherwise, it is treated as a hostname. Multiple hosts
/// can be specified, separated by commas. Each host will be tried in turn when connecting. Required if connecting
/// with the `connect` method.
/// * `port` - The port to connect to. Multiple ports can be specified, separated by commas. The number of ports must be
/// either 1, in which case it will be used for all hosts, or the same as the number of hosts. Defaults to 5432 if
/// omitted or the empty string.
/// * `connect_timeout` - The time limit in seconds applied to each socket-level connection attempt. Note that hostnames
/// can resolve to multiple IP addresses, and this limit is applied to each address. Defaults to no timeout.
/// * `keepalives` - Controls the use of TCP keepalive. A value of 0 disables keepalive and nonzero integers enable it.
/// This option is ignored when connecting with Unix sockets. Defaults to on.
/// * `keepalives_idle` - The number of seconds of inactivity after which a keepalive message is sent to the server.
/// This option is ignored when connecting with Unix sockets. Defaults to 2 hours.
/// * `target_session_attrs` - Specifies requirements of the session. If set to `read-write`, the client will check that
/// the `transaction_read_write` session parameter is set to `on`. This can be used to connect to the primary server
/// in a database cluster as opposed to the secondary read-only mirrors. Defaults to `all`.
///
/// ## Examples
///
/// ```not_rust
/// host=localhost user=postgres connect_timeout=10 keepalives=0
/// ```
///
/// ```not_rust
/// host=/var/lib/postgresql,localhost port=1234 user=postgres password='password with spaces'
/// ```
///
/// ```not_rust
/// host=host1,host2,host3 port=1234,,5678 user=postgres target_session_attrs=read-write
/// ```
///
/// # Url
///
/// This format resembles a URL with a scheme of either `postgres://` or `postgresql://`. All components are optional,
/// and the format accept query parameters for all of the key-value pairs described in the section above. Multiple
/// host/port pairs can be comma-separated. Unix socket paths in the host section of the URL should be percent-encoded,
/// as the path component of the URL specifies the database name.
///
/// ## Examples
///
/// ```not_rust
/// postgresql://user@localhost
/// ```
///
/// ```not_rust
/// postgresql://user:password@%2Fvar%2Flib%2Fpostgresql/mydb?connect_timeout=10
/// ```
///
/// ```not_rust
/// postgresql://user@host1:1234,host2host3:5678?target_session_attrs=read-write
/// ```
///
/// ```not_rust
/// postgresql:///mydb?user=user&host=/var/lib/postgresql
/// ```
#[derive(Debug, Clone, PartialEq)]
pub struct Config(pub(crate) Arc<Inner>);

166
tokio-postgres/src/lib.rs
View File

@@ -54,7 +54,7 @@
//! The client supports *pipelined* requests. Pipelining can improve performance in use cases in which multiple,
//! independent queries need to be executed. In a traditional workflow, each query is sent to the server after the
//! previous query completes. In contrast, pipelining allows the client to send all of the queries to the server up
//! front, eliminating time spent on both sides waiting for the other to finish sending data:
//! front, minimizing time spent by one side waiting for the other to finish sending data:
//!
//! ```not_rust
//! Sequential Pipelined
@@ -75,31 +75,7 @@
//! the connection to work concurrently when possible.
//!
//! Pipelining happens automatically when futures are polled concurrently (for example, by using the futures `join`
//! combinator). Say we want to prepare 2 statements:
//!
//! ```no_run
//! use futures::Future;
//! use tokio_postgres::{Client, Error, Statement};
//!
//! fn prepare_sequential(
//! client: &mut Client,
//! ) -> impl Future<Item = (Statement, Statement), Error = Error>
//! {
//! client.prepare("SELECT * FROM foo")
//! .and_then({
//! let f = client.prepare("INSERT INTO bar (id, name) VALUES ($1, $2)");
//! |s1| f.map(|s2| (s1, s2))
//! })
//! }
//!
//! fn prepare_pipelined(
//! client: &mut Client,
//! ) -> impl Future<Item = (Statement, Statement), Error = Error>
//! {
//! client.prepare("SELECT * FROM foo")
//! .join(client.prepare("INSERT INTO bar (id, name) VALUES ($1, $2)"))
//! }
//! ```
//! combinator). Say we want to prepare 2 statements.
//!
//! # Runtime
//!
@@ -143,6 +119,13 @@ fn next_portal() -> String {
format!("p{}", ID.fetch_add(1, Ordering::SeqCst))
}
/// A convenience function which parses a connection string and connects to the database.
///
/// See the documentation for [`Config`] for details on the connection string format.
///
/// Requires the `runtime` Cargo feature (enabled by default).
///
/// [`Config`]: ./Config.t.html
#[cfg(feature = "runtime")]
pub fn connect<T>(config: &str, tls_mode: T) -> Connect<T>
where
@@ -151,33 +134,73 @@ where
Connect(proto::ConnectFuture::new(tls_mode, config.parse()))
}
/// An asynchronous PostgreSQL client.
///
/// The client is one half of what is returned when a connection is established. Users interact with the database
/// through this client object.
pub struct Client(proto::Client);
impl Client {
/// Creates a new prepared statement.
///
/// Prepared statements can be executed repeatedly, and may contain query parameters (indicated by `$1`, `$2`, etc),
/// which are set when executed. Prepared statements can only be used with the connection that created them.
pub fn prepare(&mut self, query: &str) -> Prepare {
self.prepare_typed(query, &[])
}
/// Like `prepare`, but allows the types of query parameters to be explicitly specified.
///
/// The list of types may be smaller than the number of parameters - the types of the remaining parameters will be
/// inferred. For example, `client.prepare_typed(query, &[])` is equivalent to `client.prepare(query)`.
pub fn prepare_typed(&mut self, query: &str, param_types: &[Type]) -> Prepare {
Prepare(self.0.prepare(next_statement(), query, param_types))
}
/// Executes a statement, returning the number of rows modified.
///
/// If the statement does not modify any rows (e.g. `SELECT`), 0 is returned.
///
/// # Panics
///
/// Panics if the number of parameters provided does not match the number expected.
pub fn execute(&mut self, statement: &Statement, params: &[&dyn ToSql]) -> Execute {
Execute(self.0.execute(&statement.0, params))
}
/// Executes a statement, returning a stream of the resulting rows.
///
/// # Panics
///
/// Panics if the number of parameters provided does not match the number expected.
pub fn query(&mut self, statement: &Statement, params: &[&dyn ToSql]) -> Query {
Query(self.0.query(&statement.0, params))
}
/// Binds a statement to a set of parameters, creating a `Portal` which can be incrementally queried.
///
/// Portals only last for the duration of the transaction in which they are created - in particular, a portal
/// created outside of a transaction is immediately destroyed. Portals can only be used on the connection that
/// created them.
/// # Panics
///
/// Panics if the number of parameters provided does not match the number expected.
pub fn bind(&mut self, statement: &Statement, params: &[&dyn ToSql]) -> Bind {
Bind(self.0.bind(&statement.0, next_portal(), params))
}
/// Continues execution of a portal, returning a stream of the resulting rows.
///
/// Unlike `query`, portals can be incrementally evaluated by limiting the number of rows returned in each call to
/// query_portal. If the requested number is negative or 0, all rows will be returned.
pub fn query_portal(&mut self, portal: &Portal, max_rows: i32) -> QueryPortal {
QueryPortal(self.0.query_portal(&portal.0, max_rows))
}
/// Executes a `COPY FROM STDIN` statement, returning the number of rows created.
///
/// The data in the provided stream is passed along to the server verbatim; it is the caller's responsibility to
/// ensure it uses the proper format.
pub fn copy_in<S>(
&mut self,
statement: &Statement,
@@ -194,18 +217,36 @@ impl Client {
CopyIn(self.0.copy_in(&statement.0, params, stream))
}
/// Executes a `COPY TO STDOUT` statement, returning a stream of the resulting data.
pub fn copy_out(&mut self, statement: &Statement, params: &[&dyn ToSql]) -> CopyOut {
CopyOut(self.0.copy_out(&statement.0, params))
}
/// Executes a sequence of SQL statements.
///
/// Statements should be separated by semicolons. If an error occurs, execution of the sequence will stop at that
/// point. This is intended for the execution of batches of non-dynamic statements, for example, the creation of
/// a schema for a fresh database.
///
/// # Warning
///
/// Prepared statements should be use for any query which contains user-specified data, as they provided the
/// functionality to safely imbed that data in the request. Do not form statements via string concatenation and pass
/// them to this method!
pub fn batch_execute(&mut self, query: &str) -> BatchExecute {
BatchExecute(self.0.batch_execute(query))
}
pub fn transaction(&mut self) -> TransactionBuilder {
TransactionBuilder(self.0.clone())
}
pub fn batch_execute(&mut self, query: &str) -> BatchExecute {
BatchExecute(self.0.batch_execute(query))
}
/// Attempts to cancel an in-progress query.
///
/// The server provides no information about whether a cancellation attempt was successful or not. An error will
/// only be returned if the client was unable to connect to the database.
///
/// Requires the `runtime` Cargo feature (enabled by default).
#[cfg(feature = "runtime")]
pub fn cancel_query<T>(&mut self, make_tls_mode: T) -> CancelQuery<T>
where
@@ -214,6 +255,8 @@ impl Client {
CancelQuery(self.0.cancel_query(make_tls_mode))
}
/// Like `cancel_query`, but uses a stream which is already connected to the server rather than opening a new
/// connection itself.
pub fn cancel_query_raw<S, T>(&mut self, stream: S, tls_mode: T) -> CancelQueryRaw<S, T>
where
S: AsyncRead + AsyncWrite,
@@ -222,15 +265,31 @@ impl Client {
CancelQueryRaw(self.0.cancel_query_raw(stream, tls_mode))
}
/// Determines if the connection to the server has already closed.
///
/// In that case, all future queries will fail.
pub fn is_closed(&self) -> bool {
self.0.is_closed()
}
/// Polls the client to check if it is idle.
///
/// A connection is idle if there are no outstanding requests, whether they have begun being polled or not. For
/// example, this can be used by a connection pool to ensure that all work done by one checkout is done before
/// making the client available for a new request. Otherwise, any non-completed work from the first request could
/// interleave with the second.
pub fn poll_idle(&mut self) -> Poll<(), Error> {
self.0.poll_idle()
}
}
/// A connection to a PostgreSQL database.
///
/// This is one half of what is returned when a new connection is established. It performs the actual IO with the
/// server, and should generally be spawned off onto an executor to run in the background.
///
/// `Connection` implements `Future`, and only resolves when the connection is closed, either because a fatal error has
/// occurred, or because its associated `Client` has dropped and all outstanding work has completed.
#[must_use = "futures do nothing unless polled"]
pub struct Connection<S>(proto::Connection<S>);
@@ -238,10 +297,15 @@ impl<S> Connection<S>
where
S: AsyncRead + AsyncWrite,
{
/// Returns the value of a runtime parameter for this connection.
pub fn parameter(&self, name: &str) -> Option<&str> {
self.0.parameter(name)
}
/// Polls for asynchronous messages from the server.
///
/// The server can send notices as well as notifications asynchronously to the client. Applications which wish to
/// examine those messages should use this method to drive the connection rather than its `Future` implementation.
pub fn poll_message(&mut self) -> Poll<Option<AsyncMessage>, Error> {
self.0.poll_message()
}
@@ -259,9 +323,16 @@ where
}
}
/// An asynchronous message from the server.
#[allow(clippy::large_enum_variant)]
pub enum AsyncMessage {
/// A notice.
///
/// Notices use the same format as errors, but aren't "errors" per-se.
Notice(DbError),
/// A notification.
///
/// Connections can subscribe to notifications with the `LISTEN` command.
Notification(Notification),
#[doc(hidden)]
__NonExhaustive,
@@ -361,14 +432,19 @@ impl Future for Prepare {
}
}
/// A prepared statement.
///
/// Prepared statements can only be used with the connection that created them.
#[derive(Clone)]
pub struct Statement(proto::Statement);
impl Statement {
/// Returns the expected types of the statement's parameters.
pub fn params(&self) -> &[Type] {
self.0.params()
}
/// Returns information about the columns returned when the statement is queried.
pub fn columns(&self) -> &[Column] {
self.0.columns()
}
@@ -426,6 +502,10 @@ impl Stream for QueryPortal {
}
}
/// A portal.
///
/// Portals can only be used with the connection that created them, and only exist for the duration of the transaction
/// in which they were created.
pub struct Portal(proto::Portal);
#[must_use = "futures do nothing unless polled"]
@@ -512,10 +592,24 @@ impl Future for BatchExecute {
/// An asynchronous notification.
#[derive(Clone, Debug)]
pub struct Notification {
/// The process ID of the notifying backend process.
pub process_id: i32,
/// The name of the channel that the notify has been raised on.
pub channel: String,
/// The "payload" string passed from the notifying process.
pub payload: String,
process_id: i32,
channel: String,
payload: String,
}
impl Notification {
/// The process ID of the notifying backend process.
pub fn process_id(&self) -> i32 {
self.process_id
}
/// The name of the channel that the notify has been raised on.
pub fn channel(&self) -> &str {
&self.channel
}
/// The "payload" string passed from the notifying process.
pub fn payload(&self) -> &str {
&self.payload
}
}

2
tokio-postgres/src/proto/client.rs
View File

@@ -284,6 +284,8 @@ impl Client {
name: &str,
params: &[&dyn ToSql],
) -> Result<Vec<u8>, Error> {
assert_eq!(statement.params().len(), params.len());
let mut buf = vec![];
let r = frontend::bind(
name,