fetcher 0.15.3

Data automation and pipelining framework
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117

/*
 * This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at https://mozilla.org/MPL/2.0/.
 */

//! This module contains [`Exec`] source and sink. It is re-exported in the [`crate::sinks`] and [`crate::sources`] modules

use std::{io, process::Stdio, string::FromUtf8Error};
use tokio::{io::AsyncWriteExt, process::Command};

use crate::{
	entry::Entry,
	sinks::{
		Sink,
		message::{Message, MessageId},
	},
	sources::Fetch,
};

#[cfg(not(target_os = "windows"))]
const SHELL: &str = "sh";
#[cfg(target_os = "windows")]
const SHELL: &str = "cmd";

#[cfg(not(target_os = "windows"))]
const SHELL_RUN_ARG: &str = r"\C";
#[cfg(target_os = "windows")]
const SHELL: &str = "-c";

/// Exec source. It can execute a shell command and source its stdout
#[derive(Debug)]
pub struct Exec {
	/// The command to execute
	pub cmd: String,
}
/// Errors that happened while executing a process
#[expect(missing_docs, reason = "error message is self-documenting")]
#[derive(thiserror::Error, Debug)]
pub enum ExecError {
	#[error("Bad command")]
	BadCommand(#[source] io::Error),

	#[error("Command output is not valid UTF-8")]
	BadUtf8(#[from] FromUtf8Error),

	#[error("Can't start the process")]
	CantStart(#[source] io::Error),

	#[error("Can't pass data to the stdin of the process")]
	CantWriteStdin(#[source] io::Error),
}

impl Fetch for Exec {
	type Err = ExecError;

	async fn fetch(&mut self) -> Result<Vec<Entry>, Self::Err> {
		tracing::debug!("Spawning a shell with command {:?}", self.cmd);
		let out = Command::new(SHELL)
			.arg(SHELL_RUN_ARG)
			.arg(&self.cmd)
			.output()
			.await
			.map_err(ExecError::BadCommand)?
			.stdout;

		let out = String::from_utf8(out).map_err(ExecError::BadUtf8)?;
		tracing::debug!("Got {out:?} from the command");

		let entry = Entry::builder().raw_contents(out).build();

		Ok(vec![entry])
	}
}

impl Sink for Exec {
	type Err = ExecError;

	/// Passes message's body to the stdin of the process. The tag parameter is ignored
	///
	/// # Errors
	/// * if the process couldn't be started
	/// * if the data couldn't be passed to the stdin pipe of the process
	async fn send(
		&mut self,
		message: &Message,
		_reply_to: Option<&MessageId>,
		_tag: Option<&str>,
	) -> Result<Option<MessageId>, Self::Err> {
		let Some(body) = &message.body else {
			return Ok(None);
		};

		tracing::debug!("Spawning process {:?}", self.cmd);
		let mut shell = Command::new(SHELL)
			.arg(SHELL_RUN_ARG)
			.arg(&self.cmd)
			.stdin(Stdio::piped())
			.stdout(Stdio::null())
			.spawn()
			.map_err(ExecError::CantStart)?;

		if let Some(stdin) = &mut shell.stdin {
			tracing::debug!("Writing {body:?} to stdin of the process");
			stdin
				.write_all(body.as_bytes())
				.await
				.map_err(ExecError::CantWriteStdin)?;
		}

		tracing::trace!("Waiting for the process to exit");
		shell.wait().await.map_err(ExecError::CantStart)?;
		tracing::trace!("Process successfully exited");

		Ok(None)
	}
}