// Bootstrap-server dialing: rotate over the configured entries so repeated
// connects (and reconnects) spread across the listed brokers, and keep
// trying until one accepts.
///|
/// Rotating view over parsed bootstrap server addresses.
pub struct BootstrapServers {
addrs : Array[HostPort]
mut cursor : Int
} derive(@debug.Debug)
///|
pub fn BootstrapServers::new(addrs : Array[HostPort]) -> BootstrapServers raise {
if addrs.is_empty() {
raise ProtocolError::ProtocolError(
"bootstrap server list must not be empty",
)
}
{ addrs, cursor: 0, }
}
///|
/// Next address in round-robin order.
pub fn BootstrapServers::next(self : BootstrapServers) -> HostPort {
let addr = self.addrs[self.cursor % self.addrs.length()]
self.cursor += 1
addr
}
///|
/// Dial the bootstrap servers in rotating order; the first connection that
/// succeeds wins. Raises the last failure if every entry refuses.
pub async fn connect_bootstrap(
servers : BootstrapServers,
client_id : String,
timeout_ms? : Int = 10000,
sasl? : SaslConfig? = None,
use_tls? : Bool = false,
tls? : TlsClientOptions? = None,
) -> BrokerConnection {
let n = servers.addrs.length()
let mut last_error : Error = ProtocolError::ProtocolError(
"no bootstrap servers",
)
for _ in 0.. return conn
Err(e) => last_error = e
}
}
raise TransportError::ConnectionClosed(
"all \{n} bootstrap servers failed; last error: \{last_error}",
)
}
///|
/// The Result escapes to the caller as data, so Ok/Err wrapping is the
/// point of this helper rather than a mechanical try? replacement.
async fn dial(
addr : HostPort,
client_id : String,
timeout_ms : Int,
sasl : SaslConfig?,
use_tls : Bool,
tls : TlsClientOptions?,
) -> Result[BrokerConnection, Error] {
// Without explicit TLS options, a TLS protocol still gets defaults whose
// server name is the dialed host.
let tls_opts = match tls {
Some(opts) => Some(opts)
None => if use_tls { Some(TlsClientOptions::new(addr.host)) } else { None }
}
try
@async.with_timeout(timeout_ms, () => {
BrokerConnection::connect(
addr.host,
addr.port,
client_id~,
sasl~,
timeout_ms~,
tls=tls_opts,
)
})
|> Ok
catch {
e => Err(e)
}
}