// Client configuration: transport/security settings shared by every client
// surface plus per-surface (producer/consumer) shells. Validation happens at
// construction; Phase 1 wires the security fields into the transport.
///|
/// Transport security. SASL layers authenticate right after connecting;
/// only plaintext is wired up so far.
pub(all) enum SecurityProtocol {
Plaintext
Ssl
SaslPlaintext
SaslSsl
} derive(@debug.Debug)
///|
/// SASL mechanism. OAUTHBEARER token callbacks arrive with the Phase 1 SASL
/// work; PLAIN and SCRAM use username/password.
pub(all) enum SaslMechanism {
Plain
ScramSha256
ScramSha512
OAuthBearer
} derive(@debug.Debug)
///|
pub(all) struct SaslConfig {
mechanism : SaslMechanism
username : String
password : String
} derive(@debug.Debug)
///|
/// One "host:port" (or "[ipv6]:port") entry of the bootstrap list.
pub(all) struct HostPort {
host : String
port : Int
} derive(Eq, Compare, Hash, @debug.Debug)
///|
const DEFAULT_KAFKA_PORT : Int = 9092
///|
/// Parse a single bootstrap entry. The port defaults to 9092 when omitted;
/// bare IPv6 addresses must be bracketed.
pub fn HostPort::parse(entry : String) -> HostPort raise ProtocolError {
if entry.length() == 0 {
raise ProtocolError::ProtocolError("empty bootstrap server entry")
}
let chars : Array[Char] = []
for c in entry {
chars.push(c)
}
let n = chars.length()
if chars[0] == '[' {
// [ipv6] or [ipv6]:port
let mut close = -1
for i in 1.. 0 else {
raise ProtocolError::ProtocolError(
"bootstrap entry \{entry} has an unterminated [ipv6] bracket",
)
}
let host = chars_to_string(chars, 1, close)
if close + 1 == n {
return { host, port: DEFAULT_KAFKA_PORT, }
}
if chars[close + 1] == ':' {
let port = parse_port(chars, close + 2, entry)
return { host, port, }
}
raise ProtocolError::ProtocolError(
"bootstrap entry \{entry} has trailing garbage after [ipv6]",
)
}
// host or host:port — split on the last colon
let mut colon = -1
for i in 0.. String {
let sb = StringBuilder()
for i in start.. Int raise ProtocolError {
if start >= chars.length() {
raise ProtocolError::ProtocolError(
"bootstrap entry \{entry} has an empty port",
)
}
let mut port = 0
for i in start.. 57 {
raise ProtocolError::ProtocolError(
"bootstrap entry \{entry} has a non-numeric port",
)
}
port = port * 10 + (v - 48)
if port > 65535 {
raise ProtocolError::ProtocolError(
"bootstrap entry \{entry} has a port above 65535",
)
}
}
if port < 1 {
raise ProtocolError::ProtocolError(
"bootstrap entry \{entry} port must be >= 1",
)
}
port
}
///|
/// Settings shared by every client surface.
pub struct CommonConfig {
bootstrap_servers : Array[String]
client_id : String
request_timeout_ms : Int
connection_max_idle_ms : Int
retries : Int
retry_backoff_ms : Int
retry_backoff_max_ms : Int
security_protocol : SecurityProtocol
sasl : SaslConfig?
/// TLS settings for Ssl / SaslSsl protocols; None synthesizes defaults
/// whose server name is the dialed host.
tls : TlsClientOptions?
} derive(@debug.Debug)
///|
/// Defaults mirror the Java client where sensible: 30s request timeout,
/// 9-minute idle connection limit. Retries are bounded at 10 for now —
/// the Java default is effectively unbounded but is gated by
/// delivery.timeout.ms, which the batching producer (Phase 3) introduces.
pub fn CommonConfig::new(
bootstrap_servers : Array[String],
request_timeout_ms? : Int = 30000,
security_protocol? : SecurityProtocol = Plaintext,
sasl? : SaslConfig? = None,
tls? : TlsClientOptions? = None,
) -> CommonConfig raise {
let config : CommonConfig = {
bootstrap_servers,
client_id: "moonkafka",
request_timeout_ms,
connection_max_idle_ms: 540000,
retries: 10,
retry_backoff_ms: 100,
retry_backoff_max_ms: 1000,
security_protocol,
sasl,
tls,
}
config.validate()
config
}
///|
/// Reject values that would make a client misbehave; called by `new`.
pub fn CommonConfig::validate(self : CommonConfig) -> Unit raise ProtocolError {
if self.bootstrap_servers.is_empty() {
raise ProtocolError::ProtocolError("bootstrap_servers must not be empty")
}
ignore(self.bootstrap_addresses())
if self.client_id.length() == 0 {
raise ProtocolError::ProtocolError("client_id must not be empty")
}
if self.request_timeout_ms <= 0 {
raise ProtocolError::ProtocolError(
"request_timeout_ms must be positive, got \{self.request_timeout_ms}",
)
}
if self.connection_max_idle_ms <= 0 {
raise ProtocolError::ProtocolError(
"connection_max_idle_ms must be positive, got \{self.connection_max_idle_ms}",
)
}
if self.retries < 0 {
raise ProtocolError::ProtocolError("retries must be >= 0")
}
if self.retry_backoff_ms <= 0 ||
self.retry_backoff_max_ms < self.retry_backoff_ms {
raise ProtocolError::ProtocolError(
"retry backoff needs backoff_ms > 0 and max_ms >= backoff_ms, got \{self.retry_backoff_ms}/\{self.retry_backoff_max_ms}",
)
}
match (self.security_protocol, self.sasl) {
(SaslPlaintext, None) | (SaslSsl, None) =>
raise ProtocolError::ProtocolError(
"SASL security protocols require sasl credentials",
)
(_, Some(sasl)) =>
if sasl.username.length() == 0 {
raise ProtocolError::ProtocolError("sasl username must not be empty")
}
_ => ()
}
}
///|
/// Parse every bootstrap entry; raises on the first malformed one.
pub fn CommonConfig::bootstrap_addresses(
self : CommonConfig,
) -> Array[HostPort] raise ProtocolError {
let out = []
for entry in self.bootstrap_servers {
out.push(HostPort::parse(entry))
}
out
}
///|
pub struct ProducerConfig {
common : CommonConfig
topic : String
acks : Int
} derive(@debug.Debug)
///|
/// `acks` is 1 (leader acknowledgement) or -1 (all in-sync replicas);
/// 0 becomes valid with the Phase 3 sender that does not read responses.
pub fn ProducerConfig::new(
bootstrap_servers : Array[String],
topic : String,
acks? : Int = 1,
request_timeout_ms? : Int = 30000,
security_protocol? : SecurityProtocol = Plaintext,
sasl? : SaslConfig? = None,
tls? : TlsClientOptions? = None,
) -> ProducerConfig raise {
if acks != 1 && acks != -1 {
raise ProtocolError::ProtocolError(
"acks must be 1 (leader) or -1 (all), got \{acks}",
)
}
if topic.length() == 0 {
raise ProtocolError::ProtocolError("topic must not be empty")
}
{
common: CommonConfig::new(
bootstrap_servers,
request_timeout_ms~,
security_protocol~,
sasl~,
tls~,
),
topic,
acks,
}
}
///|
pub struct ConsumerConfig {
common : CommonConfig
topic : String
start_from : StartFrom
} derive(@debug.Debug)
///|
pub fn ConsumerConfig::new(
bootstrap_servers : Array[String],
topic : String,
start_from? : StartFrom = Earliest,
request_timeout_ms? : Int = 30000,
security_protocol? : SecurityProtocol = Plaintext,
sasl? : SaslConfig? = None,
tls? : TlsClientOptions? = None,
) -> ConsumerConfig raise {
if topic.length() == 0 {
raise ProtocolError::ProtocolError("topic must not be empty")
}
{
common: CommonConfig::new(
bootstrap_servers,
request_timeout_ms~,
security_protocol~,
sasl~,
tls~,
),
topic,
start_from,
}
}