use std::collections::HashMap;
use std::io::{BufRead, BufReader, BufWriter, Read, Write};
use std::path::PathBuf;
use std::time::Duration;
use orfail::OrFail;
use crate::json::JsonObject;
use crate::media::{MediaSample, MediaStreamId, MediaStreamName, MediaStreamNameRegistry};
use crate::metadata::SourceId;
use crate::processor::{
MediaProcessor, MediaProcessorInput, MediaProcessorOutput, MediaProcessorSpec,
MediaProcessorWorkloadHint,
};
use crate::stats::ProcessorStats;
use crate::types::EvenUsize;
#[derive(Debug, Clone)]
pub struct PluginCommand {
pub command: PathBuf,
pub args: Vec<String>,
pub input_stream_names: Vec<MediaStreamName>,
pub output_stream_names: Vec<MediaStreamName>,
}
impl PluginCommand {
pub fn start(
&self,
registry: &mut MediaStreamNameRegistry,
) -> orfail::Result<PluginCommandProcessor> {
let mut input_stream_ids = Vec::new();
for name in &self.input_stream_names {
input_stream_ids.push(registry.get_id(name).or_fail()?);
}
let mut output_stream_ids = HashMap::new();
for name in &self.output_stream_names {
output_stream_ids.insert(
name.clone(),
registry.register_name(name.clone()).or_fail()?,
);
}
let mut process = std::process::Command::new(&self.command)
.args(&self.args)
.stdin(std::process::Stdio::piped())
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::inherit())
.spawn()
.or_fail_with(|e| format!("failed to start plugin command: {e}"))?;
let stdin = process
.stdin
.take()
.or_fail_with(|()| "failed to get stdin handle".to_owned())?;
let stdout = process
.stdout
.take()
.or_fail_with(|()| "failed to get stdout handle".to_owned())?;
Ok(PluginCommandProcessor {
process,
stdin: BufWriter::new(stdin),
stdout: BufReader::new(stdout),
input_stream_ids,
next_request_id: 0,
output_stream_ids,
})
}
}
impl<'text, 'raw> TryFrom<nojson::RawJsonValue<'text, 'raw>> for PluginCommand {
type Error = nojson::JsonParseError;
fn try_from(value: nojson::RawJsonValue<'text, 'raw>) -> Result<Self, Self::Error> {
let obj = JsonObject::new(value)?;
Ok(Self {
command: obj.get_required("command")?,
args: obj.get("args")?.unwrap_or_default(),
input_stream_names: obj.get("input_stream")?.unwrap_or_default(),
output_stream_names: obj.get("output_stream")?.unwrap_or_default(),
})
}
}
#[derive(Debug)]
pub struct PluginCommandProcessor {
process: std::process::Child,
stdin: BufWriter<std::process::ChildStdin>,
stdout: BufReader<std::process::ChildStdout>,
input_stream_ids: Vec<MediaStreamId>,
output_stream_ids: HashMap<MediaStreamName, MediaStreamId>,
next_request_id: u64,
}
impl PluginCommandProcessor {
fn cast<T>(
&mut self,
notification: &JsonRpcRequest<T>,
payload: Option<&[u8]>,
) -> orfail::Result<()>
where
T: nojson::DisplayJson,
{
let notification = nojson::Json(notification).to_string();
writeln!(self.stdin, "Content-Length: {}", notification.len()).or_fail()?;
writeln!(self.stdin, "Content-Type: application/json").or_fail()?;
writeln!(self.stdin).or_fail()?;
write!(self.stdin, "{notification}").or_fail()?;
if let Some(payload) = payload {
writeln!(self.stdin, "Content-Length: {}", payload.len()).or_fail()?;
writeln!(self.stdin, "Content-Type: application/octet-stream").or_fail()?;
writeln!(self.stdin).or_fail()?;
self.stdin.write_all(payload).or_fail()?;
}
self.stdin.flush().or_fail()?;
Ok(())
}
fn call<T, U>(&mut self, request: &JsonRpcRequest<T>) -> orfail::Result<U>
where
T: nojson::DisplayJson,
U: for<'text, 'raw> TryFrom<nojson::RawJsonValue<'text, 'raw>>,
{
let request = nojson::Json(request).to_string();
writeln!(self.stdin, "Content-Length: {}", request.len()).or_fail()?;
writeln!(self.stdin, "Content-Type: application/json").or_fail()?;
writeln!(self.stdin).or_fail()?;
write!(self.stdin, "{request}").or_fail()?;
self.stdin.flush().or_fail()?;
let mut content_length = None;
let mut line = String::new();
loop {
line.clear();
self.stdout.read_line(&mut line).or_fail()?;
if line.trim().is_empty() {
break;
}
if let Some(header_value) = line.strip_prefix("Content-Length: ") {
content_length = Some(
header_value
.trim()
.parse::<usize>()
.or_fail_with(|e| format!("invalid content length: {e}"))?,
);
}
}
let content_length = content_length
.or_fail_with(|()| "missing Content-Length header in response".to_owned())?;
let mut response_buffer = vec![0u8; content_length];
self.stdout.read_exact(&mut response_buffer).or_fail()?;
let response_text = std::str::from_utf8(&response_buffer)
.or_fail_with(|e| format!("invalid UTF-8 in response: {e}"))?;
let json = nojson::RawJson::parse(response_text)
.or_fail_with(|e| format!("failed to parse JSON response: {e}"))?;
if let Some(error) = json.value().to_member("error").or_fail()?.get() {
return Err(orfail::Failure::new(format!("JSON-RPC error: {error}",)));
}
let result = json
.value()
.to_member("result")
.or_fail()?
.required()
.or_fail()?;
U::try_from(result).map_err(|_| {
orfail::Failure::new("failed to convert response to expected type".to_owned())
})
}
fn read_payload(&mut self) -> orfail::Result<Vec<u8>> {
let mut content_length = None;
let mut line = String::new();
loop {
line.clear();
self.stdout.read_line(&mut line).or_fail()?;
if line.trim().is_empty() {
break;
}
if let Some(header_value) = line.strip_prefix("Content-Length: ") {
content_length = Some(
header_value
.trim()
.parse::<usize>()
.or_fail_with(|e| format!("invalid content length: {e}"))?,
);
}
}
let content_length = content_length
.or_fail_with(|()| "missing Content-Length header for payload".to_owned())?;
let mut payload_data = vec![0u8; content_length];
self.stdout.read_exact(&mut payload_data).or_fail()?;
Ok(payload_data)
}
}
impl MediaProcessor for PluginCommandProcessor {
fn spec(&self) -> MediaProcessorSpec {
MediaProcessorSpec {
input_stream_ids: self.input_stream_ids.clone(),
output_stream_ids: self.output_stream_ids.values().copied().collect(),
stats: ProcessorStats::other("plugin_command"),
workload_hint: MediaProcessorWorkloadHint::PLUGIN,
}
}
fn process_input(&mut self, input: MediaProcessorInput) -> orfail::Result<()> {
match input.sample {
None => {
self.input_stream_ids.retain(|id| *id != input.stream_id);
let req = JsonRpcRequest::notification(
"notify_eos",
nojson::object(|f| f.member("stream_id", input.stream_id)),
);
self.cast(&req, None).or_fail()?;
}
Some(MediaSample::Audio(data)) => {
(data.format == crate::audio::AudioFormat::I16Be).or_fail()?;
let req = JsonRpcRequest::notification(
"notify_audio",
nojson::object(|f| {
f.member("stream_id", input.stream_id)?;
f.member("stereo", data.stereo)?;
f.member("sample_rate", data.sample_rate)?;
f.member("timestamp_us", data.timestamp.as_micros())?;
f.member("duration_us", data.duration.as_micros())?;
Ok(())
}),
);
self.cast(&req, Some(&data.data)).or_fail()?;
}
Some(MediaSample::Video(frame)) => {
let req = JsonRpcRequest::notification(
"notify_video",
nojson::object(|f| {
f.member("stream_id", input.stream_id)?;
f.member("width", frame.width)?;
f.member("height", frame.height)?;
f.member("timestamp_us", frame.timestamp.as_micros())?;
f.member("duration_us", frame.duration.as_micros())?;
Ok(())
}),
);
let bgr_data = frame.to_bgr_data().or_fail()?;
self.cast(&req, Some(&bgr_data)).or_fail()?;
}
}
Ok(())
}
fn process_output(&mut self) -> orfail::Result<MediaProcessorOutput> {
let id = self.next_request_id;
self.next_request_id += 1;
let req = JsonRpcRequest::request("poll_output", id, ());
let res: PollOutputResponse = self.call(&req).or_fail()?;
let output = match res {
PollOutputResponse::WaitingInputAny => MediaProcessorOutput::awaiting_any(),
PollOutputResponse::WaitingInput { stream_id } => {
MediaProcessorOutput::pending(stream_id)
}
PollOutputResponse::AudioData {
stream_name,
stereo,
sample_rate,
timestamp,
duration,
} => {
let stream_id = self
.output_stream_ids
.get(&stream_name)
.copied()
.or_fail()?;
let audio_data = self.read_payload().or_fail()?;
let mut audio_sample = crate::audio::AudioData {
source_id: None,
data: audio_data,
format: crate::audio::AudioFormat::I16Be,
stereo,
sample_rate: sample_rate as u16,
timestamp,
duration,
sample_entry: None,
};
audio_sample.source_id = Some(SourceId::new(stream_name.get()));
MediaProcessorOutput::audio_data(stream_id, audio_sample)
}
PollOutputResponse::VideoFrame {
stream_name,
width,
height,
timestamp,
duration,
} => {
let stream_id = self
.output_stream_ids
.get(&stream_name)
.copied()
.or_fail()?;
let frame_data = self.read_payload().or_fail()?;
let mut video_frame = crate::video::VideoFrame::from_bgr_data(
&frame_data,
width,
height,
timestamp,
duration,
)
.or_fail()?;
video_frame.source_id = Some(SourceId::new(stream_name.get()));
MediaProcessorOutput::video_frame(stream_id, video_frame)
}
PollOutputResponse::Finished => MediaProcessorOutput::Finished,
};
Ok(output)
}
}
impl Drop for PluginCommandProcessor {
fn drop(&mut self) {
let _ = self.process.kill();
let _ = self.process.wait();
}
}
#[derive(Debug)]
pub struct JsonRpcRequest<'a, T> {
method: &'a str,
id: Option<u64>,
params: T,
}
impl<'a, T> JsonRpcRequest<'a, T> {
pub fn notification(method: &'a str, params: T) -> Self {
Self {
method,
id: None,
params,
}
}
pub fn request(method: &'a str, id: u64, params: T) -> Self {
Self {
method,
id: Some(id),
params,
}
}
}
impl<'a, T> nojson::DisplayJson for JsonRpcRequest<'a, T>
where
T: nojson::DisplayJson,
{
fn fmt(&self, f: &mut nojson::JsonFormatter<'_, '_>) -> std::fmt::Result {
f.object(|f| {
f.member("jsonrpc", "2.0")?;
f.member("method", self.method)?;
if let Some(id) = self.id {
f.member("id", id)?;
}
f.member("params", &self.params)?;
Ok(())
})
}
}
#[derive(Debug)]
pub enum PollOutputResponse {
WaitingInputAny,
WaitingInput {
stream_id: MediaStreamId,
},
AudioData {
stream_name: MediaStreamName,
stereo: bool,
sample_rate: u32,
timestamp: Duration,
duration: Duration,
},
VideoFrame {
stream_name: MediaStreamName,
width: EvenUsize,
height: EvenUsize,
timestamp: Duration,
duration: Duration,
},
Finished,
}
impl<'text, 'raw> TryFrom<nojson::RawJsonValue<'text, 'raw>> for PollOutputResponse {
type Error = nojson::JsonParseError;
fn try_from(value: nojson::RawJsonValue<'text, 'raw>) -> Result<Self, Self::Error> {
let obj = JsonObject::new(value)?;
let response_type: String = obj.get_required("type")?;
match response_type.as_str() {
"waiting_input_any" => Ok(Self::WaitingInputAny),
"waiting_input" => {
let stream_id = obj.get_required("stream_id")?;
Ok(Self::WaitingInput { stream_id })
}
"audio_data" => {
let stream_name = obj.get_required("stream_name")?;
let stereo = obj.get_required("stereo")?;
let sample_rate = obj.get_required("sample_rate")?;
let timestamp_us: u64 = obj.get_required("timestamp_us")?;
let duration_us: u64 = obj.get_required("duration_us")?;
Ok(Self::AudioData {
stream_name,
stereo,
sample_rate,
timestamp: Duration::from_micros(timestamp_us),
duration: Duration::from_micros(duration_us),
})
}
"video_frame" => {
let stream_name = obj.get_required("stream_name")?;
let width_raw: u32 = obj.get_required("width")?;
let height_raw: u32 = obj.get_required("height")?;
let timestamp_us: u64 = obj.get_required("timestamp_us")?;
let duration_us: u64 = obj.get_required("duration_us")?;
let width = EvenUsize::new(width_raw as usize)
.ok_or_else(|| value.invalid("width must be even"))?;
let height = EvenUsize::new(height_raw as usize)
.ok_or_else(|| value.invalid("height must be even"))?;
Ok(Self::VideoFrame {
stream_name,
width,
height,
timestamp: Duration::from_micros(timestamp_us),
duration: Duration::from_micros(duration_us),
})
}
"finished" => Ok(Self::Finished),
unknown => {
Err(value.invalid(format!("unknown poll output response type: {unknown:?}")))
}
}
}
}