/// Streaming variant of `{{ method_name }}`. Drives the Rust core stream
/// natively, yielding each chunk to the caller's block. When called
/// without a block, returns a `{{ iter_name }}` (Enumerable via its `each`).
fn {{ method_name }}(&self, req: {{ request_binding }}) -> Result<magnus::Value, Error> {
use magnus::IntoValue;
use magnus::value::ReprValue;
let inner = self.inner.clone();
let core_req: {{ request_core }} = req.into();
// 16 MiB: tokio's ~2 MB default worker stack can overflow on a deep extraction
// future (a nested archive member, a multi-stage OCR pipeline), and a stack overflow
// aborts the process with SIGBUS instead of raising a catchable panic.
const STREAM_RUNTIME_STACK_SIZE_BYTES: usize = 16 * 1024 * 1024;
let runtime = std::sync::Arc::new(tokio::runtime::Builder::new_multi_thread()
.enable_all()
.thread_stack_size(STREAM_RUNTIME_STACK_SIZE_BYTES)
.build()
.map_err(|e| {
magnus::Error::new(unsafe { Ruby::get_unchecked() }.exception_runtime_error(), e.to_string())
})?);
let stream = runtime.block_on(async { {{ core_call }}.await })
.map_err(alef_magnus_err)?;
let iterator = {{ iter_name }} {
inner: Arc::new(tokio::sync::Mutex::new(Some(stream))),
runtime,
};
let ruby = unsafe { Ruby::get_unchecked() };
if ruby.block_given() {
// Drive the stream synchronously, yielding each chunk to the block.
iterator.each()?;
Ok(ruby.qnil().as_value())
} else {
Ok(iterator.into_value_with(&ruby))
}
}