Struct Connection

Source

pub struct Connection { /* private fields */ }

Expand description

A connection to a single Kafka broker.

Implementations§

Source §

impl Connection

Source

pub async fn connect( addr: SocketAddr, options: ConnectionOptions, ) -> Result<Self, ClientError>

Connect to addr, negotiate API versions, return a usable Connection.

Source

pub async fn connect_with_options( addr: SocketAddr, options: ConnectionOptions, ) -> Result<Self, ClientError>

Connect to addr honouring options.security: a secured (TLS/SASL) dial when a policy is set, plaintext otherwise.

This is the single connect entry point for every metadata-client site (pool, admin, RLMM fetch loop) so the plaintext-vs-secured branch can’t drift between them. The plaintext (None) path is byte-identical to Self::connect.

§Errors

Propagates Self::connect / Self::connect_secured failures.

Source

pub async fn connect_secured( addr: SocketAddr, options: ConnectionOptions, security: &ClientSecurity, ) -> Result<Self, ClientError>

Connect to addr, applying security (TLS then SASL) before the API-versions bootstrap. Plaintext is identical to Self::connect.

§Errors

Returns ClientError::Connect / ClientError::Timeout on the TCP dial, or ClientError::Io if the TLS or SASL handshake fails or the security policy is internally inconsistent (e.g. a TLS protocol with no TLS config).

Source

pub async fn from_stream( stream: Box<dyn ClientDuplex>, options: ConnectionOptions, ) -> Result<Self, ClientError>

Build a Connection over a pre-established, optionally pre-authenticated stream. Negotiates API versions over the stream and returns a usable Connection.

Used by the broker’s InterBrokerClient integration: TLS + SASL handshake run before this call, so the stream is already authenticated. From here on the connection’s normal request / response framing applies.

Source

pub async fn send<R: ProtocolRequest>( &self, req: R, ) -> Result<R::Response, ClientError>

Send a typed request and await the typed response.

The version is negotiated from the broker-advertised table populated during connect. The request and response headers are encoded and decoded automatically.

§Errors

Returns ClientError::IncompatibleVersion if there is no mutually supported version, ClientError::Disconnected if the I/O loop has exited, or ClientError::Timeout if no response arrives in time.

Source

pub async fn raw_request( &self, api_key: i16, api_version: i16, body: Bytes, ) -> Result<Bytes, ClientError>

Send a hand-framed request and await the raw response body.

This bypasses the typed ProtocolRequest codegen path so callers can speak Crabka-private APIs (e.g., the controller’s Raft RPCs at api keys 1000+) whose wire types live outside crabka-protocol.

The header is always written as RequestHeader v2 (flexible) with an empty trailing tagged-fields byte. The response is assumed to use ResponseHeader v1 (flexible): the I/O loop strips the 4-byte correlation id, and this method strips the leading tagged-fields byte before returning. Callers receive the raw body bytes only.

body is the encoded request body (everything after the request header), exactly as it should appear on the wire.

§Errors

Returns ClientError::Disconnected if the I/O loop has exited or ClientError::Timeout if no response arrives within the configured request timeout.

Source