barnabas_client/lib.rs
1//! The client, written once and generic over its sockets.
2//!
3//! [`barnabas_core`] holds the parts with no IO at all — framing, correlation,
4//! filtering, the producer's sequencing rules. This crate holds the part that
5//! *sends*: connection pooling, leader routing, metadata refresh, the
6//! consumer's fetch loop and the producer's transaction flow. All of it is
7//! generic over one small [`Transport`], so a runtime binding supplies four
8//! functions and nothing else.
9//!
10//! # Why the seam is four functions
11//!
12//! The first version of this crate did not exist: `barnabas-glommio` held all
13//! 1368 lines of it. That made the design's claim — "bindings are ~300 lines"
14//! — false, and would have made a second binding a copy of the first, with the
15//! usual consequence that the two drift and only one gets the bug fix.
16//!
17//! So the split is by *what actually differs between runtimes*, which turns out
18//! to be: open a socket, read, write, sleep. Everything above that is protocol
19//! and is identical everywhere.
20//!
21//! # No `Send` bounds, deliberately
22//!
23//! Nothing here requires `Send`, on the futures or on the transport. That is
24//! what lets a thread-per-core binding hold `!Send` sockets while a
25//! work-stealing binding hands out a handle usable across threads — the
26//! *binding* decides, and neither choice is imposed by this crate.
27//!
28//! An abstraction that required `Send` here would forbid the per-core case
29//! outright, which is exactly the lowest-common-denominator failure that made
30//! a runtime *trait* the wrong shape for `slipstream-rt` (see that crate's
31//! `RtCtx`). The difference is that this trait abstracts over **sockets**, not
32//! over runtimes: it never spawns, never names an executor, and has nothing to
33//! lose by staying bound-free.
34
35use std::future::Future;
36use std::io;
37use std::time::Duration;
38
39pub mod admin;
40pub mod builder;
41pub mod cluster;
42pub mod consumer;
43pub mod group;
44mod join;
45pub mod producer;
46pub mod sasl;
47mod timeout;
48
49pub use admin::{Admin, BrokerInfo, NewTopic};
50pub use builder::{ConsumerBuilder, ProducerBuilder, StartOffset};
51pub use cluster::Cluster;
52pub use consumer::{Consumer, ConsumerRecords, RebalanceListener, RecordRef, EARLIEST, LATEST};
53pub use group::{ClassicProtocol, GroupMetadata, GroupProtocol, Membership};
54pub use producer::{Producer, ProducerRecord};
55pub use sasl::{Credentials, SaslMechanism};
56
57use barnabas_core::{Disposition, ErrorCode};
58
59/// What a runtime must provide: a socket and a timer.
60///
61/// **`connect` takes `&self`** so a transport can carry configuration —
62/// a TLS client config, a root store, a server-name policy. That is what makes
63/// encryption a binding-level concern rather than something this crate has to
64/// know about: `Consumer<GlommioTls>` and `Consumer<Glommio>` are the same
65/// client over different sockets.
66///
67/// The rest are associated functions: reading, writing and sleeping need no
68/// configuration, and requiring `&self` for them would mean borrowing the
69/// transport across every request for nothing.
70pub trait Transport: 'static {
71 /// The runtime's stream — a TCP socket, or a TLS session over one.
72 type Stream: 'static;
73
74 /// Open a connection to `host:port`.
75 fn connect(&self, addr: &str) -> impl Future<Output = io::Result<Self::Stream>>;
76
77 /// Read into `buf`, returning the byte count. Zero means the peer closed.
78 fn read(stream: &mut Self::Stream, buf: &mut [u8]) -> impl Future<Output = io::Result<usize>>;
79
80 /// Write all of `buf`.
81 fn write_all(stream: &mut Self::Stream, buf: &[u8]) -> impl Future<Output = io::Result<()>>;
82
83 /// Sleep, for retry backoff.
84 fn sleep(dur: Duration) -> impl Future<Output = ()>;
85}
86
87#[derive(Debug, thiserror::Error)]
88pub enum Error {
89 #[error("core: {0}")]
90 Core(#[from] barnabas_core::Error),
91
92 #[error("io: {0}")]
93 Io(#[from] io::Error),
94
95 #[error("connect {addr}: {source}")]
96 Connect {
97 addr: String,
98 #[source]
99 source: io::Error,
100 },
101
102 /// A broker error code, with what the client should do about it. Carrying
103 /// the [`Disposition`] means a caller can react without re-deriving the
104 /// taxonomy — and cannot accidentally retry something fatal.
105 #[error("{op} failed with error code {code} ({disposition:?})")]
106 Broker {
107 op: &'static str,
108 code: i16,
109 disposition: Disposition,
110 },
111
112 /// The partition has no leader even after a metadata refresh — what a
113 /// partition mid-election looks like. Separate from [`Self::Broker`] so a
114 /// caller can back off and retry rather than treat it as fatal.
115 #[error("{topic}-{partition} has no leader")]
116 NoLeader { topic: String, partition: i32 },
117
118 /// A misuse of the producer, caught by the state machine rather than by a
119 /// broker — producing outside a transaction, to an unenrolled partition, or
120 /// after being fenced.
121 #[error("producer: {0}")]
122 Producer(#[from] barnabas_core::producer::ProducerError),
123
124 /// A request outlived its deadline. The connection is dropped with it —
125 /// see [`Cluster::call_at`](cluster::Cluster).
126 #[error("{op:?} to {addr} timed out")]
127 Timeout {
128 op: kafka_protocol::messages::ApiKey,
129 addr: String,
130 },
131
132 /// Authentication failed, or the broker does not offer the mechanism.
133 #[error("sasl: {0}")]
134 Sasl(String),
135
136 #[error("the broker's response contained no {0}")]
137 Missing(&'static str),
138
139 /// A request/response call was attempted on a connection that already has
140 /// requests in flight.
141 ///
142 /// **This is a bug in this client, surfaced rather than suffered.** Kafka
143 /// answers a connection's requests in order, so a `call` on a busy
144 /// connection reads the *previous* request's response — a fetch decoded as
145 /// an offset commit, or worse, a response that happens to parse. Anything
146 /// pipelining deliberately uses `send`/`recv` and manages the order itself;
147 /// anything else must leave the connection idle first.
148 #[error("{op:?} on {addr}: {in_flight} request(s) already in flight on this connection")]
149 ConnectionBusy {
150 op: kafka_protocol::messages::ApiKey,
151 addr: String,
152 in_flight: usize,
153 },
154}
155
156pub type Result<T> = std::result::Result<T, Error>;
157
158pub(crate) fn check(op: &'static str, code: i16) -> Result<()> {
159 let code = ErrorCode(code);
160 if code.is_ok() {
161 return Ok(());
162 }
163 Err(Error::Broker {
164 op,
165 code: code.0,
166 disposition: code.disposition(),
167 })
168}