postgleam/connection

Low-level PostgreSQL connection: TCP/TLS connect, authentication and the wire protocol state machine.

Most applications should use postgleam (a connection actor) or postgleam/pool. This module is for building on the protocol directly (COPY, streaming, replication, LISTEN/NOTIFY).

Every function takes a ConnectionState and returns the updated state. After an Error, keep using the state you had before the call only if is_recoverable says so; otherwise close the connection.

Types

ConnectionState

</>

Connection state.

pub type ConnectionState {
  ConnectionState(
    transport: @internal Transport,
    connection_id: option.Option(Int),
    connection_key: option.Option(Int),
    parameters: dict.Dict(String, String),
    transaction_status: @internal TransactionStatus,
    buffer: BitArray,
    notifications: List(Notification),
    max_message_size: Int,
  )
}

Constructors

ConnectionState(
  transport: @internal Transport,
  connection_id: option.Option(Int),
  connection_key: option.Option(Int),
  parameters: dict.Dict(String, String),
  transaction_status: @internal TransactionStatus,
  buffer: BitArray,
  notifications: List(Notification),
  max_message_size: Int,
)

Arguments

notifications: LISTEN/NOTIFY notifications received but not yet taken, oldest first.

ExtendedQueryResult

</>

Result from an extended query execution.

pub type ExtendedQueryResult {
  ExtendedQueryResult(
    tag: String,
    columns: List(@internal RowField),
    rows: List(List(option.Option(value.Value))),
  )
}

Constructors

ExtendedQueryResult(
  tag: String,
  columns: List(@internal RowField),
  rows: List(List(option.Option(value.Value))),
)

Failure

</>

A failed operation, plus the connection state if the connection is still usable (the server reached ReadyForQuery and nothing is left unread).

state: None means the connection is in an unknown state and must be closed: a timeout, a socket error or a protocol violation.

pub type Failure {
  Failure(
    error: error.Error,
    state: option.Option(ConnectionState),
  )
}

Constructors

Failure(
  error: error.Error,
  state: option.Option(ConnectionState),
)

Notification

</>

An asynchronous notification from LISTEN/NOTIFY.

pub type Notification {
  Notification(channel: String, payload: String, pg_pid: Int)
}

Constructors

Notification(channel: String, payload: String, pg_pid: Int)

PreparedStatement

</>

A prepared statement returned by prepare.

pub type PreparedStatement {
  PreparedStatement(
    name: String,
    statement: String,
    param_oids: List(Int),
    result_fields: List(@internal RowField),
  )
}

Constructors

PreparedStatement(
  name: String,
  statement: String,
  param_oids: List(Int),
  result_fields: List(@internal RowField),
)

SimpleQueryResult

</>

Result from a simple query.

pub type SimpleQueryResult {
  SimpleQueryResult(
    tag: String,
    columns: List(String),
    rows: List(List(option.Option(String))),
  )
}

Constructors

SimpleQueryResult(
  tag: String,
  columns: List(String),
  rows: List(List(option.Option(String))),
)

StreamChunk

</>

Result of one streamed Execute.

pub type StreamChunk {
  StreamMore(rows: List(List(option.Option(value.Value))))
  StreamDone(
    tag: String,
    rows: List(List(option.Option(value.Value))),
  )
}

Constructors

StreamMore(rows: List(List(option.Option(value.Value))))

More rows available; call execute_portal again.

StreamDone(
  tag: String,
  rows: List(List(option.Option(value.Value))),
)

All rows received, with the command tag. Call sync_portal next.

Values

bind_and_execute_portal

</>

