rusteron_archive/
testing.rs

1use crate::IntoCString;
2use crate::{Aeron, AeronArchive, AeronArchiveAsyncConnect, AeronArchiveContext, AeronContext};
3use log::info;
4use log::{error, warn};
5use regex::Regex;
6use std::backtrace::Backtrace;
7use std::ffi::CString;
8use std::path::Path;
9use std::process::{Child, Command, ExitStatus, Stdio};
10use std::thread::sleep;
11use std::time::{Duration, Instant};
12use std::{fs, io, panic, process};
13
14/// Check whether the test is running under Valgrind, used to scale Rust-side
15/// timeouts that would otherwise fire before their C counterpart finishes.
16pub(crate) fn running_under_valgrind() -> bool {
17    std::env::var_os("RUSTERON_VALGRIND").is_some()
18}
19
20/// Return a `Duration` scaled by 3× when running under Valgrind: the same
21/// multiplier the client crate uses for driver / liveness timeouts.
22pub(crate) fn valgrind_timeout(base_secs: u64) -> Duration {
23    if running_under_valgrind() {
24        Duration::from_secs(base_secs.saturating_mul(3))
25    } else {
26        Duration::from_secs(base_secs)
27    }
28}
29
30pub struct EmbeddedArchiveMediaDriverProcess {
31    child: Child,
32    pub aeron_dir: CString,
33    pub archive_dir: CString,
34    pub control_request_channel: String,
35    pub control_response_channel: String,
36    pub recording_events_channel: String,
37}
38
39impl EmbeddedArchiveMediaDriverProcess {
40    /// Builds the Aeron Archive project and starts an embedded Aeron Archive Media Driver process.
41    ///
42    /// This function ensures that the necessary Aeron `.jar` files are built using Gradle. If the required
43    /// `.jar` files are not found in the expected directory, it runs the Gradle build tasks to generate them.
44    /// Once the build is complete, it invokes the `start` function to initialize and run the Aeron Archive Media Driver.
45    ///
46    /// # Parameters
47    /// - `aeron_dir`: The directory for the Aeron media driver to use for its IPC mechanisms.
48    /// - `archive_dir`: The directory where the Aeron Archive will store its recordings and metadata.
49    /// - `control_request_channel`: The channel URI used for sending control requests to the Aeron Archive.
50    /// - `control_response_channel`: The channel URI used for receiving control responses from the Aeron Archive.
51    /// - `recording_events_channel`: The channel URI used for receiving recording event notifications from the Aeron Archive.
52    ///
53    /// # Returns
54    /// On success, returns an instance of `EmbeddedArchiveMediaDriverProcess` encapsulating the child process
55    /// and configuration used. Returns an `io::Result` if the process fails to start or the build fails.
56    ///
57    /// # Errors
58    /// Returns an `io::Result::Err` if:
59    /// - The Gradle build fails to execute or complete.
60    /// - The required `.jar` files are still not found after building.
61    /// - The `start` function encounters an error starting the process.
62    ///
63    /// # Example
64    /// ```
65    /// use rusteron_archive::testing::EmbeddedArchiveMediaDriverProcess;
66    /// let driver = EmbeddedArchiveMediaDriverProcess::build_and_start(
67    ///     "/tmp/aeron-dir",
68    ///     "/tmp/archive-dir",
69    ///     "aeron:udp?endpoint=localhost:8010",
70    ///     "aeron:udp?endpoint=localhost:8011",
71    ///     "aeron:udp?endpoint=localhost:8012",
72    /// ).expect("Failed to build and start Aeron Archive Media Driver");
73    /// ```
74    ///
75    /// # Notes
76    /// - This function assumes the presence of a Gradle wrapper script (`gradlew` or `gradlew.bat`)
77    ///   in the `aeron` directory relative to the project's root (`CARGO_MANIFEST_DIR`).
78    /// - The required `.jar` files will be generated in `aeron/aeron-all/build/libs` if not already present.
79    /// - The `build_and_start` function is a convenience wrapper for automating the build and initialization process.
80    pub fn build_and_start(
81        aeron_dir: &str,
82        archive_dir: &str,
83        control_request_channel: &str,
84        control_response_channel: &str,
85        recording_events_channel: &str,
86    ) -> io::Result<Self> {
87        let path = std::path::MAIN_SEPARATOR;
88        let gradle = if cfg!(target_os = "windows") {
89            &format!("{}{path}aeron{path}gradlew.bat", env!("CARGO_MANIFEST_DIR"),)
90        } else {
91            "./gradlew"
92        };
93        let dir = format!("{}{path}aeron", env!("CARGO_MANIFEST_DIR"),);
94        info!("running {} in {}", gradle, dir);
95
96        if !Path::new(&format!(
97            "{}{path}aeron{path}aeron-all{path}build{path}libs",
98            env!("CARGO_MANIFEST_DIR")
99        ))
100        .exists()
101        {
102            Command::new(&gradle)
103                .current_dir(dir)
104                .args([
105                    ":aeron-agent:jar",
106                    ":aeron-samples:jar",
107                    ":aeron-archive:jar",
108                    ":aeron-all:build",
109                ])
110                .stdout(Stdio::inherit())
111                .stderr(Stdio::inherit())
112                .spawn()?
113                .wait()?;
114        }
115
116        return Self::start(
117            &aeron_dir,
118            archive_dir,
119            control_request_channel,
120            control_response_channel,
121            recording_events_channel,
122        );
123    }
124
125    pub fn run_aeron_stats(&self) -> std::io::Result<Child> {
126        let main_dir = env!("CARGO_MANIFEST_DIR");
127        let dir = format!("{}/{}", main_dir, &self.aeron_dir.to_str().unwrap());
128        info!("running 'just aeron-stat {}'", dir);
129        Command::new("just")
130            .args(["aeron-stat", dir.as_str()])
131            .stdout(Stdio::inherit())
132            .stderr(Stdio::inherit())
133            .spawn()
134    }
135
136    pub fn archive_connect(&self) -> Result<(AeronArchive, Aeron), io::Error> {
137        let start = Instant::now();
138        let deadline = valgrind_timeout(30);
139        while start.elapsed() < deadline {
140            if let Ok(aeron_context) = AeronContext::new() {
141                aeron_context.set_dir(&self.aeron_dir).expect("invalid dir");
142                aeron_context
143                    .set_client_name(&CString::new("unit_test_client")?)
144                    .expect("invalid client name");
145                if let Ok(aeron) = Aeron::new(&aeron_context) {
146                    if aeron.start().is_ok() {
147                        if let Ok(archive_context) = AeronArchiveContext::new() {
148                            archive_context.set_aeron(&aeron).expect("invalid aeron");
149                            archive_context
150                                .set_control_request_channel(&self.control_request_channel.as_str().into_c_string())
151                                .expect("invalid control request channel");
152                            archive_context
153                                .set_control_response_channel(&self.control_response_channel.as_str().into_c_string())
154                                .expect("invalid control response channel");
155                            archive_context
156                                .set_recording_events_channel(&self.recording_events_channel.as_str().into_c_string())
157                                .expect("invalid recording events channel");
158                            if let Ok(connect) = AeronArchiveAsyncConnect::new_with_aeron(&archive_context, &aeron) {
159                                if let Ok(archive) = connect.poll_blocking(valgrind_timeout(10)) {
160                                    let i = archive.get_archive_id();
161                                    assert!(i > 0);
162                                    info!("aeron archive media driver is up [connected with archive id {i}]");
163                                    sleep(Duration::from_millis(100));
164                                    return Ok((archive, aeron));
165                                };
166                            }
167                        }
168                        error!("aeron error: {}", Aeron::errmsg());
169                    }
170                }
171            }
172            info!("waiting for aeron to start up, retrying...");
173        }
174
175        assert!(start.elapsed() < deadline, "failed to start up aeron media driver");
176
177        return Err(std::io::Error::other("unable to start up aeron media driver client"));
178    }
179
180    /// Starts an embedded Aeron Archive Media Driver process with the specified configurations.
181    ///
182    /// This function cleans and recreates the Aeron and archive directories, configures the JVM to run
183    /// the Aeron Archive Media Driver, and starts the process with the specified control channels.
184    /// It ensures that the environment is correctly prepared for Aeron communication.
185    ///
186    /// # Parameters
187    /// - `aeron_dir`: The directory for the Aeron media driver to use for its IPC mechanisms.
188    /// - `archive_dir`: The directory where the Aeron Archive will store its recordings and metadata.
189    /// - `control_request_channel`: The channel URI used for sending control requests to the Aeron Archive.
190    /// - `control_response_channel`: The channel URI used for receiving control responses from the Aeron Archive.
191    /// - `recording_events_channel`: The channel URI used for receiving recording event notifications from the Aeron Archive.
192    ///
193    /// # Returns
194    /// On success, returns an instance of `EmbeddedArchiveMediaDriverProcess` encapsulating the child process
195    /// and configuration used. Returns an `io::Result` if the process fails to start.
196    ///
197    /// # Errors
198    /// Returns an `io::Result::Err` if:
199    /// - Cleaning or creating the directories fails.
200    /// - The required `.jar` files are missing or not found.
201    /// - The Java process fails to start.
202    ///
203    /// # Example
204    /// ```
205    /// use rusteron_archive::testing::EmbeddedArchiveMediaDriverProcess;
206    /// let driver = EmbeddedArchiveMediaDriverProcess::start(
207    ///     "/tmp/aeron-dir",
208    ///     "/tmp/archive-dir",
209    ///     "aeron:udp?endpoint=localhost:8010",
210    ///     "aeron:udp?endpoint=localhost:8011",
211    ///     "aeron:udp?endpoint=localhost:8012",
212    /// ).expect("Failed to start Aeron Archive Media Driver");
213    /// ```
214    ///
215    /// # Notes
216    /// - The Aeron `.jar` files must be available under the directory `aeron/aeron-all/build/libs` relative
217    ///   to the project's root (`CARGO_MANIFEST_DIR`).
218    /// - The function configures the JVM with properties for Aeron, such as enabling event logging and disabling bounds checks.
219    pub fn start(
220        aeron_dir: &str,
221        archive_dir: &str,
222        control_request_channel: &str,
223        control_response_channel: &str,
224        recording_events_channel: &str,
225    ) -> io::Result<Self> {
226        Self::clean_directory(aeron_dir)?;
227        Self::clean_directory(archive_dir)?;
228
229        // Ensure directories are recreated
230        fs::create_dir_all(aeron_dir)?;
231        fs::create_dir_all(archive_dir)?;
232
233        let binding = fs::read_dir(format!("{}/aeron/aeron-all/build/libs", env!("CARGO_MANIFEST_DIR")))?
234            .filter(|f| f.is_ok())
235            .map(|f| f.unwrap())
236            .filter(|f| f.file_name().to_string_lossy().to_string().ends_with(".jar"))
237            .next()
238            .unwrap()
239            .path();
240        let mut jar_path = binding.to_str().unwrap();
241        let agent_jar = jar_path.replace("aeron-all", "aeron-agent");
242
243        assert!(fs::exists(jar_path).unwrap_or_default());
244        let mut args = vec![];
245
246        if fs::exists(&agent_jar).unwrap_or_default() {
247            args.push(format!("-javaagent:{}", agent_jar));
248        }
249        let separator = if cfg!(target_os = "windows") { ";" } else { ":" };
250
251        let combined_jars = format!(
252            "{}{separator}{}",
253            jar_path,
254            jar_path.replace("aeron-all", "aeron-archive")
255        );
256        jar_path = &combined_jars;
257
258        args.push("--add-opens".to_string());
259        args.push("java.base/jdk.internal.misc=ALL-UNNAMED".to_string());
260        args.push("-cp".to_string());
261        args.push(jar_path.to_string());
262        args.push(format!("-Daeron.dir={}", aeron_dir));
263        args.push(format!("-Daeron.archive.dir={}", archive_dir));
264        args.push("-Daeron.spies.simulate.connection=true".to_string());
265        args.push("-Daeron.event.log=all".to_string());
266        args.push("-Daeron.event.log.disable=FRAME_IN,FRAME_OUT".to_string());
267        args.push("-Daeron.event.archive.log=all".to_string());
268        args.push("-Daeron.event.cluster.log=all".to_string());
269        args.push("-Dagrona.disable.bounds.checks=true".to_string());
270        args.push(format!("-Daeron.archive.control.channel={}", control_request_channel));
271        args.push(format!(
272            "-Daeron.archive.control.response.channel={}",
273            control_response_channel
274        ));
275        args.push(format!(
276            "-Daeron.archive.recording.events.channel={}",
277            recording_events_channel
278        ));
279        args.push("-Daeron.archive.replication.channel=aeron:udp?endpoint=localhost:0".to_string());
280        args.push("-Daeron.client.liveness.timeout=60000000000".to_string());
281        args.push("-Daeron.image.liveness.timeout=60000000000".to_string());
282        args.push("-Daeron.publication.unblock.timeout=65000000000".to_string());
283        args.push("io.aeron.archive.ArchivingMediaDriver".to_string());
284
285        info!("starting archive media driver [\n\tjava {}\n]", args.join(" "));
286
287        let child = Command::new("java")
288            .args(args)
289            .stdout(Stdio::inherit())
290            .stderr(Stdio::inherit())
291            .spawn()?;
292
293        info!(
294            "started archive media driver [{:?}",
295            fs::read_dir(aeron_dir)?.collect::<Vec<_>>()
296        );
297
298        Ok(EmbeddedArchiveMediaDriverProcess {
299            child,
300            aeron_dir: aeron_dir.into_c_string(),
301            archive_dir: archive_dir.into_c_string(),
302            control_request_channel: control_request_channel.to_string(),
303            control_response_channel: control_response_channel.to_string(),
304            recording_events_channel: recording_events_channel.to_string(),
305        })
306    }
307
308    fn clean_directory(dir: &str) -> io::Result<()> {
309        info!("cleaning directory {}", dir);
310        let path = Path::new(dir);
311        if path.exists() {
312            fs::remove_dir_all(path)?;
313        }
314        Ok(())
315    }
316
317    pub fn kill_all_java_processes() -> io::Result<ExitStatus> {
318        if cfg!(not(target_os = "windows")) {
319            return Ok(std::process::Command::new("pkill")
320                .args(["-9", "java"])
321                .stdout(Stdio::inherit())
322                .stderr(Stdio::inherit())
323                .spawn()?
324                .wait()?);
325        }
326        Ok(ExitStatus::default())
327    }
328}
329
330// Use the Drop trait to ensure process cleanup and directory removal after test completion
331impl Drop for EmbeddedArchiveMediaDriverProcess {
332    fn drop(&mut self) {
333        warn!("WARN: stopping aeron archive media driver!!!!");
334        // Attempt to kill the Java process if it’s still running
335        if let Err(e) = self.child.kill() {
336            error!("Failed to kill Java process: {}", e);
337        }
338
339        // Clean up directories after the process has terminated
340        if let Err(e) = Self::clean_directory(&self.aeron_dir.to_str().unwrap()) {
341            error!("Failed to clean up Aeron directory: {}", e);
342        }
343        if let Err(e) = Self::clean_directory(&self.archive_dir.to_str().unwrap()) {
344            error!("Failed to clean up Archive directory: {}", e);
345        }
346    }
347}
348
349/// True if a `java` executable is resolvable on `PATH`.
350///
351/// The embedded archive media driver is a Java process, so the
352/// persistent-subscription tests use this to **skip themselves** when Java
353/// isn't installed — instead of being blanket `#[ignore]`d. With Java present
354/// they run normally under `cargo test`; without it they no-op and pass.
355pub fn java_available() -> bool {
356    Command::new("java")
357        .arg("-version")
358        .stdout(Stdio::null())
359        .stderr(Stdio::null())
360        .status()
361        .is_ok()
362}
363
364/// Place at the top of any test that needs the Java archive. Skips the test
365/// (early-returns `Ok(())`) when `java` is not on `PATH`. Requires the test to
366/// return a `Result<(), _>`.
367#[macro_export]
368macro_rules! skip_unless_java {
369    () => {
370        if !$crate::testing::java_available() {
371            eprintln!("skipping: java not available on PATH");
372            return Ok(());
373        }
374    };
375}
376
377pub fn set_panic_hook() {
378    panic::set_hook(Box::new(|info| {
379        // Get the backtrace
380        let backtrace = Backtrace::force_capture();
381        error!("Stack trace: {backtrace:#?}");
382
383        let backtrace = format!("{:?}", backtrace);
384        // Regular expression to match the function, file, and line
385        let re = Regex::new(r#"fn: "([^"]+)", file: "([^"]+)", line: (\d+)"#).unwrap();
386
387        // Extract and print in IntelliJ format with function
388        for cap in re.captures_iter(&backtrace) {
389            let function = &cap[1];
390            let file = &cap[2];
391            let line = &cap[3];
392            info!("{file}:{line} in {function}");
393        }
394
395        error!("Panic occurred: {:#?}", info);
396
397        if let Some(payload) = info.payload().downcast_ref::<&str>() {
398            error!("Panic message: {}", payload);
399        } else if let Some(payload) = info.payload().downcast_ref::<String>() {
400            error!("Panic message: {}", payload);
401        } else {
402            // If it's not a &str or String, try to print it as Debug
403            error!("Panic with non-standard payload: {:?}", info.payload().type_id());
404        }
405
406        warn!("shutdown");
407
408        process::abort();
409    }))
410}
411
412/// First free UDP port at or above `start` (binds a probe socket to check).
413pub fn find_unused_udp_port(start: u16) -> Option<u16> {
414    (start..65535).find(|p| std::net::UdpSocket::bind(("127.0.0.1", *p)).is_ok())
415}
416
417pub fn find_counter_id_by_session_blocking(
418    counters_reader: &crate::AeronCountersReader,
419    session_id: i32,
420    wait: Duration,
421) -> Result<i32, crate::AeronCError> {
422    let start = Instant::now();
423    loop {
424        let counter_id = crate::RecordingPos::find_counter_id_by_session(counters_reader, session_id);
425        if counter_id >= 0 {
426            return Ok(counter_id);
427        }
428        if start.elapsed() >= wait {
429            return Err(crate::AeronCError::from_code(-1));
430        }
431        std::thread::sleep(Duration::from_millis(10));
432    }
433}
rusteron_archive/testing.rs

rusteron_archive/
testing.rs
