rumtk_core/pipelines.rs
1/*
2 * rumtk attempts to implement HL7 and medical protocols for interoperability in medicine.
3 * This toolkit aims to be reliable, simple, performant, and standards compliant.
4 * Copyright (C) 2025 Luis M. Santos, M.D. <lsantos@medicalmasses.com>
5 * Copyright (C) 2025 MedicalMasses L.L.C. <contact@medicalmasses.com>
6 *
7 * This program is free software: you can redistribute it and/or modify
8 * it under the terms of the GNU General Public License as published by
9 * the Free Software Foundation, either version 3 of the License, or
10 * (at your option) any later version.
11 *
12 * This program is distributed in the hope that it will be useful,
13 * but WITHOUT ANY WARRANTY; without even the implied warranty of
14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 * GNU General Public License for more details.
16 *
17 * You should have received a copy of the GNU General Public License
18 * along with this program. If not, see <https://www.gnu.org/licenses/>.
19 */
20
21pub mod pipeline_types {
22 use crate::base::{RUMResult, RUMVec};
23 use crate::buffers::*;
24 use crate::serde::json::{RUMDeJson, RUMSerJson};
25 use crate::strings::{RUMString, RUMStringConversions};
26 use crate::types::RUMHashMap;
27 use std::process::{Child, Command};
28
29 pub static EMPTY_COMMAND_LINE: RUMCommandLine = RUMCommandLine::new();
30
31 pub type RUMCommandArgs = Vec<RUMString>;
32 pub type RUMCommandEnv = RUMHashMap<RUMString, RUMString>;
33 #[derive(RUMSerJson, RUMDeJson, PartialEq, Default, Debug, Clone)]
34 pub struct RUMCommand {
35 pub path: RUMString,
36 #[serde(skip)]
37 pub data: Option<RUMBuffer>,
38 pub args: RUMCommandArgs,
39 pub env: RUMCommandEnv,
40 }
41
42 impl RUMCommand {
43 pub fn new(
44 prog: &str,
45 data: &Option<RUMBuffer>,
46 args: &RUMCommandArgs,
47 env: &RUMCommandEnv,
48 ) -> Self {
49 RUMCommand {
50 path: prog.to_string(),
51 args: args.clone(),
52 env: env.clone(),
53 data: data.clone(),
54 }
55 }
56 }
57
58 pub type RUMCommandLine = RUMVec<RUMCommand>;
59 pub type RUMPipelineCommand = Command;
60 pub type RUMPipelineProcess = Child;
61 pub type RUMPipeline = RUMVec<RUMPipelineProcess>;
62 pub type RUMPipelineResult = RUMResult<RUMBuffer>;
63}
64
65pub mod pipeline_functions {
66 use super::pipeline_types::*;
67 use crate::base::RUMResult;
68 use crate::strings::{rumtk_format, string_format, RUMArrayConversions, RUMString, StringReplacementPair};
69 use std::io::{Read, Write};
70 use std::os::unix::ffi::OsStrExt;
71
72 use crate::buffers::*;
73 use crate::rumtk_resolve_sync_task;
74 use std::process::{Command, Stdio};
75
76 const DEFAULT_PROCESS_ASYNC_WAIT: f32 = 0.001;
77 const DEFAULT_STDOUT_CHUNK_SIZE: usize = 1024;
78
79 ///
80 /// Given a command of type [RUMCommand](RUMCommand), generate a command instance the Rust
81 /// runtime can use to spawn a process.
82 ///
83 /// ## Example
84 ///
85 /// ```
86 /// use std::any::{Any, TypeId};
87 ///
88 /// use rumtk_core::strings::RUMString;
89 /// use rumtk_core::pipelines::pipeline_types::{RUMCommand, RUMPipelineCommand};
90 /// use rumtk_core::pipelines::pipeline_functions::pipeline_create_command;
91 ///
92 /// let command_name = "ls";
93 /// let mut command = RUMCommand::default();
94 /// command.path = RUMString::from(command_name);
95 ///
96 /// let sys_command = pipeline_create_command(&command);
97 ///
98 /// assert_eq!(sys_command.get_program().to_str().unwrap(), command_name, "");
99 ///
100 /// ```
101 ///
102 pub fn pipeline_create_command(command: &RUMCommand) -> RUMPipelineCommand {
103 let mut cmd = Command::new(command.path.as_str());
104
105 for arg in command.args.iter() {
106 cmd.arg(arg);
107 }
108
109 cmd.envs(command.env.iter());
110
111 cmd.stdin(Stdio::piped())
112 .stdout(Stdio::piped())
113 .stderr(Stdio::piped());
114
115 cmd
116 }
117
118 ///
119 /// Spawns a process out of the [RUMPipelineCommand](RUMPipelineCommand).
120 ///
121 /// ## Example
122 ///
123 /// ```
124 /// use std::any::{Any, TypeId};
125 ///
126 /// use rumtk_core::strings::RUMString;
127 /// use rumtk_core::pipelines::pipeline_types::{RUMCommand, RUMPipelineCommand};
128 /// use rumtk_core::pipelines::pipeline_functions::{pipeline_create_command, pipeline_spawn_process};
129 ///
130 /// let command_name = "ls";
131 /// let mut command = RUMCommand::default();
132 /// command.path = RUMString::from(command_name);
133 ///
134 /// let mut sys_command = pipeline_create_command(&command);
135 ///
136 /// let mut process = pipeline_spawn_process(&mut sys_command).unwrap();
137 ///
138 /// process.wait();
139 /// ```
140 ///
141 pub fn pipeline_spawn_process(cmd: &mut RUMPipelineCommand) -> RUMResult<RUMPipelineProcess> {
142 match cmd.spawn() {
143 Ok(process) => {
144 println!("Spawned process {} => {} with args {:?}", process.id(), cmd.get_program().as_bytes().to_string()?, cmd.get_args());
145 Ok(process)
146 },
147 Err(e) => Err(rumtk_format!(
148 "Failed to spawn process {:?} because => {}",
149 cmd.get_program(),
150 e
151 )),
152 }
153 }
154
155 ///
156 /// Given a process of type [RUMPipelineProcess](RUMPipelineProcess) and a rhs of type [RUMPipelineCommand](RUMPipelineCommand),
157 /// create a pipe of the lhs's `stdout` into the next command descriptor which is the rhs.
158 ///
159 /// ## Example
160 ///
161 /// ```
162 /// use std::any::{Any, TypeId};
163 /// use std::process::Stdio;
164 ///
165 /// use rumtk_core::strings::RUMString;
166 /// use rumtk_core::pipelines::pipeline_types::{RUMCommand, RUMPipelineCommand};
167 /// use rumtk_core::pipelines::pipeline_functions::{pipeline_create_command, pipeline_pipe_processes, pipeline_spawn_process};
168 ///
169 /// let ls_name = "ls";
170 /// let mut ls_command = RUMCommand::default();
171 /// ls_command.path = RUMString::from(ls_name);
172 /// let mut sys_ls_command = pipeline_create_command(&ls_command);
173 ///
174 /// let wc_name = "wc";
175 /// let mut wc_command = RUMCommand::default();
176 /// wc_command.path = RUMString::from(wc_name);
177 /// let mut sys_wc_command = pipeline_create_command(&wc_command);
178 ///
179 /// let mut sys_ls_process = pipeline_spawn_process(&mut sys_ls_command).unwrap();
180 /// pipeline_pipe_processes(&mut sys_ls_process, &mut sys_wc_command).unwrap();
181 /// let mut sys_wc_process = pipeline_spawn_process(&mut sys_wc_command).unwrap();
182 ///
183 /// sys_ls_process.wait();
184 /// sys_wc_process.wait();
185 /// ```
186 ///
187 pub fn pipeline_pipe_processes(
188 process: &mut RUMPipelineProcess,
189 piped: &mut RUMPipelineCommand,
190 ) -> RUMResult<()> {
191 let process_stdout = Stdio::from(match process.stdout.take() {
192 Some(stdout) => stdout,
193 None => {
194 return Err(rumtk_format!(
195 "No stdout handle found for process {}.",
196 process.id()
197 ));
198 }
199 });
200 let _ = piped.stdin(process_stdout);
201 Ok(())
202 }
203
204 ///
205 /// Retrieves the standard output generated by the completed process.
206 ///
207 /// ## Example
208 ///
209 /// ```
210 /// use std::any::{Any, TypeId};
211 /// use std::process::Stdio;
212 ///
213 /// use rumtk_core::strings::RUMString;
214 /// use rumtk_core::pipelines::pipeline_types::{RUMCommand, RUMPipelineCommand};
215 /// use rumtk_core::pipelines::pipeline_functions::{pipeline_create_command, pipeline_spawn_process, pipeline_get_stdout, pipeline_pipe_processes};
216 ///
217 /// let ls_name = "ls";
218 /// let mut ls_command = RUMCommand::default();
219 /// ls_command.path = RUMString::from(ls_name);
220 /// let mut sys_ls_command = pipeline_create_command(&ls_command);
221 ///
222 /// let wc_name = "wc";
223 /// let mut wc_command = RUMCommand::default();
224 /// wc_command.path = RUMString::from(wc_name);
225 /// let mut sys_wc_command = pipeline_create_command(&wc_command);
226 ///
227 /// let mut sys_ls_process = pipeline_spawn_process(&mut sys_ls_command).unwrap();
228 /// pipeline_pipe_processes(&mut sys_ls_process, &mut sys_wc_command).unwrap();
229 /// let mut sys_wc_process = pipeline_spawn_process(&mut sys_wc_command).unwrap();
230 ///
231 /// sys_ls_process.wait();
232 /// sys_wc_process.wait();
233 ///
234 /// let out_data = pipeline_get_stdout(sys_wc_process).unwrap();
235 ///
236 /// assert_eq!(out_data.is_empty(), false, "No output detected... {:?}", &out_data);
237 /// ```
238 ///
239 pub fn pipeline_get_stdout(mut process: RUMPipelineProcess) -> RUMResult<RUMBuffer> {
240 match process.wait_with_output() {
241 Ok(stdout) => Ok(RUMBuffer::from(stdout.stdout.clone())),
242 Err(e) => Err(rumtk_format!(
243 "Issue reading last process output because => {}",
244 e
245 )),
246 }
247 }
248
249 ///
250 /// Closes the `stdin` standard in file for process. Useful to trigger a resolution of the pipeline.
251 ///
252 /// ## Example
253 ///
254 /// ```
255 /// use rumtk_core::pipelines::pipeline_functions::pipeline_close_process_stdin;
256 /// use rumtk_core::strings::RUMString;
257 /// use rumtk_core::pipelines::pipeline_types::{RUMCommand, RUMPipelineCommand};
258 /// use rumtk_core::pipelines::pipeline_functions::{pipeline_create_command, pipeline_pipe_into_process, pipeline_spawn_process};
259 /// use rumtk_core::buffers::*;
260 ///
261 /// let ls_name = "ls";
262 /// let mut ls_command = RUMCommand::default();
263 /// ls_command.path = RUMString::from(ls_name);
264 /// let mut sys_ls_command = pipeline_create_command(&ls_command);
265 /// let mut sys_ls_process = pipeline_spawn_process(&mut sys_ls_command).unwrap();
266 ///
267 /// pipeline_close_process_stdin(&mut sys_ls_process);
268 ///
269 ///
270 /// ```
271 ///
272 pub fn pipeline_close_process_stdin(process: &mut RUMPipelineProcess) {
273 // Do not change into an expect() or such unwrap. We just want to ignore and assume stdin is closed.
274 match process.stdin.take() {
275 Some(stdin) => {
276 drop(stdin);
277 }
278 None => {}
279 };
280 }
281
282 ///
283 /// Pipe data into a process.
284 ///
285 /// ## Example
286 ///
287 /// ```
288 /// use rumtk_core::strings::RUMString;
289 /// use rumtk_core::pipelines::pipeline_types::{RUMCommand, RUMPipelineCommand};
290 /// use rumtk_core::pipelines::pipeline_functions::{pipeline_create_command, pipeline_pipe_into_process, pipeline_spawn_process};
291 /// use rumtk_core::buffers::*;
292 ///
293 /// let ls_name = "ls";
294 /// let mut ls_command = RUMCommand::default();
295 /// ls_command.path = RUMString::from(ls_name);
296 /// let mut sys_ls_command = pipeline_create_command(&ls_command);
297 /// let mut sys_ls_process = pipeline_spawn_process(&mut sys_ls_command).unwrap();
298 /// pipeline_pipe_into_process(&mut sys_ls_process, &RUMBuffer::default()).unwrap();
299 ///
300 /// let out = sys_ls_process.wait_with_output().unwrap();
301 ///
302 /// assert_eq!(out.stdout.is_empty(), false, "Piped command returned an empty buffer? => {:?}", String::from_utf8_lossy(out.stdout.as_slice()))
303 /// ```
304 ///
305 pub fn pipeline_pipe_into_process(
306 process: &mut RUMPipelineProcess,
307 data: &RUMBuffer,
308 ) -> RUMResult<()> {
309 match process.stdin.take() {
310 Some(ref mut stdin) => match stdin.write_all(data.as_slice()) {
311 Ok(_) => {}
312 Err(e) => {
313 return Err(rumtk_format!(
314 "Failed to pipe data to stdin of process {} because => {}",
315 process.id(),
316 e
317 ))
318 }
319 },
320 None => {}
321 }
322 Ok(())
323 }
324
325 ///
326 /// Builds an executable pipeline out of a list of [RUMCommand](RUMCommand).
327 ///
328 /// ## Example
329 ///
330 /// ```
331 /// use rumtk_core::strings::{RUMStringConversions, RUMString};
332 /// use rumtk_core::pipelines::pipeline_types::{RUMCommand};
333 /// use rumtk_core::pipelines::pipeline_functions::{pipeline_generate_command, pipeline_get_stdout, pipeline_wait_pipeline};
334 /// use rumtk_core::buffers::*;
335 ///
336 /// let ls_name = "ls";
337 /// let mut ls_command = RUMCommand::default();
338 /// ls_command.path = RUMString::from(ls_name);
339 /// ls_command.args = vec!["-la".to_string()];
340 ///
341 /// let mut process = pipeline_generate_command(&ls_command, &RUMBuffer::default()).unwrap();
342 /// let result = pipeline_get_stdout(process).unwrap();
343 ///
344 /// assert!(result.len() > 0, "Command generation failed to generate a valid command!");
345 /// ```
346 ///
347 pub fn pipeline_generate_command(command: &RUMCommand, data: &RUMBuffer) -> RUMResult<RUMPipelineProcess> {
348 let mut root = pipeline_create_command(&command);
349 let mut process = pipeline_spawn_process(&mut root)?;
350 pipeline_pipe_into_process(&mut process, data)?;
351 pipeline_close_process_stdin(&mut process);
352
353 Ok(process)
354 }
355
356 ///
357 /// Add buffer at the beginning of pipeline to pipe in. This buffer serves as the initial input of
358 /// a pipe aware program.
359 ///
360 /// ## Example
361 /// ```
362 /// use rumtk_core::base::RUMResult;
363 /// use rumtk_core::pipelines::pipeline_types::{RUMCommand, RUMCommandLine};
364 /// use rumtk_core::rumtk_pipeline_run;
365 /// use rumtk_core::strings::{RUMString};
366 /// use rumtk_core::buffers::{buffer_to_string};
367 /// use rumtk_core::buffers::*;
368 ///
369 /// let data = RUMBuffer::from(b"Hello World");
370 /// let wc_name = "wc";
371 /// let mut wc_command = RUMCommand::default();
372 /// wc_command.path = RUMString::from(wc_name);
373 /// let mut pipeline = vec![
374 /// wc_command
375 /// ];
376 ///
377 /// let processor = || -> RUMResult<RUMBuffer> {rumtk_pipeline_run!(&pipeline, &data)};
378 /// let result_string = buffer_to_string(&processor().unwrap()).unwrap();
379 /// let binding = result_string.as_str().replace('\n', "");
380 /// let result_items: Vec<&str> = binding.split(" ").collect();
381 /// let result = result_items.get(2).unwrap().trim().parse::<i32>().unwrap();
382 ///
383 /// assert_eq!(result, 2, "Data was not piped properly!");
384 /// ```
385 ///
386 pub fn pipeline_add_stdin_data_to_pipeline<'a>(pipeline: &'a mut RUMCommandLine, data: &RUMBuffer) -> &'a RUMCommandLine {
387 match pipeline.get_mut(0) {
388 Some(command) => command.data = Some(data.clone()),
389 None => {
390 return pipeline;
391 }
392 };
393
394 pipeline
395 }
396
397 ///
398 /// Flatten the [RUMCommandLine] structure into a single string representing the pipeline and
399 /// print it or log it.
400 ///
401 fn print_pipeline(pipeline: &RUMCommandLine) {
402 let mut pipeline_components = Vec::<RUMString>::with_capacity(pipeline.len());
403
404 for pipe in pipeline.iter() {
405 pipeline_components.push(rumtk_format!("{} {}", pipe.path, pipe.args.clone().join(" ")));
406 }
407
408 println!("Executing {}", pipeline_components.join(" | "));
409 }
410
411 pub fn pipeline_patch_command_args<'a>(cmd: &'a mut RUMCommand, replacements: &StringReplacementPair) -> RUMResult<&'a RUMCommand> {
412 let mut new_args = RUMCommandArgs::with_capacity(cmd.args.len());
413
414 for arg in cmd.args.iter() {
415 new_args.push(string_format(arg, replacements));
416 }
417
418 cmd.args = new_args;
419
420 Ok(cmd)
421 }
422
423 ///
424 /// Patches the arguments of the first command with the pattern=replacement pairs!
425 ///
426 /// ## Example
427 /// ```
428 /// use rumtk_core::base::RUMResult;
429 /// use rumtk_core::pipelines::pipeline_functions::pipeline_patch_args;
430 /// use rumtk_core::pipelines::pipeline_types::{RUMCommand, RUMCommandLine};
431 /// use rumtk_core::rumtk_pipeline_run;
432 /// use rumtk_core::strings::{RUMString};
433 /// use rumtk_core::buffers::{buffer_to_string};
434 /// use rumtk_core::buffers::*;
435 ///
436 /// let ls_name = "ls";
437 /// let mut ls_command = RUMCommand::default();
438 /// ls_command.path = RUMString::from(ls_name);
439 /// ls_command.args.push(RUMString::from("{options}"));
440 /// let mut pipeline = vec![
441 /// ls_command
442 /// ];
443 /// pipeline_patch_args(&mut pipeline, &[("{options}", "-la")]);
444 ///
445 /// let processor = || -> RUMResult<RUMBuffer> {rumtk_pipeline_run!(&pipeline)};
446 /// let result_string = buffer_to_string(&processor().unwrap()).unwrap();
447 /// let results: Vec<&str> = result_string.as_str().split("\n").collect();
448 /// let dot_dir = results.get(1).unwrap().chars().last().unwrap();
449 ///
450 /// assert_eq!(dot_dir, '.', "Incorrect options passed!");
451 /// ```
452 ///
453 pub fn pipeline_patch_args<'a>(pipeline: &'a mut RUMCommandLine, replacements: &StringReplacementPair) -> RUMResult<&'a RUMCommandLine> {
454 for mut cmd in pipeline.iter_mut() {
455 pipeline_patch_command_args(&mut cmd, replacements)?;
456 }
457 Ok(pipeline)
458 }
459
460 ///
461 /// Await for pipeline to execute in a async friendly manner. Once the pipeline execution ends,
462 /// consume the pipeline and return the output.
463 ///
464 /// ## Example
465 ///
466 /// ```
467 /// use rumtk_core::strings::RUMString;
468 /// use rumtk_core::buffers::*;
469 /// use rumtk_core::pipelines::pipeline_types::{RUMCommand};
470 /// use rumtk_core::pipelines::pipeline_functions::{pipeline_await_pipeline};
471 /// use rumtk_core::{rumtk_resolve_task};
472 ///
473 /// let ls_name = "ls";
474 /// let mut ls_command = RUMCommand::default();
475 /// ls_command.path = RUMString::from(ls_name);
476 ///
477 /// let wc_name = "wc";
478 /// let mut wc_command = RUMCommand::default();
479 /// wc_command.path = RUMString::from(wc_name);
480 ///
481 /// let pipeline = vec![
482 /// ls_command,
483 /// wc_command
484 /// ];
485 ///
486 /// let result = rumtk_resolve_task!(pipeline_await_pipeline(&pipeline, &RUMBuffer::default())).unwrap();
487 ///
488 /// assert_eq!(result.is_empty(), false, "Pipeline returned no buffer from command wc! => {:?}", &result);
489 /// ```
490 ///
491 pub async fn pipeline_await_pipeline(pipeline: &RUMCommandLine, initial_data: &RUMBuffer) -> RUMPipelineResult {
492 let pipeline_copy = pipeline.clone();
493 let data_copy = initial_data.clone();
494 rumtk_resolve_sync_task!(move || {
495 pipeline_wait_pipeline(&pipeline_copy, &data_copy)
496 })
497 }
498
499 ///
500 /// Await for pipeline to complete execution. Once the pipeline execution ends,
501 /// consume the pipeline and return the output.
502 ///
503 /// ## Example
504 ///
505 /// ```
506 /// use rumtk_core::strings::RUMString;
507 /// use rumtk_core::pipelines::pipeline_types::{RUMCommand};
508 /// use rumtk_core::pipelines::pipeline_functions::{pipeline_wait_pipeline};
509 /// use rumtk_core::buffers::*;
510 ///
511 /// let ls_name = "ls";
512 /// let mut ls_command = RUMCommand::default();
513 /// ls_command.path = RUMString::from(ls_name);
514 ///
515 /// let wc_name = "wc";
516 /// let mut wc_command = RUMCommand::default();
517 /// wc_command.path = RUMString::from(wc_name);
518 ///
519 /// let pipeline = vec![
520 /// ls_command,
521 /// wc_command
522 /// ];
523 ///
524 /// let result = pipeline_wait_pipeline(&pipeline, &RUMBuffer::default()).unwrap();
525 ///
526 /// assert_eq!(result.is_empty(), false, "Pipeline returned no buffer from command wc! => {:?}", &result);
527 /// ```
528 ///
529 pub fn pipeline_wait_pipeline(pipeline: &RUMCommandLine, initial_data: &RUMBuffer) -> RUMPipelineResult {
530 let mut last_data = initial_data.clone();
531
532 // Print pipeline
533 print_pipeline(&pipeline);
534
535 // Now let's visit each process and await their completion!
536 for c in pipeline.iter() {
537 let mut p = pipeline_generate_command(&c, &last_data)?;
538 pipeline_close_process_stdin(&mut p);
539 println!("Waiting on {}", p.id());
540 last_data = pipeline_get_stdout(p)?;
541 }
542
543 Ok(last_data)
544 }
545}
546
547pub mod pipeline_macros {
548 ///
549 /// Creates a pipeline command out of the provided parameters. Parameters include `path`, `args`,
550 /// and `env`. The command has [RUMCommand](super::pipeline_types::RUMCommand).
551 ///
552 /// `env` is a map of type [RUMCommandEnv](super::pipeline_types::RUMCommandEnv) containing a set of
553 /// key value pair strings that we can use to update the process environment.
554 ///
555 /// ## Example
556 ///
557 /// ### Program Only
558 ///
559 /// ```
560 /// use rumtk_core::rumtk_pipeline_command;
561 ///
562 /// let command = rumtk_pipeline_command!("ls");
563 /// ```
564 ///
565 /// ### Program with Piped Data
566 ///
567 /// ```
568 /// use rumtk_core::rumtk_pipeline_command;
569 /// use rumtk_core::buffers::*;
570 /// use rumtk_core::strings::RUMStringConversions;
571 ///
572 /// let command = rumtk_pipeline_command!("ls", RUMBuffer::default());
573 /// ```
574 ///
575 /// ### Program with Args
576 ///
577 /// ```
578 /// use rumtk_core::rumtk_pipeline_command;
579 /// use rumtk_core::buffers::*;
580 /// use rumtk_core::strings::RUMStringConversions;
581 ///
582 /// let command = rumtk_pipeline_command!("ls", RUMBuffer::default(), &vec![
583 /// "-l".to_string()
584 /// ]);
585 /// ```
586 ///
587 #[macro_export]
588 macro_rules! rumtk_pipeline_command {
589 ( $path:expr, $data:expr, $args:expr, $env:expr ) => {{
590 use $crate::pipelines::pipeline_types::RUMCommand;
591
592 RUMCommand::new($path, &Some($data), $args, $env)
593 }};
594 ( $path:expr, $data:expr, $args:expr ) => {{
595 use $crate::pipelines::pipeline_types::{RUMCommand, RUMCommandEnv};
596
597 RUMCommand::new($path, &Some($data), $args, &RUMCommandEnv::default())
598 }};
599 ( $path:expr, $data:expr ) => {{
600 use $crate::pipelines::pipeline_types::{RUMCommand, RUMCommandArgs, RUMCommandEnv};
601
602 RUMCommand::new(
603 $path,
604 &Some($data),
605 &RUMCommandArgs::default(),
606 &RUMCommandEnv::default(),
607 )
608 }};
609 ( $path:expr ) => {{
610 use $crate::pipelines::pipeline_types::{RUMCommand, RUMCommandArgs, RUMCommandEnv};
611 use $crate::buffers::*;
612
613 RUMCommand::new(
614 $path,
615 &None,
616 &RUMCommandArgs::default(),
617 &RUMCommandEnv::default(),
618 )
619 }};
620 }
621
622 ///
623 /// Given a series of [RUMCommand](super::pipeline_types::RUMCommand) passed to this macro, prepare
624 /// and execute the commands in a pipeline. A pipeline here refers to the Unix style pipeline which is the
625 /// terminal form of a pipeline. The pipeline behaves like it would in the terminal => `ls | wc`.
626 ///
627 /// See [this article](https://cscie26.dce.harvard.edu/~dce-lib113/reference/unix/unix2.html)
628 /// and the [Unix Philosophy](https://cscie2x.dce.harvard.edu/hw/ch01s06.html) to learn more!
629 ///
630 /// ## Example
631 ///
632 /// ### Simple
633 ///
634 /// ```
635 /// use rumtk_core::{rumtk_pipeline_command, rumtk_pipeline_quick_run, rumtk_resolve_task, rumtk_init_threads};
636 /// use rumtk_core::base::{RUMResult};
637 /// use rumtk_core::strings::RUMStringConversions;
638 /// use rumtk_core::buffers::*;
639 ///
640 /// let f = async || -> RUMResult<()> {
641 /// let result = rumtk_pipeline_quick_run!(
642 /// rumtk_pipeline_command!("ls"),
643 /// rumtk_pipeline_command!("wc")
644 /// ).unwrap();
645 ///
646 /// assert_eq!(result.is_empty(), false, "Pipeline returned no buffer from command wc! => {:?}", &result);
647 /// Ok(())
648 /// };
649 ///
650 /// rumtk_resolve_task!(f()).unwrap();
651 /// ```
652 ///
653 #[macro_export]
654 macro_rules! rumtk_pipeline_quick_run {
655 ( $($command:expr),+ ) => {{
656 use $crate::buffers::*;
657 use $crate::pipelines::pipeline_functions::{pipeline_wait_pipeline};
658
659 let pipeline = vec![$($command),+];
660
661 pipeline_wait_pipeline(&pipeline, &RUMBuffer::default())
662 }};
663 }
664
665 ///
666 /// Given a series of [RUMCommand](super::pipeline_types::RUMCommand) passed to this macro, prepare
667 /// and execute the commands in a pipeline. A pipeline here refers to the Unix style pipeline which is the
668 /// terminal form of a pipeline. The pipeline behaves like it would in the terminal => `ls | wc`.
669 ///
670 /// See [this article](https://cscie26.dce.harvard.edu/~dce-lib113/reference/unix/unix2.html)
671 /// and the [Unix Philosophy](https://cscie2x.dce.harvard.edu/hw/ch01s06.html) to learn more!
672 ///
673 /// This is the `async` flavor.
674 ///
675 /// ## Example
676 ///
677 /// ### Simple
678 ///
679 /// ```
680 /// use rumtk_core::{rumtk_pipeline_command, rumtk_pipeline_quick_run_async, rumtk_resolve_task, rumtk_init_threads};
681 /// use rumtk_core::base::{RUMResult};
682 /// use rumtk_core::strings::RUMStringConversions;
683 /// use rumtk_core::buffers::*;
684 ///
685 /// let f = async || -> RUMResult<()> {
686 /// let result = rumtk_pipeline_quick_run_async!(
687 /// rumtk_pipeline_command!("ls"),
688 /// rumtk_pipeline_command!("wc")
689 /// ).await?;
690 ///
691 /// assert_eq!(result.is_empty(), false, "Pipeline returned no buffer from command wc! => {:?}", &result);
692 /// Ok(())
693 /// };
694 ///
695 /// rumtk_resolve_task!(f()).unwrap();
696 /// ```
697 ///
698 #[macro_export]
699 macro_rules! rumtk_pipeline_quick_run_async {
700 ( $($command:expr),+ ) => {{
701 use $crate::buffers::*;
702 use $crate::pipelines::pipeline_functions::{pipeline_await_pipeline};
703
704 let pipeline = vec![$($command),+];
705
706 pipeline_await_pipeline(&pipeline.clone(), &RUMBuffer::default().clone())
707 }};
708 }
709
710 ///
711 /// This macro is similar to [rumtk_pipeline_quick_run](crate::rumtk_pipeline_quick_run). The difference here
712 /// is that the function takes a pipeline structure ([RUMCommandLine](crate::pipelines::pipeline_types::RUMCommandLine))
713 /// In other words, this macro simply runs an already defined pipeline.
714 ///
715 /// ## Example
716 ///
717 /// ### Run the pipeline
718 /// ```
719 /// use rumtk_core::{rumtk_pipeline_command, rumtk_pipeline_run, rumtk_resolve_task, rumtk_init_threads};
720 /// use rumtk_core::base::{RUMResult};
721 /// use rumtk_core::strings::RUMStringConversions;
722 /// use rumtk_core::buffers::*;
723 ///
724 /// let f = || -> RUMResult<RUMBuffer> {
725 /// let pipeline = vec![
726 /// rumtk_pipeline_command!("ls"),
727 /// rumtk_pipeline_command!("wc")
728 /// ];
729 ///
730 /// rumtk_pipeline_run!(&pipeline)
731 /// };
732 ///
733 /// f().unwrap();
734 /// ```
735 ///
736 /// ### Pipe Buffer to Pipeline
737 ///
738 /// ```
739 /// use rumtk_core::{rumtk_pipeline_command, rumtk_pipeline_run};
740 /// use rumtk_core::base::RUMResult;
741 /// use rumtk_core::buffers::*;
742 /// use rumtk_core::strings::{string_to_buffer};
743 /// use rumtk_core::buffers::{buffer_to_string};
744 ///
745 /// const data: &str = "Hello World!";
746 /// const expected: &str = " 0 2 12\n";
747 ///
748 ///
749 /// let f = |input: &RUMBuffer| -> RUMResult<RUMBuffer> {
750 /// let mut pipeline = vec![
751 /// rumtk_pipeline_command!("wc")
752 /// ];
753 ///
754 /// rumtk_pipeline_run!(&pipeline, &input)
755 /// };
756 /// let result = buffer_to_string(&f(&string_to_buffer(data)).unwrap()).unwrap();
757 ///
758 /// assert_eq!(result, expected, "Buffer correctly piped into pipeline!");
759 /// ```
760 ///
761 /// ### Pipe String to Pipeline
762 ///
763 /// ```
764 /// use rumtk_core::{rumtk_pipeline_command, rumtk_pipeline_run};
765 /// use rumtk_core::base::RUMResult;
766 /// use rumtk_core::strings::{string_to_buffer};
767 /// use rumtk_core::buffers::{buffer_to_string};
768 /// use rumtk_core::buffers::*;
769 ///
770 /// const data: &str = "Hello World!";
771 /// const expected: &str = " 0 2 12\n";
772 ///
773 ///
774 /// let f = |input: &str| -> RUMResult<RUMBuffer> {
775 /// let mut pipeline = vec![
776 /// rumtk_pipeline_command!("wc")
777 /// ];
778 ///
779 /// rumtk_pipeline_run!(&pipeline, &string_to_buffer(input))
780 /// };
781 /// let result = buffer_to_string(&f(data).unwrap()).unwrap();
782 ///
783 /// assert_eq!(result, expected, "String correctly piped into pipeline!");
784 /// ```
785 ///
786 #[macro_export]
787 macro_rules! rumtk_pipeline_run {
788 ( $pipeline:expr ) => {{
789 use $crate::buffers::*;
790
791 rumtk_pipeline_run!($pipeline, &RUMBuffer::default())
792 }};
793 ( $pipeline:expr, $data:expr ) => {{
794 use $crate::pipelines::pipeline_functions::{pipeline_wait_pipeline};
795
796 pipeline_wait_pipeline($pipeline, $data)
797 }};
798 }
799
800
801 ///
802 /// This macro is similar to [rumtk_pipeline_quick_run_async](crate::rumtk_pipeline_quick_run_async). The difference here
803 /// is that the function takes a pipeline structure ([RUMCommandLine](crate::pipelines::pipeline_types::RUMCommandLine))
804 /// In other words, this macro simply runs an already defined pipeline.
805 ///
806 /// ## Example
807 /// ```
808 /// use rumtk_core::{rumtk_pipeline_command, rumtk_pipeline_run_async, rumtk_resolve_task, rumtk_init_threads};
809 /// use rumtk_core::base::{RUMResult};
810 /// use rumtk_core::strings::RUMStringConversions;
811 /// use rumtk_core::buffers::*;
812 ///
813 /// let f = async || -> RUMResult<RUMBuffer> {
814 /// let pipeline = vec![
815 /// rumtk_pipeline_command!("ls"),
816 /// rumtk_pipeline_command!("wc")
817 /// ];
818 ///
819 /// rumtk_pipeline_run_async!(&pipeline).await
820 /// };
821 ///
822 /// rumtk_resolve_task!(f()).unwrap();
823 /// ```
824 ///
825 /// ### Pipe Buffer to Pipeline
826 ///
827 /// ```
828 /// use rumtk_core::{rumtk_pipeline_command, rumtk_pipeline_run_async, rumtk_resolve_task, rumtk_spawn_task};
829 /// use rumtk_core::base::RUMResult;
830 /// use rumtk_core::buffers::*;
831 /// use rumtk_core::buffers::{buffer_to_string};
832 /// use rumtk_core::strings::{string_to_buffer};
833 ///
834 /// const data: &str = "Hello World!";
835 /// const expected: &str = " 0 2 12\n";
836 ///
837 ///
838 /// let task = rumtk_spawn_task!(async {
839 /// let input = RUMBuffer::from("Hello World!");
840 /// let mut pipeline = vec![
841 /// rumtk_pipeline_command!("wc")
842 /// ];
843 ///
844 /// rumtk_pipeline_run_async!(&pipeline, &input).await
845 /// });
846 ///
847 /// let result = buffer_to_string(&rumtk_resolve_task!(task).unwrap().unwrap()).unwrap();
848 ///
849 /// assert_eq!(result, expected, "Buffer correctly piped into pipeline!");
850 /// ```
851 ///
852 /// ### Pipe String to Pipeline
853 ///
854 /// ```
855 /// use rumtk_core::{rumtk_pipeline_command, rumtk_pipeline_run_async, rumtk_resolve_task, rumtk_spawn_task};
856 /// use rumtk_core::base::RUMResult;
857 /// use rumtk_core::strings::{string_to_buffer};
858 /// use rumtk_core::buffers::{buffer_to_string};
859 /// use rumtk_core::buffers::*;
860 ///
861 /// const data: &str = "Hello World!";
862 /// const expected: &str = " 0 2 12\n";
863 ///
864 ///
865 /// let task = rumtk_spawn_task!(async {
866 /// let input = string_to_buffer(data);
867 /// let mut pipeline = vec![
868 /// rumtk_pipeline_command!("wc")
869 /// ];
870 ///
871 /// rumtk_pipeline_run_async!(&pipeline, &input).await
872 /// });
873 ///
874 /// let result = buffer_to_string(&rumtk_resolve_task!(task).unwrap().unwrap()).unwrap();
875 ///
876 /// assert_eq!(result, expected, "String correctly piped into pipeline!");
877 /// ```
878 ///
879 #[macro_export]
880 macro_rules! rumtk_pipeline_run_async {
881 ( $pipeline:expr ) => {{
882 use $crate::buffers::*;
883
884 rumtk_pipeline_run_async!($pipeline, &RUMBuffer::default())
885 }};
886 ( $pipeline:expr, $data:expr ) => {{
887 use $crate::pipelines::pipeline_functions::{pipeline_await_pipeline};
888
889 pipeline_await_pipeline($pipeline, $data)
890 }};
891 }
892
893 ///
894 /// Patch the pipeline's arguments with desired dynamic options.
895 ///
896 /// ## Example
897 /// ```
898 /// use rumtk_core::{rumtk_pipeline_patch_args, rumtk_pipeline_command, rumtk_pipeline_run};
899 /// use rumtk_core::base::RUMResult;
900 /// use rumtk_core::buffers::{buffer_to_string};
901 /// use rumtk_core::buffers::*;
902 /// use rumtk_core::strings::string_to_buffer;
903 ///
904 ///
905 /// let f = || -> RUMResult<RUMBuffer> {
906 /// let mut pipeline = vec![
907 /// rumtk_pipeline_command!("ls", RUMBuffer::default(), &vec![
908 /// "{options}".into()
909 /// ])
910 /// ];
911 ///
912 /// rumtk_pipeline_patch_args!(&mut pipeline, &[("{options}", "-la")]);
913 ///
914 /// rumtk_pipeline_run!(&pipeline)
915 /// };
916 ///
917 /// let result_string = buffer_to_string(&f().unwrap()).unwrap();
918 /// let results: Vec<&str> = result_string.as_str().split("\n").collect();
919 /// let dot_dir = results.get(1).unwrap().chars().last().unwrap();
920 ///
921 /// assert_eq!(dot_dir, '.', "Incorrect options passed!");
922 /// ```
923 ///
924 #[macro_export]
925 macro_rules! rumtk_pipeline_patch_args {
926 ( $pipeline:expr, $replacements:expr ) => {{
927 use $crate::pipelines::pipeline_functions::{pipeline_patch_args};
928
929 pipeline_patch_args($pipeline, $replacements)
930 }};
931 }
932}