// Connection establishment and top-level client lifecycle APIs.
///|
/// Open a PostgreSQL connection and return both cooperating runtime handles.
///
/// The returned `Client` only enqueues work; no I/O happens unless the paired
/// `Connection` is actively driven by calling `Connection::run` in another task.
/// This split keeps request submission cheap and makes backpressure explicit.
pub async fn connect(config : Config) -> (Client, Connection) {
match config.connect_timeout_ms {
Some(timeout_ms) =>
@async.with_timeout(
timeout_ms,
() => connect_inner(config),
error=ClientError::Protocol("connection timed out"),
)
None => connect_inner(config)
}
}
///|
/// Establish a fully initialized client runtime for one concrete target.
async fn connect_inner(config : Config) -> (Client, Connection) {
let setup = connect_stream(config)
let startup = startup(setup, config)
let types : Map[@proto.Oid, Type] = Map([])
let background_group : @ref.Ref[@async.TaskGroup[Unit]?] = @ref.new(None)
seed_builtin_types(types)
let shared = {
config,
requests: Queue(kind=Unbounded),
async_messages: Queue(kind=Unbounded),
background_group,
parameters: startup.parameters,
types,
process_id: @ref.new(startup.process_id),
secret_key: @ref.new(startup.secret_key),
transaction_status: @ref.new(startup.transaction_status),
closed: @ref.new(false),
closing: @ref.new(false),
next_id: @ref.new(0),
}
({ shared, }, { shared, stream: setup.stream, })
}
///|
/// Return whether the shared runtime has fully closed.
///
/// This becomes `true` after the connection loop exits or after a graceful
/// shutdown request has been processed to completion.
pub fn Client::is_closed(self : Client) -> Bool {
self.shared.closed.val
}
///|
/// Request a graceful shutdown of the client runtime.
///
/// This method only enqueues a `Terminate` request; it does not block waiting
/// for the socket to close. Callers that need to observe completion should wait
/// for the `Connection::run` task to return.
pub fn Client::close(self : Client) -> Unit {
if self.shared.closed.val || self.shared.closing.val {
return
}
self.shared.closing.val = true
let responses = @async.Queue(kind=Unbounded)
try
self.shared.requests.try_put({
kind: Terminate,
bytes: terminate_bytes(),
responses,
copy_input: None,
})
catch {
_ => {
self.shared.closed.val = true
responses.close()
}
} noraise {
_ => ()
}
}
///|
/// Create a token capable of cancelling the current backend operation.
///
/// The token stays valid until the underlying connection closes. It can be
/// stored and used from another task or timeout handler when a long-running
/// query should be interrupted.
pub fn Client::cancel_token(self : Client) -> CancelToken {
{
config: self.shared.config,
process_id: self.shared.process_id,
secret_key: self.shared.secret_key,
closed: self.shared.closed,
}
}
///|
/// Return the backend process ID embedded in the cancellation token.
pub fn CancelToken::process_id(self : CancelToken) -> Int {
self.process_id.val
}
///|
/// Return the backend secret key embedded in the cancellation token.
pub fn CancelToken::secret_key(self : CancelToken) -> Int {
self.secret_key.val
}
///|
/// Reset the shared type cache to the built-in PostgreSQL descriptors only.
///
/// Use this when the application expects server-side type definitions to change
/// and wants later queries to re-fetch catalog metadata lazily.
pub fn Client::clear_type_cache(self : Client) -> Unit {
self.shared.types.clear()
seed_builtin_types(self.shared.types)
}
///|
/// Look up a server parameter learned during startup or later async updates.
pub fn Client::parameter(self : Client, name : String) -> String? {
self.shared.parameters.get(name)
}
///|
/// Perform a cheap round trip that validates the connection is still usable.
///
/// The method sends a bare `Sync`, waits for the matching `ReadyForQuery`, and
/// therefore confirms that both the outbound request path and inbound response
/// path are still alive.
pub async fn Client::check_connection(self : Client) -> Unit {
ensure_open(self.shared)
let responses = self.send_request(Messages, sync_bytes())
drain_sync_response(responses)
}
///|
/// Send a PostgreSQL cancel request over a short-lived control connection.
///
/// PostgreSQL requires cancellation to happen on a separate TCP connection that
/// carries only the backend process ID and secret key from startup.
pub async fn CancelToken::cancel(self : CancelToken) -> Unit {
if self.closed.val {
return
}
let conn = @socket.Tcp::connect_to_host(
self.config.connect_host(),
port=self.config.port,
)
defer conn.close()
let buf = Buffer()
@frontend.cancel_request(self.process_id.val, self.secret_key.val, buf)
conn.write(buf.to_bytes())
}