rightsize_modules/
redpanda.rs

1//! A single-node Redpanda broker (Kafka API-compatible) with its schema registry
2//! enabled.
3
4use rightsize::{Container, ContainerGuard, Result, Wait};
5
6/// The alias siblings resolve INTERNAL through — native docker networks, or the msb
7/// exec-tunnel alias emulation (best-effort; this port has no sibling Kafka consumer
8/// module of its own yet).
9const INTERNAL_ALIAS: &str = "redpanda";
10
11/// A single-node Redpanda broker.
12pub struct RedpandaContainer(Container);
13
14impl RedpandaContainer {
15    const KAFKA_PORT: u16 = 9092;
16    const INTERNAL_KAFKA_PORT: u16 = 9093;
17    const SCHEMA_REGISTRY_PORT: u16 = 8081;
18
19    /// Builds a container from the pinned default image
20    /// (`redpandadata/redpanda:v24.2.4`).
21    pub fn new() -> Self {
22        Self::with_image("redpandadata/redpanda:v24.2.4")
23    }
24
25    /// Builds a container from a caller-chosen image.
26    pub fn with_image(image: &str) -> Self {
27        let container = Container::new(image)
28            .with_exposed_ports(&[
29                Self::KAFKA_PORT,
30                Self::INTERNAL_KAFKA_PORT,
31                Self::SCHEMA_REGISTRY_PORT,
32            ])
33            .waiting_for(Wait::for_log_message(
34                ".*Successfully started Redpanda.*",
35                1,
36            ))
37            .with_spec_customizer(|mut spec, mapped| {
38                // The advertised listener must carry the *mapped host port*, known
39                // only now (ports were allocated before boot). This is why the
40                // customizer takes `mapped` — a one-shot rewrite the instant before
41                // create, so the broker advertises an address host clients can
42                // actually reach. EXTERNAL advertises that mapped host port for host
43                // clients; INTERNAL advertises the fixed alias:port siblings resolve
44                // on the container network. See KafkaContainer's customizer for the
45                // same trick applied to a single advertised listener.
46                let kafka_mapped = mapped(Self::KAFKA_PORT);
47                spec.command = Some(
48                    [
49                        "redpanda",
50                        "start",
51                        "--mode",
52                        "dev-container",
53                        "--smp",
54                        "1",
55                        "--kafka-addr",
56                        "EXTERNAL://0.0.0.0:9092,INTERNAL://0.0.0.0:9093",
57                        "--advertise-kafka-addr",
58                    ]
59                    .into_iter()
60                    .map(String::from)
61                    .chain(std::iter::once(format!(
62                        "EXTERNAL://127.0.0.1:{kafka_mapped},INTERNAL://{INTERNAL_ALIAS}:9093"
63                    )))
64                    .chain(
65                        ["--schema-registry-addr", "0.0.0.0:8081"]
66                            .into_iter()
67                            .map(String::from),
68                    )
69                    .collect(),
70                );
71                spec
72            });
73        Self(container)
74    }
75
76    /// Boots the container.
77    pub async fn start(self) -> Result<RedpandaGuard> {
78        crate::register_default_backends();
79        Ok(RedpandaGuard(self.0.start().await?))
80    }
81}
82
83impl Default for RedpandaContainer {
84    fn default() -> Self {
85        Self::new()
86    }
87}
88
89/// The running guard for a [`RedpandaContainer`].
90pub struct RedpandaGuard(ContainerGuard);
91
92impl RedpandaGuard {
93    /// The `PLAINTEXT://` bootstrap-servers address for the running broker (EXTERNAL
94    /// listener).
95    pub fn bootstrap_servers(&self) -> String {
96        format!(
97            "PLAINTEXT://{}:{}",
98            self.0.host(),
99            self.0
100                .get_mapped_port(RedpandaContainer::KAFKA_PORT)
101                .unwrap()
102        )
103    }
104
105    /// The schema registry's base URI for the running broker.
106    pub fn schema_registry_url(&self) -> String {
107        format!(
108            "http://{}:{}",
109            self.0.host(),
110            self.0
111                .get_mapped_port(RedpandaContainer::SCHEMA_REGISTRY_PORT)
112                .unwrap()
113        )
114    }
115
116    /// Stops and removes the container, releasing its host ports.
117    pub async fn stop(self) -> Result<()> {
118        self.0.stop().await
119    }
120}
121
122impl std::ops::Deref for RedpandaGuard {
123    type Target = ContainerGuard;
124    fn deref(&self) -> &ContainerGuard {
125        &self.0
126    }
127}
128
129#[cfg(test)]
130mod tests {
131    use super::*;
132
133    #[test]
134    fn with_image_smoke() {
135        let _ = RedpandaContainer::new();
136    }
137}
rightsize_modules/redpanda.rs

rightsize_modules/
redpanda.rs