pub fn bind_and_execute_portal(
  state: ConnectionState,
  prepared: PreparedStatement,
  params: List(option.Option(value.Value)),
  registry: dict.Dict(Int, codec.Codec),
  max_rows: Int,
  timeout: Int,
) -> Result(#(StreamChunk, ConnectionState), error.Error)

Bind parameters and fetch the first chunk: Bind + Execute(max_rows) + Flush. Flush (not Sync) keeps the unnamed portal open. On error the connection is resynchronised before returning.

cancel

</>

pub fn cancel(
  state: ConnectionState,
  config: config.Config,
) -> Result(Nil, error.Error)

Ask the server to cancel whatever this connection is running. Opens a separate short-lived connection, as the protocol requires. Best effort: the server may have finished already.

close_statement

</>

pub fn close_statement(
  state: ConnectionState,
  name: String,
  timeout: Int,
) -> Result(ConnectionState, error.Error)

Close a prepared statement: Close + Sync.

connect

</>

pub fn connect(
  config: config.Config,
) -> Result(ConnectionState, error.Error)

Establish a connection, complete authentication, and return ready state.

disconnect

</>

pub fn disconnect(state: ConnectionState) -> Nil

Send Terminate and close the socket.

execute_pipeline

</>

pub fn execute_pipeline(
  state: ConnectionState,
  prepared: PreparedStatement,
  param_sets: List(List(option.Option(value.Value))),
  registry: dict.Dict(Int, codec.Codec),
  timeout: Int,
) -> Result(
  #(List(ExtendedQueryResult), ConnectionState),
  error.Error,
)

Execute a prepared statement with multiple param sets in one pipeline: N × (Bind + Execute) + one Sync, in a single write.

execute_portal

</>

pub fn execute_portal(
  state: ConnectionState,
  prepared: PreparedStatement,
  registry: dict.Dict(Int, codec.Codec),
  max_rows: Int,
  timeout: Int,
) -> Result(#(StreamChunk, ConnectionState), error.Error)

Fetch the next chunk from the open portal: Execute(max_rows) + Flush.

execute_prepared

</>

pub fn execute_prepared(
  state: ConnectionState,
  prepared: PreparedStatement,
  params: List(option.Option(value.Value)),
  registry: dict.Dict(Int, codec.Codec),
  timeout: Int,
) -> Result(#(ExtendedQueryResult, ConnectionState), error.Error)

Execute a prepared statement: Bind + Execute + Sync, sent as one write.

extended_query

</>

pub fn extended_query(
  state: ConnectionState,
  sql: String,
  params: List(option.Option(value.Value)),
  registry: dict.Dict(Int, codec.Codec),
  timeout: Int,
) -> Result(#(ExtendedQueryResult, ConnectionState), error.Error)

One-off extended query with an unnamed statement: Parse + Describe + Sync, then Bind + Execute + Sync (two round trips).

extended_query_pipeline_unnamed

</>

pub fn extended_query_pipeline_unnamed(
  state: ConnectionState,
  sql: String,
  param_sets: List(List(option.Option(value.Value))),
  registry: dict.Dict(Int, codec.Codec),
  timeout: Int,
) -> Result(
  #(List(ExtendedQueryResult), ConnectionState),
  error.Error,
)

PgBouncer-safe batch query: Parse + Describe + Flush, then N × (Bind + Execute) + Sync.

extended_query_unnamed

</>

pub fn extended_query_unnamed(
  state: ConnectionState,
  sql: String,
  params: List(option.Option(value.Value)),
  registry: dict.Dict(Int, codec.Codec),
  timeout: Int,
) -> Result(#(ExtendedQueryResult, ConnectionState), error.Error)

PgBouncer-safe extended query using the unnamed statement. Parse + Describe + Flush (no ReadyForQuery, so PgBouncer keeps the backend), then Bind + Execute + Sync.

is_recoverable

</>

pub fn is_recoverable(err: error.Error) -> Bool

Whether an error leaves the connection usable. Socket errors, timeouts and protocol errors do not; server errors, encode and decode errors do.

message_name

</>

pub fn message_name(msg: @internal BackendMessage) -> String

A short name for a backend message, for error messages.

prepare

</>

pub fn prepare(
  state: ConnectionState,
  name: String,
  sql: String,
  type_oids: List(Int),
  timeout: Int,
) -> Result(#(PreparedStatement, ConnectionState), error.Error)

Prepare a named statement: Parse + Describe + Sync.

receive_message

</>

pub fn receive_message(
  state: ConnectionState,
  timeout: Int,
) -> Result(
  #(@internal BackendMessage, ConnectionState),
  error.Error,
)

Receive the next backend message.

Asynchronous messages are handled here and never returned: NotificationResponse is queued in state.notifications, ParameterStatus updates state.parameters, and NoticeResponse is dropped.

send_bytes

</>

pub fn send_bytes(
  state: ConnectionState,
  bytes: BitArray,
) -> Result(ConnectionState, error.Error)

Send raw bytes.

send_message

</>

pub fn send_message(
  state: ConnectionState,
  msg: @internal FrontendMessage,
) -> Result(ConnectionState, error.Error)

Send one frontend message.

send_messages

</>

pub fn send_messages(
  state: ConnectionState,
  msgs: List(@internal FrontendMessage),
) -> Result(ConnectionState, error.Error)

Send several frontend messages in one write.

simple_query

</>

pub fn simple_query(
  state: ConnectionState,
  sql: String,
  timeout: Int,
) -> Result(
  #(List(SimpleQueryResult), ConnectionState),
  error.Error,
)

Execute a simple query (text protocol) and return all results.

sync_portal

</>

pub fn sync_portal(
  state: ConnectionState,
  timeout: Int,
) -> Result(ConnectionState, error.Error)

End portal streaming: Sync, then ReadyForQuery.

take_notifications

</>

pub fn take_notifications(
  state: ConnectionState,
) -> #(List(Notification), ConnectionState)

Take all queued notifications, oldest first.

wait_for_notifications

</>

pub fn wait_for_notifications(
  state: ConnectionState,
  timeout: Int,
) -> Result(#(List(Notification), ConnectionState), error.Error)

Take the notifications received so far (oldest first) and, if there are none, wait up to timeout milliseconds for one to arrive. The connection must be idle (not in the middle of a query).