extern crate std;
use crate::{StdioConfig, StdioError, StdioSystem, System};
use protoflow_core::{
prelude::{vec, Bytes},
Block, BlockResult, BlockRuntime, OutputPort,
};
use protoflow_derive::Block;
use simple_mermaid::mermaid;
use std::io::Read;
const DEFAULT_BUFFER_SIZE: usize = 1024;
#[doc = mermaid!("../../doc/sys/read_stdin.mmd")]
#[doc = mermaid!("../../doc/sys/read_stdin.seq.mmd" framed)]
#[derive(Block, Clone)]
pub struct ReadStdin {
#[output]
pub output: OutputPort<Bytes>,
#[parameter]
pub buffer_size: usize,
}
impl ReadStdin {
pub fn new(output: OutputPort<Bytes>) -> Self {
Self::with_params(output, None)
}
pub fn with_params(output: OutputPort<Bytes>, buffer_size: Option<usize>) -> Self {
Self {
output,
buffer_size: buffer_size.unwrap_or(DEFAULT_BUFFER_SIZE),
}
}
}
impl Block for ReadStdin {
fn execute(&mut self, runtime: &dyn BlockRuntime) -> BlockResult {
let stdin = std::io::stdin().lock();
let mut reader = std::io::BufReader::new(stdin);
let mut buffer = vec![0; self.buffer_size];
runtime.wait_for(&self.output)?;
loop {
buffer.resize(self.buffer_size, b'\0'); buffer.fill(b'\0');
match reader.read(&mut buffer) {
Err(ref e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
Err(err) => return Err(err.into()),
Ok(0) => break, Ok(buffer_len) => {
buffer.resize(buffer_len, b'\0'); let bytes = Bytes::from(buffer.clone());
self.output.send(&bytes)?;
}
}
}
Ok(())
}
}
#[cfg(feature = "std")]
impl StdioSystem for ReadStdin {
fn build_system(_config: StdioConfig) -> Result<System, StdioError> {
use crate::{SysBlocks, SystemBuilding};
Ok(System::build(|s| {
let stdin = s.read_stdin();
let stdout = s.write_stdout();
s.connect(&stdin.output, &stdout.input);
}))
}
}
#[cfg(test)]
mod tests {
use super::ReadStdin;
use crate::{System, SystemBuilding};
#[test]
fn instantiate_block() {
let _ = System::build(|s| {
let _ = s.block(ReadStdin::new(s.output()));
});
}
}