Struct Connection 

pub struct Connection { /* private fields */ }

A connection to a single Kafka broker.

Implementations

impl Connection

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

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

Errors

Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.

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

Connect to addr and honour options.security.

This method makes a secured (TLS/SASL) dial when a policy is set, and a plaintext dial otherwise. It is the single connect entry point for every metadata-client site (pool, admin, RLMM fetch loop), so the plaintext-versus-secured branch cannot drift between them. The plaintext (None) path is byte-identical to Self::connect.

Errors

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

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

Connect to addr and apply 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, ClientError::Tls or ClientError::Sasl if the TLS handshake or the SASL exchange fails with no verdict from the peer, ClientError::Authentication if the peer rejects authentication, ClientError::InvalidConfig if the TLS settings do not build, or ClientError::Io if the security policy is internally inconsistent (e.g. a TLS protocol with no TLS config).

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.

This method negotiates API versions over the stream and returns a usable Connection. The broker’s InterBrokerClient integration calls it: the TLS + SASL handshake runs before this call, so the stream is already authenticated. From here on the connection’s normal request / response framing applies.

Errors

Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.

Panics

Panics if synchronized client state is poisoned or a response violates an invariant established by protocol validation.

pub fn advertised_api_range(&self, api_key: i16) -> Option<(i16, i16)>

Return the peer-advertised version range for api_key.

Higher-level clients use this to distinguish an API that is absent from a listener’s surface from one whose advertised versions merely do not overlap their codec range.

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

Send a typed request and await the typed response.

This method negotiates the version from the broker-advertised table that connect populated. It encodes and decodes the request and response headers 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.

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

Encode and enqueue a typed request that intentionally has no response. Kafka Produce with acks=0 is the standard use case.

Errors

Returns an error if version negotiation or encoding fails, or if the connection writer has stopped before accepting the frame.

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 method bypasses the typed ProtocolRequest codegen path so callers can speak Krabka-private APIs whose wire types live outside krabka-protocol, for example the controller’s Raft RPCs at api keys 1000+.

This method always writes the header as RequestHeader v2 (flexible) with an empty trailing tagged-fields byte. It assumes the response uses ResponseHeader v1 (flexible): the I/O loop strips the 4-byte correlation id, and this method strips the leading tagged-fields byte before it returns. Callers receive the raw body bytes only.

body is the encoded request body, which is 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.

pub fn versions(&self) -> &ApiVersionTable

Negotiated API versions known to this connection.

pub fn in_flight(&self) -> usize

The number of requests that wait for a response on this connection.

pub fn is_closed(&self) -> bool

Whether the connection has closed: the peer closed it, an I/O error ended it, it was idle for connections_max_idle, or Self::close ran. A closed connection fails every request with Disconnected.

pub fn close(self)

Close the connection, cancelling all background tasks.

pub fn attach_metrics(&self, metrics: &NetworkMetrics)

Count this connection in the client’s network metrics from now on. A connection counts in one set of metrics; a second call does nothing.

Trait Implementations

impl Clone for Connection

fn clone(&self) -> Connection

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.