cyberrex/mirakurun-rs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

First commit

Browse Source
This commit is contained in:
CyberRex
2026-07-31 14:39:33 +09:00
commit 38670efd46
103 changed files with 22514 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
crates/mirakurun-core/Cargo.toml Normal file
View File

@@ -0,0 +1,28 @@
[package]
name = "mirakurun-core"
version.workspace = true
edition.workspace = true
rust-version.workspace = true
license.workspace = true
[dependencies]
base64.workspace = true
bytes.workspace = true
chrono.workspace = true
encoding_rs.workspace = true
mirakurun-types = { path = "../mirakurun-types" }
nix.workspace = true
serde.workspace = true
serde_json.workspace = true
serde_yaml_ng.workspace = true
sha2.workspace = true
tempfile.workspace = true
thiserror.workspace = true
tokio.workspace = true
tracing.workspace = true
[dev-dependencies]
tempfile.workspace = true
[lints]
workspace = true

229
crates/mirakurun-core/src/config.rs Normal file
View File

@@ -0,0 +1,229 @@
// Copyright 2026 Mirakurun contributors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
use std::{
env,
path::{Path, PathBuf},
};
use mirakurun_types::{
ConfigChannel, ConfigServer, ConfigTuner, default_ipv4_ranges, default_ipv6_ranges,
default_origins,
};
use serde::{Serialize, de::DeserializeOwned};
use crate::{
Result,
error::CoreError,
persistence::{atomic_write, ensure_parent},
};
pub const DEFAULT_SERVER_CONFIG_PATH: &str = "/usr/local/etc/mirakurun/server.yml";
pub const DEFAULT_TUNERS_CONFIG_PATH: &str = "/usr/local/etc/mirakurun/tuners.yml";
pub const DEFAULT_CHANNELS_CONFIG_PATH: &str = "/usr/local/etc/mirakurun/channels.yml";
pub const DEFAULT_SERVICES_DB_PATH: &str = "/usr/local/var/db/mirakurun/services.json";
pub const DEFAULT_PROGRAMS_DB_PATH: &str = "/usr/local/var/db/mirakurun/programs.json";
pub const DEFAULT_LOGO_DATA_DIR_PATH: &str = "/usr/local/var/db/mirakurun/logo-data";
const DEFAULT_SERVER_YAML: &str = "logLevel: 2\npath: /var/run/mirakurun.sock\nport: 40772\n";
const DEFAULT_LIST_YAML: &str = "[]\n";
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ConfigPaths {
pub server: PathBuf,
pub tuners: PathBuf,
pub channels: PathBuf,
pub services_db: PathBuf,
pub programs_db: PathBuf,
pub logo_data_dir: PathBuf,
}
impl Default for ConfigPaths {
fn default() -> Self {
Self::from_environment()
}
}
impl ConfigPaths {
#[must_use]
pub fn from_environment() -> Self {
Self {
server: env_path("SERVER_CONFIG_PATH", DEFAULT_SERVER_CONFIG_PATH),
tuners: env_path("TUNERS_CONFIG_PATH", DEFAULT_TUNERS_CONFIG_PATH),
channels: env_path("CHANNELS_CONFIG_PATH", DEFAULT_CHANNELS_CONFIG_PATH),
services_db: env_path("SERVICES_DB_PATH", DEFAULT_SERVICES_DB_PATH),
programs_db: env_path("PROGRAMS_DB_PATH", DEFAULT_PROGRAMS_DB_PATH),
logo_data_dir: env_path("LOGO_DATA_DIR_PATH", DEFAULT_LOGO_DATA_DIR_PATH),
}
}
}
#[derive(Debug, Clone)]
pub struct LoadedConfig {
pub server: ConfigServer,
pub tuners: Vec<ConfigTuner>,
pub channels: Vec<ConfigChannel>,
}
impl LoadedConfig {
/// Loads all Mirakurun configuration files, creating defaults when absent.
///
/// # Errors
///
/// Returns an error when a file cannot be created/read, YAML is invalid,
/// or a validated server setting is outside its accepted range.
pub async fn load(paths: &ConfigPaths) -> Result<Self> {
let mut server: ConfigServer =
load_or_create_yaml(&paths.server, DEFAULT_SERVER_YAML, "server config").await?;
apply_server_defaults(&mut server);
validate_server(&server)?;
let tuners = load_or_create_yaml(&paths.tuners, DEFAULT_LIST_YAML, "tuners config").await?;
let channels =
load_or_create_yaml(&paths.channels, DEFAULT_LIST_YAML, "channels config").await?;
Ok(Self {
server,
tuners,
channels,
})
}
}
/// Saves the server configuration atomically.
///
/// # Errors
///
/// Returns an error when serialization or filesystem operations fail.
pub async fn save_server(path: &Path, config: &ConfigServer) -> Result<()> {
save_yaml(path, config, "server config").await
}
/// Saves the tuner configuration atomically.
///
/// # Errors
///
/// Returns an error when serialization or filesystem operations fail.
pub async fn save_tuners(path: &Path, config: &[ConfigTuner]) -> Result<()> {
save_yaml(path, config, "tuners config").await
}
/// Saves the channel configuration atomically.
///
/// # Errors
///
/// Returns an error when serialization or filesystem operations fail.
pub async fn save_channels(path: &Path, config: &[ConfigChannel]) -> Result<()> {
save_yaml(path, config, "channels config").await
}
async fn load_or_create_yaml<T>(
path: &Path,
default_contents: &str,
kind: &'static str,
) -> Result<T>
where
T: DeserializeOwned,
{
if !path.exists() {
ensure_parent(path, kind).await?;
atomic_write(path, default_contents.as_bytes(), kind).await?;
}
let contents = tokio::fs::read_to_string(path)
.await
.map_err(|source| CoreError::Read {
kind,
path: path.to_path_buf(),
source,
})?;
serde_yaml_ng::from_str(&contents).map_err(|source| CoreError::Yaml {
path: path.to_path_buf(),
source,
})
}
async fn save_yaml<T>(path: &Path, value: &T, kind: &'static str) -> Result<()>
where
T: Serialize + ?Sized,
{
let contents = serde_yaml_ng::to_string(value)?;
ensure_parent(path, kind).await?;
atomic_write(path, contents.as_bytes(), kind).await
}
fn apply_server_defaults(config: &mut ConfigServer) {
if config.allow_ipv4_cidr_ranges.is_empty() {
config.allow_ipv4_cidr_ranges = default_ipv4_ranges();
}
if config.allow_ipv6_cidr_ranges.is_empty() {
config.allow_ipv6_cidr_ranges = default_ipv6_ranges();
}
if config.allow_origins.is_empty() {
config.allow_origins = default_origins();
}
if config.hostname.as_deref().is_none_or(str::is_empty) {
config.hostname = detect_hostname();
}
}
fn validate_server(config: &ConfigServer) -> Result<()> {
if let Some(value) = config.job_max_running {
if !(1..=100).contains(&value) {
return Err(CoreError::InvalidConfig(
"jobMaxRunning must be between 1 and 100".into(),
));
}
}
if let Some(value) = config.job_max_standby {
if !(1..=100).contains(&value) {
return Err(CoreError::InvalidConfig(
"jobMaxStandby must be between 1 and 100".into(),
));
}
}
Ok(())
}
fn env_path(name: &str, default: &str) -> PathBuf {
env::var_os(name).map_or_else(|| PathBuf::from(default), PathBuf::from)
}
fn detect_hostname() -> Option<String> {
env::var("HOSTNAME")
.ok()
.filter(|value| !value.trim().is_empty())
.or_else(|| {
std::fs::read_to_string("/etc/hostname")
.ok()
.map(|value| value.trim().to_owned())
.filter(|value| !value.is_empty())
})
}
#[cfg(test)]
mod tests {
use tempfile::tempdir;
use super::{ConfigPaths, LoadedConfig};
#[tokio::test]
async fn creates_and_loads_missing_configuration() {
let directory = tempdir().expect("create tempdir");
let paths = ConfigPaths {
server: directory.path().join("server.yml"),
tuners: directory.path().join("tuners.yml"),
channels: directory.path().join("channels.yml"),
services_db: directory.path().join("services.json"),
programs_db: directory.path().join("programs.json"),
logo_data_dir: directory.path().join("logo-data"),
};
let loaded = LoadedConfig::load(&paths).await.expect("load config");
assert_eq!(loaded.server.port, Some(40_772));
assert!(loaded.tuners.is_empty());
assert!(loaded.channels.is_empty());
}
}

509
crates/mirakurun-core/src/epg.rs Normal file
View File

@@ -0,0 +1,509 @@
// Copyright 2026 Mirakurun contributors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
use std::collections::{BTreeMap, HashMap, HashSet};
use encoding_rs::EUC_JP;
use mirakurun_types::{
Program, ProgramGenre, ProgramVideo, ProgramVideoResolution, ProgramVideoType, program_id,
};
use crate::{
filter::SectionAssembler,
ts::{Packet, PacketFramer, crc32_mpeg2},
};
const PID_EIT: u16 = 0x0012;
#[derive(Debug, Default)]
pub struct EitCollector {
framer: PacketFramer,
assembler: SectionAssembler,
programs: HashMap<u64, Program>,
sections_seen: HashSet<(u8, u16, u8, u8)>,
packet_count: u64,
section_count: u64,
}
impl EitCollector {
pub fn push(&mut self, chunk: &[u8]) {
let mut packets = Vec::new();
self.framer.push(chunk, |packet| packets.push(*packet));
for packet_bytes in packets {
self.packet_count = self.packet_count.saturating_add(1);
let Ok(packet) = Packet::new(&packet_bytes) else {
continue;
};
if packet.pid() != PID_EIT {
continue;
}
for section in self.assembler.push(packet) {
if parse_eit_section(&section, &mut self.programs)
&& self.sections_seen.insert((
section[0],
u16::from_be_bytes([section[3], section[4]]),
(section[5] >> 1) & 0x1f,
section[6],
))
{
self.section_count = self.section_count.saturating_add(1);
}
}
}
}
#[must_use]
pub const fn packet_count(&self) -> u64 {
self.packet_count
}
#[must_use]
pub const fn section_count(&self) -> u64 {
self.section_count
}
#[must_use]
pub fn into_programs(self) -> Vec<Program> {
let mut programs = self.programs.into_values().collect::<Vec<_>>();
programs.sort_by_key(|program| (program.start_at, program.id));
programs
}
}
fn parse_eit_section(section: &[u8], programs: &mut HashMap<u64, Program>) -> bool {
if section.len() < 18
|| !matches!(section[0], 0x4e..=0x6f)
|| section[5] & 0x01 == 0
|| crc32_mpeg2(section) != 0
{
return false;
}
let is_present_following = matches!(section[0], 0x4e | 0x4f);
if is_present_following && section[6] > 1 {
return false;
}
let service_id = u16::from_be_bytes([section[3], section[4]]);
let network_id = u16::from_be_bytes([section[10], section[11]]);
let Some(end) = section.len().checked_sub(4) else {
return false;
};
let mut offset = 14;
while offset + 12 <= end {
let event_id = u16::from_be_bytes([section[offset], section[offset + 1]]);
let start = &section[offset + 2..offset + 7];
let duration = &section[offset + 7..offset + 10];
let descriptor_length =
(usize::from(section[offset + 10] & 0x0f) << 8) | usize::from(section[offset + 11]);
let Some(descriptor_end) = offset
.checked_add(12)
.and_then(|value| value.checked_add(descriptor_length))
else {
return false;
};
if descriptor_end > end {
return false;
}
if let Some(start_at) = decode_mjd_time(start) {
let id = program_id(network_id, service_id, event_id);
let program = programs.entry(id).or_insert_with(|| Program {
id,
event_id,
service_id,
network_id,
start_at,
duration: decode_bcd_duration(duration).unwrap_or(1),
is_free: section[offset + 10] & 0x10 == 0,
name: None,
description: None,
genres: None,
video: None,
audios: None,
series: None,
extended: None,
related_items: None,
});
program.start_at = start_at;
program.duration = decode_bcd_duration(duration).unwrap_or(1);
program.is_free = section[offset + 10] & 0x10 == 0;
parse_descriptors(&section[offset + 12..descriptor_end], program);
}
offset = descriptor_end;
}
true
}
fn parse_descriptors(mut descriptors: &[u8], program: &mut Program) {
while descriptors.len() >= 2 {
let length = usize::from(descriptors[1]);
let Some(end) = 2usize.checked_add(length) else {
return;
};
if end > descriptors.len() {
return;
}
let body = &descriptors[2..end];
match descriptors[0] {
0x4d => parse_short_event(body, program),
0x4e => parse_extended_event(body, program),
0x50 => parse_video_component(body, program),
0x54 => parse_content(body, program),
_ => {}
}
descriptors = &descriptors[end..];
}
}
fn parse_short_event(body: &[u8], program: &mut Program) {
if body.len() < 5 {
return;
}
let name_length = usize::from(body[3]);
let Some(text_length_offset) = 4usize.checked_add(name_length) else {
return;
};
if text_length_offset >= body.len() {
return;
}
let text_length = usize::from(body[text_length_offset]);
let text_start = text_length_offset + 1;
let Some(text_end) = text_start.checked_add(text_length) else {
return;
};
if text_end > body.len() {
return;
}
program.name = Some(decode_arib(&body[4..text_length_offset]));
program.description = Some(decode_arib(&body[text_start..text_end]));
}
fn parse_extended_event(body: &[u8], program: &mut Program) {
if body.len() < 6 {
return;
}
let items_length = usize::from(body[4]);
let Some(items_end) = 5usize.checked_add(items_length) else {
return;
};
if items_end > body.len() {
return;
}
let extended = program.extended.get_or_insert_with(BTreeMap::new);
let mut items = &body[5..items_end];
let mut previous_key = String::new();
while !items.is_empty() {
let description_length = usize::from(items[0]);
let Some(description_end) = 1usize.checked_add(description_length) else {
return;
};
if description_end >= items.len() {
return;
}
let item_length = usize::from(items[description_end]);
let item_start = description_end + 1;
let Some(item_end) = item_start.checked_add(item_length) else {
return;
};
if item_end > items.len() {
return;
}
let decoded_key = decode_arib(&items[1..description_end]);
let key = if decoded_key.is_empty() {
previous_key.clone()
} else {
previous_key.clone_from(&decoded_key);
decoded_key
};
let value = decode_arib(&items[item_start..item_end]);
extended
.entry(key)
.and_modify(|current| current.push_str(&value))
.or_insert(value);
items = &items[item_end..];
}
}
fn parse_video_component(body: &[u8], program: &mut Program) {
if body.len() < 2 {
return;
}
let stream_content = body[0] & 0x0f;
let component_type = body[1];
let video_type = match stream_content {
1 => ProgramVideoType::Mpeg2,
5 => ProgramVideoType::H264,
9 => ProgramVideoType::H265,
_ => return,
};
let resolution = match component_type {
0x01..=0x04 => ProgramVideoResolution::I480,
0x83 => ProgramVideoResolution::P4320,
0x91..=0x94 => ProgramVideoResolution::P2160,
0xa1..=0xa4 => ProgramVideoResolution::P480,
0xb1..=0xb4 => ProgramVideoResolution::I1080,
0xc1..=0xc4 => ProgramVideoResolution::P720,
0xd1..=0xd4 => ProgramVideoResolution::P240,
0xe1..=0xe4 => ProgramVideoResolution::P1080,
_ => return,
};
program.video = Some(ProgramVideo {
video_type,
resolution,
stream_content,
component_type,
});
}
fn parse_content(body: &[u8], program: &mut Program) {
let genres = body
.chunks_exact(2)
.map(|content| ProgramGenre {
lv1: content[0] >> 4,
lv2: content[0] & 0x0f,
un1: content[1] >> 4,
un2: content[1] & 0x0f,
})
.collect::<Vec<_>>();
if !genres.is_empty() {
program.genres = Some(genres);
}
}
fn decode_mjd_time(value: &[u8]) -> Option<u64> {
if value.len() != 5 || value.iter().all(|byte| *byte == 0xff) {
return None;
}
let mjd = i64::from(u16::from_be_bytes([value[0], value[1]]));
let hour = i64::from(decode_bcd(value[2])?);
let minute = i64::from(decode_bcd(value[3])?);
let second = i64::from(decode_bcd(value[4])?);
if hour > 23 || minute > 59 || second > 59 {
return None;
}
let unix_seconds = (mjd - 40_587)
.checked_mul(86_400)?
.checked_add((hour - 9).checked_mul(3600)?)?
.checked_add(minute.checked_mul(60)?)?
.checked_add(second)?;
u64::try_from(unix_seconds.checked_mul(1000)?).ok()
}
fn decode_bcd_duration(value: &[u8]) -> Option<u64> {
if value.len() != 3 || value.iter().all(|byte| *byte == 0xff) {
return None;
}
let hours = u64::from(decode_bcd(value[0])?);
let minutes = u64::from(decode_bcd(value[1])?);
let seconds = u64::from(decode_bcd(value[2])?);
if minutes > 59 || seconds > 59 {
return None;
}
hours
.checked_mul(3600)?
.checked_add(minutes.checked_mul(60)?)?
.checked_add(seconds)?
.checked_mul(1000)
}
fn decode_bcd(value: u8) -> Option<u8> {
let high = value >> 4;
let low = value & 0x0f;
(high <= 9 && low <= 9).then_some(high * 10 + low)
}
#[derive(Debug, Clone, Copy)]
enum GraphicSet {
Kanji,
Alphanumeric,
Hiragana,
Katakana,
}
pub(crate) fn decode_arib(input: &[u8]) -> String {
let mut output = String::new();
let mut sets = [
GraphicSet::Kanji,
GraphicSet::Alphanumeric,
GraphicSet::Hiragana,
GraphicSet::Katakana,
];
let mut left = 0usize;
let mut right = 2usize;
let mut single_shift = None;
let mut offset = 0;
while offset < input.len() {
let byte = input[offset];
match byte {
0x0e => left = 1,
0x0f => left = 0,
0x19 => single_shift = Some(2),
0x1d => single_shift = Some(3),
0x1b => {
offset += consume_escape(&input[offset..], &mut sets, &mut left, &mut right)
.saturating_sub(1);
}
0x0d => output.push('\n'),
0x20 | 0xa0 => output.push(' '),
0x21..=0x7e => {
let set = single_shift.take().unwrap_or(left);
let consumed = decode_graphic(sets[set], &input[offset..], false, &mut output);
offset += consumed.saturating_sub(1);
}
0xa1..=0xfe => {
let consumed = decode_graphic(sets[right], &input[offset..], true, &mut output);
offset += consumed.saturating_sub(1);
}
_ => {}
}
offset += 1;
}
output
}
fn decode_graphic(set: GraphicSet, input: &[u8], high_bit: bool, output: &mut String) -> usize {
let first = if high_bit { input[0] & 0x7f } else { input[0] };
match set {
GraphicSet::Kanji => {
let Some(second) = input.get(1).copied() else {
return 1;
};
let second = if high_bit { second & 0x7f } else { second };
append_euc_jp(&[first | 0x80, second | 0x80], output);
2
}
GraphicSet::Alphanumeric => {
output.push(char::from(first));
1
}
GraphicSet::Hiragana => {
append_euc_jp(&[0xa4, first | 0x80], output);
1
}
GraphicSet::Katakana => {
append_euc_jp(&[0xa5, first | 0x80], output);
1
}
}
}
fn append_euc_jp(bytes: &[u8], output: &mut String) {
let (decoded, had_errors) = EUC_JP.decode_without_bom_handling(bytes);
if had_errors {
output.push('\u{fffd}');
} else {
output.push_str(&decoded);
}
}
fn consume_escape(
input: &[u8],
sets: &mut [GraphicSet; 4],
left: &mut usize,
right: &mut usize,
) -> usize {
let Some(second) = input.get(1).copied() else {
return 1;
};
match second {
0x6e => *left = 2,
0x6f => *left = 3,
0x7c => *right = 3,
0x7d => *right = 2,
0x7e => *right = 1,
0x28..=0x2b => {
let Some(code) = input.get(2).copied() else {
return 2;
};
if let Some(set) = graphic_set(code) {
sets[usize::from(second - 0x28)] = set;
}
return 3;
}
0x24 => {
let Some(third) = input.get(2).copied() else {
return 2;
};
if matches!(third, 0x28..=0x2b) {
let Some(code) = input.get(3).copied() else {
return 3;
};
if let Some(set) = graphic_set(code) {
sets[usize::from(third - 0x28)] = set;
}
return 4;
}
if let Some(set) = graphic_set(third) {
sets[0] = set;
}
return 3;
}
_ => {}
}
2
}
const fn graphic_set(code: u8) -> Option<GraphicSet> {
match code {
0x42 | 0x39 => Some(GraphicSet::Kanji),
0x4a | 0x36 => Some(GraphicSet::Alphanumeric),
0x30 | 0x37 => Some(GraphicSet::Hiragana),
0x31 | 0x38 | 0x49 => Some(GraphicSet::Katakana),
_ => None,
}
}
#[cfg(test)]
mod tests {
use crate::ts::{PACKET_SIZE, crc32_mpeg2};
use super::{EitCollector, decode_arib, decode_bcd_duration, decode_mjd_time};
#[test]
fn decodes_common_arib_character_sets() {
assert_eq!(decode_arib(&[0x0e, b'T', b'e', b's', b't']), "Test");
assert_eq!(decode_arib(&[0x19, 0x22]), "あ");
assert_eq!(decode_arib(&[0x1d, 0x22]), "ア");
}
#[test]
fn converts_arib_time_fields() {
assert_eq!(decode_mjd_time(&[0x9e, 0x8b, 0x09, 0x00, 0x00]), Some(0));
assert_eq!(decode_bcd_duration(&[0x01, 0x30, 0x00]), Some(5_400_000));
}
#[test]
fn collects_program_from_eit_packet() {
let short_event = [
0x4d, 0x0d, b'j', b'p', b'n', 0x05, 0x0e, b'T', b'e', b's', b't', 0x03, 0x0e, b'O',
b'K',
];
let mut section = vec![
0x4e, 0xb0, 0x00, 0x00, 0x65, 0xc1, 0x00, 0x00, 0x00, 0x01, 0x7f, 0xf0, 0x00, 0x4e,
0x00, 0x01, 0x9e, 0x8b, 0x09, 0x00, 0x00, 0x01, 0x00, 0x00, 0x80, 0x0f,
];
section.extend_from_slice(&short_event);
let section_length = section.len() - 3 + 4;
section[1] = 0xb0 | u8::try_from(section_length >> 8).expect("section length");
section[2] = u8::try_from(section_length & 0xff).expect("section length");
let crc = crc32_mpeg2(&section);
section.extend_from_slice(&crc.to_be_bytes());
let mut packet = [0xff; PACKET_SIZE];
packet[0] = 0x47;
packet[1] = 0x40;
packet[2] = 0x12;
packet[3] = 0x10;
packet[4] = 0;
packet[5..5 + section.len()].copy_from_slice(&section);
let mut collector = EitCollector::default();
collector.push(&packet);
let programs = collector.into_programs();
assert_eq!(programs.len(), 1);
assert_eq!(programs[0].name.as_deref(), Some("Test"));
assert_eq!(programs[0].description.as_deref(), Some("OK"));
assert_eq!(programs[0].duration, 3_600_000);
assert!(programs[0].is_free);
}
}

48
crates/mirakurun-core/src/error.rs Normal file
View File

@@ -0,0 +1,48 @@
// Copyright 2026 Mirakurun contributors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
use std::path::PathBuf;
#[derive(Debug, thiserror::Error)]
pub enum CoreError {
#[error("failed to read {kind} from {path}: {source}")]
Read {
kind: &'static str,
path: PathBuf,
#[source]
source: std::io::Error,
},
#[error("failed to write {kind} to {path}: {source}")]
Write {
kind: &'static str,
path: PathBuf,
#[source]
source: std::io::Error,
},
#[error("invalid YAML in {path}: {source}")]
Yaml {
path: PathBuf,
#[source]
source: serde_yaml_ng::Error,
},
#[error("invalid JSON in {path}: {source}")]
Json {
path: PathBuf,
#[source]
source: serde_json::Error,
},
#[error("failed to serialize JSON: {0}")]
JsonSerialize(#[from] serde_json::Error),
#[error("failed to serialize YAML: {0}")]
YamlSerialize(#[from] serde_yaml_ng::Error),
#[error("blocking task failed: {0}")]
Join(#[from] tokio::task::JoinError),
#[error("invalid configuration: {0}")]
InvalidConfig(String),
#[error("invalid MPEG-TS packet: {0}")]
InvalidTransportStream(&'static str),
}
pub type Result<T> = std::result::Result<T, CoreError>;

319
crates/mirakurun-core/src/filter.rs Normal file
View File

@@ -0,0 +1,319 @@
// Copyright 2026 Mirakurun contributors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
use std::collections::{HashMap, HashSet};
use bytes::Bytes;
use crate::ts::{PACKET_SIZE, Packet, PacketFramer, crc32_mpeg2};
const PID_PAT: u16 = 0x0000;
const PID_CAT: u16 = 0x0001;
const PID_NIT: u16 = 0x0010;
const PID_SDT: u16 = 0x0011;
const PID_EIT: u16 = 0x0012;
const PID_RST: u16 = 0x0013;
const PID_TOT: u16 = 0x0014;
#[derive(Debug)]
pub struct ServiceFilter {
service_id: u16,
framer: PacketFramer,
assemblers: HashMap<u16, SectionAssembler>,
pmt_pid: Option<u16>,
allowed_pids: HashSet<u16>,
pat_counter: u8,
}
impl ServiceFilter {
#[must_use]
pub fn new(service_id: u16) -> Self {
Self {
service_id,
framer: PacketFramer::default(),
assemblers: HashMap::new(),
pmt_pid: None,
allowed_pids: HashSet::from([PID_CAT, PID_NIT, PID_SDT, PID_EIT, PID_RST, PID_TOT]),
pat_counter: 0,
}
}
#[must_use]
pub fn push(&mut self, chunk: &[u8]) -> Bytes {
let mut packets = Vec::new();
self.framer.push(chunk, |packet| packets.push(*packet));
let mut output = Vec::with_capacity(packets.len() * PACKET_SIZE);
for packet_bytes in packets {
let Ok(packet) = Packet::new(&packet_bytes) else {
continue;
};
let pid = packet.pid();
if pid == PID_PAT {
for section in self.assemblers.entry(pid).or_default().push(packet) {
if let Some(program) = parse_pat(&section, self.service_id) {
self.pmt_pid = Some(program.pmt_pid);
self.allowed_pids.insert(program.pmt_pid);
output.extend_from_slice(&build_pat_packet(
&section,
program,
self.pat_counter,
));
self.pat_counter = self.pat_counter.wrapping_add(1) & 0x0f;
}
}
continue;
}
if Some(pid) == self.pmt_pid {
for section in self.assemblers.entry(pid).or_default().push(packet) {
if let Some(program_pids) = parse_pmt(&section, self.service_id) {
self.allowed_pids.extend(program_pids);
}
}
}
if self.allowed_pids.contains(&pid) {
output.extend_from_slice(&packet_bytes);
}
}
Bytes::from(output)
}
}
#[derive(Debug, Clone, Copy)]
struct PatProgram {
program_number: u16,
pmt_pid: u16,
}
#[derive(Debug, Default)]
pub(crate) struct SectionAssembler {
buffer: Vec<u8>,
expected: Option<usize>,
}
impl SectionAssembler {
pub(crate) fn push(&mut self, packet: Packet<'_>) -> Vec<Vec<u8>> {
let Some(payload) = packet.payload() else {
return Vec::new();
};
if payload.is_empty() {
return Vec::new();
}
let mut completed = Vec::new();
if packet.payload_unit_start() {
let pointer = usize::from(payload[0]);
let pointer_end = 1usize.saturating_add(pointer).min(payload.len());
if !self.buffer.is_empty() && pointer_end > 1 {
self.consume(&payload[1..pointer_end], &mut completed);
}
self.buffer.clear();
self.expected = None;
self.consume(&payload[pointer_end..], &mut completed);
} else {
self.consume(payload, &mut completed);
}
completed
}
fn consume(&mut self, mut data: &[u8], completed: &mut Vec<Vec<u8>>) {
while !data.is_empty() {
if self.buffer.is_empty() && data[0] == 0xff {
return;
}
if self.expected.is_none() && self.buffer.len() < 3 {
let needed = 3 - self.buffer.len();
let taken = needed.min(data.len());
self.buffer.extend_from_slice(&data[..taken]);
data = &data[taken..];
if self.buffer.len() == 3 {
let section_length =
(usize::from(self.buffer[1] & 0x0f) << 8) | usize::from(self.buffer[2]);
self.expected = Some(3 + section_length);
}
continue;
}
let Some(expected) = self.expected else {
continue;
};
let needed = expected.saturating_sub(self.buffer.len());
let taken = needed.min(data.len());
self.buffer.extend_from_slice(&data[..taken]);
data = &data[taken..];
if self.buffer.len() == expected {
completed.push(std::mem::take(&mut self.buffer));
self.expected = None;
}
}
}
}
fn parse_pat(section: &[u8], service_id: u16) -> Option<PatProgram> {
if section.first().copied() != Some(0x00) || section.len() < 12 {
return None;
}
let end = section.len().checked_sub(4)?;
let mut offset = 8;
while offset + 4 <= end {
let program_number = u16::from_be_bytes([section[offset], section[offset + 1]]);
let pmt_pid = (u16::from(section[offset + 2] & 0x1f) << 8) | u16::from(section[offset + 3]);
if program_number == service_id {
return Some(PatProgram {
program_number,
pmt_pid,
});
}
offset += 4;
}
None
}
fn parse_pmt(section: &[u8], service_id: u16) -> Option<HashSet<u16>> {
if section.first().copied() != Some(0x02) || section.len() < 16 {
return None;
}
if u16::from_be_bytes([section[3], section[4]]) != service_id {
return None;
}
let end = section.len().checked_sub(4)?;
let mut pids = HashSet::new();
pids.insert((u16::from(section[8] & 0x1f) << 8) | u16::from(section[9]));
let program_info_length = (usize::from(section[10] & 0x0f) << 8) | usize::from(section[11]);
let program_info_end = 12usize.checked_add(program_info_length)?;
if program_info_end > end {
return None;
}
collect_ca_pids(&section[12..program_info_end], &mut pids);
let mut offset = program_info_end;
while offset + 5 <= end {
let elementary_pid =
(u16::from(section[offset + 1] & 0x1f) << 8) | u16::from(section[offset + 2]);
let info_length =
(usize::from(section[offset + 3] & 0x0f) << 8) | usize::from(section[offset + 4]);
let info_end = offset.checked_add(5)?.checked_add(info_length)?;
if info_end > end {
return None;
}
pids.insert(elementary_pid);
collect_ca_pids(&section[offset + 5..info_end], &mut pids);
offset = info_end;
}
Some(pids)
}
fn collect_ca_pids(descriptors: &[u8], pids: &mut HashSet<u16>) {
let mut offset = 0;
while offset + 2 <= descriptors.len() {
let length = usize::from(descriptors[offset + 1]);
let end = offset.saturating_add(2).saturating_add(length);
if end > descriptors.len() {
return;
}
if descriptors[offset] == 0x09 && length >= 4 {
pids.insert(
(u16::from(descriptors[offset + 4] & 0x1f) << 8)
| u16::from(descriptors[offset + 5]),
);
}
offset = end;
}
}
fn build_pat_packet(original: &[u8], program: PatProgram, continuity: u8) -> [u8; PACKET_SIZE] {
let transport_stream_id = original.get(3..5).unwrap_or(&[0, 0]);
let version = original.get(5).copied().unwrap_or(0xc1);
let mut section = vec![
0x00,
0xb0,
0x0d,
transport_stream_id[0],
transport_stream_id[1],
version,
0x00,
0x00,
(program.program_number >> 8) as u8,
u8::try_from(program.program_number & 0xff).unwrap_or(0),
0xe0 | u8::try_from(program.pmt_pid >> 8).unwrap_or(0),
u8::try_from(program.pmt_pid & 0xff).unwrap_or(0),
];
section.extend_from_slice(&crc32_mpeg2(&section).to_be_bytes());
let mut packet = [0xff; PACKET_SIZE];
packet[0] = 0x47;
packet[1] = 0x40;
packet[2] = 0x00;
packet[3] = 0x10 | (continuity & 0x0f);
packet[4] = 0x00;
packet[5..5 + section.len()].copy_from_slice(&section);
packet
}
#[cfg(test)]
mod tests {
use crate::ts::{PACKET_SIZE, crc32_mpeg2};
use super::ServiceFilter;
#[test]
fn retains_only_selected_program_pids_and_rewrites_pat() {
let association_packet = psi_packet(
0,
&section_with_crc(vec![
0x00, 0xb0, 0x11, 0x00, 0x01, 0xc1, 0x00, 0x00, 0x00, 0x65, 0xe1, 0x00, 0x00, 0x66,
0xe2, 0x00,
]),
);
let program_map_packet = psi_packet(
0x0100,
&section_with_crc(vec![
0x02, 0xb0, 0x12, 0x00, 0x65, 0xc1, 0x00, 0x00, 0xe1, 0x01, 0xf0, 0x00, 0x1b, 0xe1,
0x01, 0xf0, 0x00,
]),
);
let video = payload_packet(0x0101);
let other = payload_packet(0x0201);
let mut input = Vec::new();
input.extend_from_slice(&association_packet);
input.extend_from_slice(&program_map_packet);
input.extend_from_slice(&video);
input.extend_from_slice(&other);
let output = ServiceFilter::new(101).push(&input);
assert_eq!(output.len(), PACKET_SIZE * 3);
assert_eq!(&output[0..3], &[0x47, 0x40, 0x00]);
assert_eq!(&output[PACKET_SIZE..PACKET_SIZE + 3], &[0x47, 0x41, 0x00]);
assert_eq!(
&output[PACKET_SIZE * 2..PACKET_SIZE * 2 + 3],
&[0x47, 0x01, 0x01]
);
}
fn section_with_crc(mut section: Vec<u8>) -> Vec<u8> {
let crc = crc32_mpeg2(&section);
section.extend_from_slice(&crc.to_be_bytes());
section
}
fn psi_packet(pid: u16, section: &[u8]) -> [u8; PACKET_SIZE] {
let mut packet = [0xff; PACKET_SIZE];
packet[0] = 0x47;
packet[1] = 0x40 | u8::try_from(pid >> 8).unwrap_or(0);
packet[2] = u8::try_from(pid & 0xff).unwrap_or(0);
packet[3] = 0x10;
packet[4] = 0;
packet[5..5 + section.len()].copy_from_slice(section);
packet
}
fn payload_packet(pid: u16) -> [u8; PACKET_SIZE] {
let mut packet = [0xff; PACKET_SIZE];
packet[0] = 0x47;
packet[1] = u8::try_from(pid >> 8).unwrap_or(0);
packet[2] = u8::try_from(pid & 0xff).unwrap_or(0);
packet[3] = 0x10;
packet
}
}

15
crates/mirakurun-core/src/lib.rs Normal file
View File

@@ -0,0 +1,15 @@
// Copyright 2026 Mirakurun contributors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
pub mod config;
pub mod epg;
pub mod error;
pub mod filter;
pub mod persistence;
pub mod service;
pub mod ts;
pub mod tuner;
pub use error::{CoreError, Result};

241
crates/mirakurun-core/src/persistence.rs Normal file
View File

@@ -0,0 +1,241 @@
// Copyright 2026 Mirakurun contributors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
use std::{
io::Write,
path::{Path, PathBuf},
};
use base64::{Engine as _, engine::general_purpose::STANDARD};
use mirakurun_types::ConfigChannel;
use serde::{Serialize, de::DeserializeOwned};
use serde_json::{Map, Value};
use sha2::{Digest, Sha256};
use crate::{Result, error::CoreError};
/// Loads a Mirakurun JSON database after checking its optional integrity sentinel.
///
/// Missing, syntactically broken, and integrity-mismatched databases are treated
/// as empty for compatibility with the Node.js implementation.
///
/// # Errors
///
/// Returns an error for filesystem failures other than a missing file, or when
/// an individual stored item cannot be deserialized.
pub async fn load_json_db<T>(path: &Path, expected_integrity: &str) -> Result<Vec<T>>
where
T: DeserializeOwned,
{
let contents = match tokio::fs::read(path).await {
Ok(contents) => contents,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
Err(source) => {
return Err(CoreError::Read {
kind: "database",
path: path.to_path_buf(),
source,
});
}
};
let mut values: Vec<Value> = match serde_json::from_slice(&contents) {
Ok(values) => values,
Err(error) => {
tracing::warn!(path = %path.display(), %error, "broken database; starting empty");
return Ok(Vec::new());
}
};
if let Some(integrity) = values.first().and_then(integrity_value) {
if integrity != expected_integrity {
tracing::warn!(
path = %path.display(),
expected = expected_integrity,
actual = integrity,
"database integrity check failed; starting empty"
);
return Ok(Vec::new());
}
values.remove(0);
}
values
.into_iter()
.map(|value| {
serde_json::from_value(value).map_err(|source| CoreError::Json {
path: path.to_path_buf(),
source,
})
})
.collect()
}
/// Saves a Mirakurun JSON database with an integrity sentinel.
///
/// # Errors
///
/// Returns an error when serialization or the atomic filesystem update fails.
pub async fn save_json_db<T>(path: &Path, values: &[T], integrity: &str) -> Result<()>
where
T: Serialize,
{
let mut output = Vec::with_capacity(values.len() + 1);
let mut sentinel = Map::new();
sentinel.insert("__integrity__".into(), Value::String(integrity.into()));
output.push(Value::Object(sentinel));
output.extend(
values
.iter()
.map(serde_json::to_value)
.collect::<std::result::Result<Vec<_>, _>>()?,
);
let bytes = serde_json::to_vec(&output)?;
ensure_parent(path, "database").await?;
atomic_write(path, &bytes, "database").await
}
/// Calculates the base64 channel-config SHA-256 used by database integrity sentinels.
///
/// # Errors
///
/// Returns an error if the channel configuration cannot be serialized.
pub fn channels_integrity(channels: &[ConfigChannel]) -> Result<String> {
let json = serde_json::to_vec(channels)?;
let hash = Sha256::digest(json);
Ok(STANDARD.encode(hash))
}
/// Creates the parent directory for a configured data path.
///
/// # Errors
///
/// Returns an error when the directory cannot be created.
pub async fn ensure_parent(path: &Path, kind: &'static str) -> Result<()> {
if let Some(parent) = path.parent() {
tokio::fs::create_dir_all(parent)
.await
.map_err(|source| CoreError::Write {
kind,
path: parent.to_path_buf(),
source,
})?;
}
Ok(())
}
/// Replaces a file atomically using a temporary file in the same directory.
///
/// # Errors
///
/// Returns an error if the temporary file cannot be written, synchronized, or
/// persisted.
pub async fn atomic_write(path: &Path, contents: &[u8], kind: &'static str) -> Result<()> {
let path = path.to_path_buf();
let bytes = contents.to_vec();
tokio::task::spawn_blocking(move || atomic_write_blocking(&path, &bytes, kind)).await?
}
fn atomic_write_blocking(path: &Path, contents: &[u8], kind: &'static str) -> Result<()> {
let parent = path
.parent()
.map_or_else(|| PathBuf::from("."), Path::to_path_buf);
let mut temporary =
tempfile::NamedTempFile::new_in(&parent).map_err(|source| CoreError::Write {
kind,
path: path.to_path_buf(),
source,
})?;
temporary
.write_all(contents)
.and_then(|()| temporary.as_file().sync_all())
.map_err(|source| CoreError::Write {
kind,
path: path.to_path_buf(),
source,
})?;
temporary.persist(path).map_err(|error| CoreError::Write {
kind,
path: path.to_path_buf(),
source: error.error,
})?;
if let Ok(directory) = std::fs::File::open(parent) {
let _ = directory.sync_all();
}
Ok(())
}
fn integrity_value(value: &Value) -> Option<&str> {
value
.as_object()
.and_then(|object| object.get("__integrity__"))
.and_then(Value::as_str)
}
#[cfg(test)]
mod tests {
use mirakurun_types::{ChannelType, ConfigChannel, Service};
use tempfile::tempdir;
use super::{channels_integrity, load_json_db, save_json_db};
#[tokio::test]
async fn database_round_trip_and_integrity_check() {
let directory = tempdir().expect("create tempdir");
let path = directory.path().join("services.json");
let services = vec![Service {
id: 100_002,
service_id: 2,
network_id: 1,
name: "service".into(),
service_type: 1,
logo_id: None,
has_logo_data: None,
remote_control_key_id: None,
epg_ready: None,
epg_updated_at: None,
channel: None,
}];
save_json_db(&path, &services, "expected")
.await
.expect("save database");
let loaded: Vec<Service> = load_json_db(&path, "expected")
.await
.expect("load database");
assert_eq!(loaded, services);
let rejected: Vec<Service> = load_json_db(&path, "different")
.await
.expect("load with mismatched integrity");
assert!(rejected.is_empty());
}
#[test]
fn channel_integrity_is_sha256_of_compact_json() {
let channels = vec![ConfigChannel {
name: "test".into(),
channel_type: ChannelType::Gr,
channel: "27".into(),
service_id: None,
tsmf_rel_ts: None,
command_vars: None,
is_disabled: None,
satelite: None,
satellite: None,
space: None,
freq: None,
polarity: None,
}];
let integrity = channels_integrity(&channels).expect("calculate integrity");
assert_eq!(integrity.len(), 44);
}
#[test]
fn empty_channel_integrity_matches_node_crypto() {
let integrity = channels_integrity(&[]).expect("calculate integrity");
assert_eq!(integrity, "T1PNoYwrqgwDVLtfmj7L5e0Sq02OEbqHPC8RFhICuUU=");
}
}

242
crates/mirakurun-core/src/service.rs Normal file
View File

@@ -0,0 +1,242 @@
// Copyright 2026 Mirakurun contributors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
use std::collections::{HashMap, HashSet};
use mirakurun_types::{Channel, ConfigChannel, Service, service_item_id};
use crate::{
epg::decode_arib,
filter::SectionAssembler,
ts::{Packet, PacketFramer, crc32_mpeg2},
};
const PID_SDT: u16 = 0x0011;
const TABLE_ID_SDT_ACTUAL: u8 = 0x42;
#[derive(Debug)]
pub struct ServiceCollector {
channel: ConfigChannel,
framer: PacketFramer,
assembler: SectionAssembler,
services: HashMap<u64, Service>,
sections_seen: HashSet<(u8, u8)>,
last_section: Option<u8>,
packet_count: u64,
}
impl ServiceCollector {
#[must_use]
pub fn new(channel: ConfigChannel) -> Self {
Self {
channel,
framer: PacketFramer::default(),
assembler: SectionAssembler::default(),
services: HashMap::new(),
sections_seen: HashSet::new(),
last_section: None,
packet_count: 0,
}
}
pub fn push(&mut self, chunk: &[u8]) {
let mut packets = Vec::new();
self.framer.push(chunk, |packet| packets.push(*packet));
for packet_bytes in packets {
self.packet_count = self.packet_count.saturating_add(1);
let Ok(packet) = Packet::new(&packet_bytes) else {
continue;
};
if packet.pid() != PID_SDT {
continue;
}
for section in self.assembler.push(packet) {
let Some(parsed) = parse_sdt_section(&section, &self.channel) else {
continue;
};
self.last_section = Some(parsed.last_section);
self.sections_seen
.insert((parsed.version, parsed.section_number));
for service in parsed.services {
self.services.insert(service.id, service);
}
}
}
}
#[must_use]
pub fn is_complete(&self) -> bool {
self.last_section.is_some_and(|last| {
(0..=last).all(|section| {
self.sections_seen
.iter()
.any(|(_, seen_section)| *seen_section == section)
})
})
}
#[must_use]
pub const fn packet_count(&self) -> u64 {
self.packet_count
}
#[must_use]
pub fn into_services(self) -> Vec<Service> {
let mut services = self.services.into_values().collect::<Vec<_>>();
services.sort_by_key(|service| service.service_id);
services
}
}
struct ParsedSdt {
version: u8,
section_number: u8,
last_section: u8,
services: Vec<Service>,
}
fn parse_sdt_section(section: &[u8], channel: &ConfigChannel) -> Option<ParsedSdt> {
if section.len() < 15
|| section[0] != TABLE_ID_SDT_ACTUAL
|| section[5] & 0x01 == 0
|| crc32_mpeg2(section) != 0
{
return None;
}
let network_id = u16::from_be_bytes([section[8], section[9]]);
let end = section.len().checked_sub(4)?;
let mut services = Vec::new();
let mut offset = 11;
while offset + 5 <= end {
let service_id = u16::from_be_bytes([section[offset], section[offset + 1]]);
let descriptors_length =
(usize::from(section[offset + 3] & 0x0f) << 8) | usize::from(section[offset + 4]);
let descriptor_start = offset.checked_add(5)?;
let descriptor_end = descriptor_start.checked_add(descriptors_length)?;
if descriptor_end > end {
return None;
}
if channel
.service_id
.is_none_or(|configured| configured == service_id)
{
if let Some((service_type, name)) =
parse_service_descriptor(&section[descriptor_start..descriptor_end])
{
services.push(Service {
id: service_item_id(network_id, service_id),
service_id,
network_id,
name,
service_type,
logo_id: None,
has_logo_data: None,
remote_control_key_id: None,
epg_ready: None,
epg_updated_at: None,
channel: Some(Box::new(Channel {
channel_type: channel.channel_type,
channel: channel.channel.clone(),
name: Some(channel.name.clone()),
services: None,
})),
});
}
}
offset = descriptor_end;
}
Some(ParsedSdt {
version: (section[5] >> 1) & 0x1f,
section_number: section[6],
last_section: section[7],
services,
})
}
fn parse_service_descriptor(mut descriptors: &[u8]) -> Option<(u8, String)> {
while descriptors.len() >= 2 {
let length = usize::from(descriptors[1]);
let end = 2usize.checked_add(length)?;
if end > descriptors.len() {
return None;
}
if descriptors[0] == 0x48 {
let body = &descriptors[2..end];
if body.len() < 3 {
return None;
}
let provider_length = usize::from(body[1]);
let name_length_offset = 2usize.checked_add(provider_length)?;
if name_length_offset >= body.len() {
return None;
}
let name_length = usize::from(body[name_length_offset]);
let name_start = name_length_offset + 1;
let name_end = name_start.checked_add(name_length)?;
if name_end > body.len() {
return None;
}
return Some((body[0], decode_arib(&body[name_start..name_end])));
}
descriptors = &descriptors[end..];
}
None
}
#[cfg(test)]
mod tests {
use mirakurun_types::{ChannelType, ConfigChannel};
use crate::ts::{PACKET_SIZE, crc32_mpeg2};
use super::ServiceCollector;
#[test]
fn discovers_service_from_sdt() {
let descriptor = [
0x48, 0x0c, 0x01, 0x00, 0x09, 0x0e, b'T', b'e', b's', b't', b' ', b'T', b'V', b'!',
];
let mut section = vec![
0x42, 0xf0, 0x00, 0x00, 0x01, 0xc1, 0x00, 0x00, 0x7f, 0xf0, 0xff, 0x00, 0x65, 0xfc,
0x80, 0x0e,
];
section.extend_from_slice(&descriptor);
let section_length = section.len() - 3 + 4;
section[1] = 0xf0 | u8::try_from(section_length >> 8).expect("section length");
section[2] = u8::try_from(section_length & 0xff).expect("section length");
let crc = crc32_mpeg2(&section);
section.extend_from_slice(&crc.to_be_bytes());
let mut packet = [0xff; PACKET_SIZE];
packet[0] = 0x47;
packet[1] = 0x40;
packet[2] = 0x11;
packet[3] = 0x10;
packet[4] = 0;
packet[5..5 + section.len()].copy_from_slice(&section);
let mut collector = ServiceCollector::new(ConfigChannel {
name: "channel".into(),
channel_type: ChannelType::Gr,
channel: "27".into(),
service_id: None,
tsmf_rel_ts: None,
command_vars: None,
is_disabled: None,
satelite: None,
satellite: None,
space: None,
freq: None,
polarity: None,
});
collector.push(&packet);
assert!(collector.is_complete());
let services = collector.into_services();
assert_eq!(services.len(), 1);
assert_eq!(services[0].name, "Test TV!");
assert_eq!(services[0].network_id, 0x7ff0);
assert_eq!(services[0].service_id, 101);
}
}

220
crates/mirakurun-core/src/ts.rs Normal file
View File

@@ -0,0 +1,220 @@
// Copyright 2026 Mirakurun contributors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
use std::collections::HashMap;
use crate::{Result, error::CoreError};
pub const PACKET_SIZE: usize = 188;
pub const SYNC_BYTE: u8 = 0x47;
#[derive(Debug, Clone, Copy)]
pub struct Packet<'a> {
bytes: &'a [u8; PACKET_SIZE],
}
impl<'a> Packet<'a> {
/// Creates a view over one 188-byte transport-stream packet.
///
/// # Errors
///
/// Returns an error if the MPEG-TS sync byte is missing.
pub fn new(bytes: &'a [u8; PACKET_SIZE]) -> Result<Self> {
if bytes[0] != SYNC_BYTE {
return Err(CoreError::InvalidTransportStream("missing sync byte"));
}
Ok(Self { bytes })
}
#[must_use]
pub fn bytes(self) -> &'a [u8; PACKET_SIZE] {
self.bytes
}
#[must_use]
pub fn transport_error(self) -> bool {
self.bytes[1] & 0x80 != 0
}
#[must_use]
pub fn payload_unit_start(self) -> bool {
self.bytes[1] & 0x40 != 0
}
#[must_use]
pub fn pid(self) -> u16 {
(u16::from(self.bytes[1] & 0x1f) << 8) | u16::from(self.bytes[2])
}
#[must_use]
pub fn has_payload(self) -> bool {
self.bytes[3] & 0x10 != 0
}
#[must_use]
pub fn has_adaptation_field(self) -> bool {
self.bytes[3] & 0x20 != 0
}
#[must_use]
pub fn continuity_counter(self) -> u8 {
self.bytes[3] & 0x0f
}
#[must_use]
pub fn payload(self) -> Option<&'a [u8]> {
if !self.has_payload() {
return None;
}
let offset = if self.has_adaptation_field() {
5usize.saturating_add(usize::from(self.bytes[4]))
} else {
4
};
(offset <= PACKET_SIZE).then(|| &self.bytes[offset..])
}
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct PidStats {
pub packets: u64,
pub drops: u64,
}
#[derive(Debug, Default)]
pub struct ContinuityTracker {
entries: HashMap<u16, (u8, PidStats)>,
}
impl ContinuityTracker {
pub fn observe(&mut self, packet: Packet<'_>) {
let pid = packet.pid();
if pid == 0x1fff {
return;
}
let counter = packet.continuity_counter();
let entry = self.entries.entry(pid).or_default();
if packet.has_payload() && entry.1.packets > 0 {
let expected = entry.0.wrapping_add(1) & 0x0f;
if counter != expected {
entry.1.drops += u64::from(counter.wrapping_sub(expected) & 0x0f);
}
}
entry.0 = counter;
entry.1.packets += 1;
}
#[must_use]
pub fn stats(&self) -> HashMap<u16, PidStats> {
self.entries
.iter()
.map(|(&pid, &(_, stats))| (pid, stats))
.collect()
}
}
#[derive(Debug, Default)]
pub struct PacketFramer {
buffer: Vec<u8>,
}
impl PacketFramer {
pub fn push<F>(&mut self, chunk: &[u8], mut emit: F)
where
F: FnMut(&[u8; PACKET_SIZE]),
{
self.buffer.extend_from_slice(chunk);
loop {
let Some(sync_position) = find_sync(&self.buffer) else {
let keep = self.buffer.len().min(PACKET_SIZE - 1);
self.buffer.drain(..self.buffer.len() - keep);
break;
};
if sync_position > 0 {
self.buffer.drain(..sync_position);
}
if self.buffer.len() < PACKET_SIZE {
break;
}
let mut packet = [0_u8; PACKET_SIZE];
packet.copy_from_slice(&self.buffer[..PACKET_SIZE]);
emit(&packet);
self.buffer.drain(..PACKET_SIZE);
}
}
}
#[must_use]
pub fn crc32_mpeg2(bytes: &[u8]) -> u32 {
let mut crc = 0xffff_ffffu32;
for &byte in bytes {
crc ^= u32::from(byte) << 24;
for _ in 0..8 {
crc = if crc & 0x8000_0000 != 0 {
(crc << 1) ^ 0x04c1_1db7
} else {
crc << 1
};
}
}
crc
}
fn find_sync(buffer: &[u8]) -> Option<usize> {
buffer.iter().enumerate().find_map(|(index, byte)| {
if *byte != SYNC_BYTE {
return None;
}
if index + PACKET_SIZE >= buffer.len() || buffer[index + PACKET_SIZE] == SYNC_BYTE {
Some(index)
} else {
None
}
})
}
#[cfg(test)]
mod tests {
use super::{ContinuityTracker, PACKET_SIZE, Packet, PacketFramer, crc32_mpeg2};
fn packet(pid: u16, counter: u8) -> [u8; PACKET_SIZE] {
let mut packet = [0xff; PACKET_SIZE];
packet[0] = 0x47;
packet[1] = (pid >> 8) as u8 & 0x1f;
packet[2] = u8::try_from(pid & 0xff).unwrap_or(0);
packet[3] = 0x10 | (counter & 0x0f);
packet
}
#[test]
fn frames_unaligned_packets() {
let first = packet(0, 0);
let second = packet(0, 1);
let mut bytes = vec![0, 1, 2];
bytes.extend_from_slice(&first);
bytes.extend_from_slice(&second);
let mut count = 0;
PacketFramer::default().push(&bytes, |_| count += 1);
assert_eq!(count, 2);
}
#[test]
fn detects_continuity_drops() {
let mut tracker = ContinuityTracker::default();
let first = packet(256, 0);
let second = packet(256, 2);
tracker.observe(Packet::new(&first).expect("valid packet"));
tracker.observe(Packet::new(&second).expect("valid packet"));
assert_eq!(tracker.stats()[&256].drops, 1);
}
#[test]
fn mpeg_crc_matches_known_pat_section() {
let section = [
0x00, 0xb0, 0x0d, 0x00, 0x01, 0xc1, 0x00, 0x00, 0x00, 0x01, 0xe1, 0x00,
];
assert_eq!(crc32_mpeg2(&section), 0xe8f9_5e7d);
}
}

846
crates/mirakurun-core/src/tuner.rs Normal file
View File

@@ -0,0 +1,846 @@
// Copyright 2026 Mirakurun contributors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
use std::{
collections::HashMap,
process::Stdio,
sync::{
Arc,
atomic::{AtomicU64, Ordering},
},
time::Duration,
};
use bytes::Bytes;
use mirakurun_types::{
ChannelType, CommandVariable, ConfigChannel, ConfigTuner, StreamSetting, TunerDevice, TunerUser,
};
use nix::{
sys::signal::{Signal, kill},
unistd::Pid,
};
use tokio::{
io::{AsyncRead, AsyncReadExt},
process::{Child, ChildStdout, Command},
sync::{broadcast, mpsc, oneshot},
};
use crate::{CoreError, Result};
const STREAM_CHANNEL_CAPACITY: usize = 512;
const DEVICE_COMMAND_CAPACITY: usize = 64;
const IDLE_TIMEOUT: Duration = Duration::from_secs(3);
const TERMINATE_TIMEOUT: Duration = Duration::from_secs(6);
static NEXT_SUBSCRIPTION_ID: AtomicU64 = AtomicU64::new(1);
#[derive(Debug, Clone)]
pub struct StreamRequest {
pub channel: ConfigChannel,
pub service_id: Option<u16>,
pub event_id: Option<u16>,
pub priority: i32,
pub agent: Option<String>,
pub url: Option<String>,
pub disable_decoder: bool,
}
#[derive(Debug)]
pub struct TunerSubscription {
pub device_index: usize,
pub receiver: broadcast::Receiver<Bytes>,
pub decoder: Option<String>,
pub user_id: String,
command: mpsc::Sender<DeviceCommand>,
subscription_id: u64,
}
impl Drop for TunerSubscription {
fn drop(&mut self) {
let _ = self.command.try_send(DeviceCommand::Unsubscribe {
subscription_id: self.subscription_id,
});
}
}
#[derive(Debug, Clone)]
pub struct TunerManager {
devices: Arc<Vec<DeviceHandle>>,
}
impl TunerManager {
#[must_use]
pub fn new(configs: &[ConfigTuner]) -> Self {
let devices = configs
.iter()
.filter(|config| config.is_disabled != Some(true))
.cloned()
.enumerate()
.map(|(index, config)| DeviceHandle::spawn(index, config))
.collect();
Self {
devices: Arc::new(devices),
}
}
/// Subscribes to a compatible tuner, sharing an existing multiplex when possible.
///
/// # Errors
///
/// Returns an error when no tuner can satisfy the channel and priority, or
/// when the tuner process cannot be started.
pub async fn subscribe(&self, request: StreamRequest) -> Result<TunerSubscription> {
let mut candidates = self.candidates(&request.channel, request.priority).await;
while let Some(index) = candidates.first().copied() {
candidates.remove(0);
match self.devices[index].subscribe(request.clone()).await {
Ok(subscription) => return Ok(subscription),
Err(CoreError::InvalidConfig(_)) => continue,
Err(error) => return Err(error),
}
}
Err(CoreError::InvalidConfig("no available tuners".into()))
}
pub async fn statuses(&self) -> Vec<TunerDevice> {
let mut statuses = Vec::with_capacity(self.devices.len());
for device in self.devices.iter() {
if let Some(status) = device.status().await {
statuses.push(status);
}
}
statuses
}
/// Terminates the process owned by a tuner index.
///
/// # Errors
///
/// Returns an error if the index does not exist or the device task stopped.
pub async fn kill(&self, index: usize) -> Result<()> {
let Some(device) = self.devices.get(index) else {
return Err(CoreError::InvalidConfig(format!(
"tuner index {index} does not exist"
)));
};
device.kill().await
}
async fn candidates(&self, channel: &ConfigChannel, priority: i32) -> Vec<usize> {
let mut joined = Vec::new();
let mut free = Vec::new();
let mut idle = Vec::new();
let mut takeover = Vec::new();
for device in self.devices.iter() {
let Some(snapshot) = device.snapshot().await else {
continue;
};
if !snapshot.types.contains(&channel.channel_type) || snapshot.fault {
continue;
}
if snapshot.channel.as_ref() == Some(channel) && snapshot.available {
joined.push((snapshot.index, snapshot.priority));
} else if snapshot.channel.is_none() && snapshot.users == 0 && snapshot.available {
free.push((snapshot.index, snapshot.priority));
} else if snapshot.users == 0 && snapshot.available {
idle.push((snapshot.index, snapshot.priority));
} else if priority >= 0
&& snapshot.available
&& snapshot.users > 0
&& snapshot.priority < priority
{
takeover.push((snapshot.index, snapshot.priority));
}
}
takeover.sort_by_key(|(_, current_priority)| *current_priority);
joined
.into_iter()
.chain(free)
.chain(idle)
.chain(takeover)
.map(|(index, _)| index)
.collect()
}
}
#[derive(Debug, Clone)]
struct DeviceHandle {
command: mpsc::Sender<DeviceCommand>,
}
impl DeviceHandle {
fn spawn(index: usize, config: ConfigTuner) -> Self {
let (command, receiver) = mpsc::channel(DEVICE_COMMAND_CAPACITY);
let actor_command = command.clone();
tokio::spawn(async move {
DeviceActor::new(index, config, actor_command)
.run(receiver)
.await;
});
Self { command }
}
async fn subscribe(&self, request: StreamRequest) -> Result<TunerSubscription> {
let (response, result) = oneshot::channel();
self.command
.send(DeviceCommand::Subscribe { request, response })
.await
.map_err(|_| CoreError::InvalidConfig("tuner task stopped".into()))?;
result
.await
.map_err(|_| CoreError::InvalidConfig("tuner task stopped".into()))?
}
async fn status(&self) -> Option<TunerDevice> {
let (response, result) = oneshot::channel();
self.command
.send(DeviceCommand::Status { response })
.await
.ok()?;
result.await.ok()
}
async fn snapshot(&self) -> Option<DeviceSnapshot> {
let (response, result) = oneshot::channel();
self.command
.send(DeviceCommand::Snapshot { response })
.await
.ok()?;
result.await.ok()
}
async fn kill(&self) -> Result<()> {
let (response, result) = oneshot::channel();
self.command
.send(DeviceCommand::Kill { response })
.await
.map_err(|_| CoreError::InvalidConfig("tuner task stopped".into()))?;
result
.await
.map_err(|_| CoreError::InvalidConfig("tuner task stopped".into()))?
}
}
#[derive(Debug)]
enum DeviceCommand {
Subscribe {
request: StreamRequest,
response: oneshot::Sender<Result<TunerSubscription>>,
},
Unsubscribe {
subscription_id: u64,
},
StopIfIdle {
generation: u64,
},
OutputEnded {
generation: u64,
},
Status {
response: oneshot::Sender<TunerDevice>,
},
Snapshot {
response: oneshot::Sender<DeviceSnapshot>,
},
Kill {
response: oneshot::Sender<Result<()>>,
},
}
#[derive(Debug)]
struct DeviceActor {
index: usize,
config: ConfigTuner,
command_sender: mpsc::Sender<DeviceCommand>,
current_channel: Option<ConfigChannel>,
command_display: Option<String>,
process: Option<ProcessGroup>,
broadcaster: Option<broadcast::Sender<Bytes>>,
users: HashMap<u64, TunerUser>,
available: bool,
fault: bool,
fatal_count: u8,
generation: u64,
}
impl DeviceActor {
fn new(index: usize, config: ConfigTuner, command_sender: mpsc::Sender<DeviceCommand>) -> Self {
Self {
index,
config,
command_sender,
current_channel: None,
command_display: None,
process: None,
broadcaster: None,
users: HashMap::new(),
available: true,
fault: false,
fatal_count: 0,
generation: 0,
}
}
async fn run(mut self, mut commands: mpsc::Receiver<DeviceCommand>) {
while let Some(command) = commands.recv().await {
match command {
DeviceCommand::Subscribe { request, response } => {
let result = self.subscribe(request).await;
let _ = response.send(result);
}
DeviceCommand::Unsubscribe { subscription_id } => {
self.unsubscribe(subscription_id);
}
DeviceCommand::StopIfIdle { generation } => {
if generation == self.generation && self.users.is_empty() {
if let Err(error) = self.stop_process().await {
tracing::warn!(device = self.index, %error, "failed to stop idle tuner");
}
}
}
DeviceCommand::OutputEnded { generation } => {
if generation == self.generation {
self.output_ended().await;
}
}
DeviceCommand::Status { response } => {
let _ = response.send(self.status());
}
DeviceCommand::Snapshot { response } => {
let _ = response.send(self.snapshot());
}
DeviceCommand::Kill { response } => {
self.users.clear();
self.broadcaster = None;
let result = self.stop_process().await;
let _ = response.send(result);
}
}
}
let _ = self.stop_process().await;
}
async fn subscribe(&mut self, request: StreamRequest) -> Result<TunerSubscription> {
if self.fault || !self.config.types.contains(&request.channel.channel_type) {
return Err(CoreError::InvalidConfig(
"tuner is not available for this channel".into(),
));
}
if self.current_channel.as_ref() != Some(&request.channel) {
let current_priority = self
.users
.values()
.map(|user| user.priority)
.max()
.unwrap_or(-2);
if !self.users.is_empty() && request.priority <= current_priority {
return Err(CoreError::InvalidConfig(
"tuner has a higher priority user".into(),
));
}
self.users.clear();
self.broadcaster = None;
self.stop_process().await?;
self.start_process(request.channel.clone()).await?;
} else if self.process.is_none() {
self.start_process(request.channel.clone()).await?;
}
let broadcaster = self
.broadcaster
.as_ref()
.ok_or_else(|| CoreError::InvalidConfig("tuner stream is unavailable".into()))?;
let subscription_id = NEXT_SUBSCRIPTION_ID.fetch_add(1, Ordering::Relaxed);
let user_id = subscription_id.to_string();
let decoder = if request.disable_decoder
|| (self.config.remote_mirakurun_host.is_some()
&& self.config.remote_mirakurun_decoder == Some(true))
{
None
} else {
self.config.decoder.clone()
};
self.users.insert(
subscription_id,
TunerUser {
id: user_id.clone(),
priority: request.priority,
agent: request.agent,
url: request.url,
disable_decoder: request.disable_decoder.then_some(true),
stream_setting: Some(StreamSetting {
channel: request.channel,
network_id: None,
service_id: request.service_id,
event_id: request.event_id,
no_provide: None,
parse_nit: None,
parse_sdt: None,
parse_eit: None,
}),
stream_info: None,
},
);
Ok(TunerSubscription {
device_index: self.index,
receiver: broadcaster.subscribe(),
decoder,
user_id,
command: self.command_sender.clone(),
subscription_id,
})
}
fn unsubscribe(&mut self, subscription_id: u64) {
if self.users.remove(&subscription_id).is_none() || !self.users.is_empty() {
return;
}
let generation = self.generation;
let command = self.command_sender.clone();
tokio::spawn(async move {
tokio::time::sleep(IDLE_TIMEOUT).await;
let _ = command.send(DeviceCommand::StopIfIdle { generation }).await;
});
}
async fn start_process(&mut self, channel: ConfigChannel) -> Result<()> {
self.generation = self.generation.wrapping_add(1);
let generation = self.generation;
let broadcaster = self.broadcaster.clone().unwrap_or_else(|| {
let (broadcaster, _) = broadcast::channel(STREAM_CHANNEL_CAPACITY);
broadcaster
});
let (process, output, display) = spawn_tuner(&self.config, &channel).await?;
self.current_channel = Some(channel);
self.command_display = Some(display);
self.process = Some(process);
self.broadcaster = Some(broadcaster.clone());
self.available = true;
let command = self.command_sender.clone();
tokio::spawn(async move {
read_output(output, broadcaster).await;
let _ = command
.send(DeviceCommand::OutputEnded { generation })
.await;
});
Ok(())
}
async fn output_ended(&mut self) {
let was_running = self.process.is_some();
if let Some(mut process) = self.process.take() {
process.wait().await;
}
self.command_display = None;
self.available = false;
if !was_running {
return;
}
if self.users.is_empty() {
self.release();
return;
}
self.fatal_count = self.fatal_count.saturating_add(1);
if self.fatal_count >= 3 {
self.fault = true;
self.users.clear();
self.broadcaster = None;
self.current_channel = None;
return;
}
let Some(channel) = self.current_channel.clone() else {
self.release();
return;
};
tracing::warn!(device = self.index, "respawning tuner after unexpected EOF");
if let Err(error) = self.start_process(channel).await {
tracing::error!(device = self.index, %error, "failed to respawn tuner");
self.users.clear();
self.broadcaster = None;
self.fault = true;
}
}
async fn stop_process(&mut self) -> Result<()> {
self.generation = self.generation.wrapping_add(1);
self.available = false;
if let Some(mut process) = self.process.take() {
process.terminate().await;
}
self.release();
Ok(())
}
fn release(&mut self) {
self.command_display = None;
self.current_channel = None;
self.broadcaster = None;
self.fatal_count = 0;
self.available = !self.fault;
}
fn status(&self) -> TunerDevice {
TunerDevice {
index: self.index,
name: self.config.name.clone(),
types: self.config.types.clone(),
command: self.command_display.clone(),
pid: self.process.as_ref().and_then(ProcessGroup::pid),
users: self.users.values().cloned().collect(),
is_available: self.available,
is_remote: self.config.remote_mirakurun_host.is_some(),
is_free: self.available && self.current_channel.is_none() && self.users.is_empty(),
is_using: self.available && self.current_channel.is_some() && !self.users.is_empty(),
is_fault: self.fault,
}
}
fn snapshot(&self) -> DeviceSnapshot {
DeviceSnapshot {
index: self.index,
types: self.config.types.clone(),
channel: self.current_channel.clone(),
users: self.users.len(),
priority: self
.users
.values()
.map(|user| user.priority)
.max()
.unwrap_or(-2),
available: self.available,
fault: self.fault,
}
}
}
#[derive(Debug)]
struct DeviceSnapshot {
index: usize,
types: Vec<ChannelType>,
channel: Option<ConfigChannel>,
users: usize,
priority: i32,
available: bool,
fault: bool,
}
#[derive(Debug)]
struct ProcessGroup {
children: Vec<Child>,
}
impl ProcessGroup {
fn pid(&self) -> Option<u32> {
self.children.first().and_then(Child::id)
}
async fn terminate(&mut self) {
for child in &mut self.children {
if let Some(id) = child.id() {
if let Ok(pid) = i32::try_from(id) {
let _ = kill(Pid::from_raw(pid), Signal::SIGTERM);
}
}
}
let deadline = tokio::time::Instant::now() + TERMINATE_TIMEOUT;
for child in &mut self.children {
let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
if tokio::time::timeout(remaining, child.wait()).await.is_err() {
let _ = child.start_kill();
let _ = child.wait().await;
}
}
}
async fn wait(&mut self) {
for child in &mut self.children {
let _ = child.wait().await;
}
}
}
async fn spawn_tuner(
config: &ConfigTuner,
channel: &ConfigChannel,
) -> Result<(ProcessGroup, TunerOutput, String)> {
if let Some(host) = &config.remote_mirakurun_host {
spawn_remote_tuner(config, channel, host)
.map(|(process, output, display)| (process, TunerOutput::Process(output), display))
} else {
spawn_local_tuner(config, channel).await
}
}
async fn spawn_local_tuner(
config: &ConfigTuner,
channel: &ConfigChannel,
) -> Result<(ProcessGroup, TunerOutput, String)> {
let template = config
.command
.as_deref()
.ok_or_else(|| CoreError::InvalidConfig("tuner command is missing".into()))?;
let display = replace_command_template(template, channel);
let (program, args) = parse_command(&display)?;
let use_dvb_device = config.dvb_device_path.is_some();
let mut command = Command::new(&program);
command
.args(args)
.stdout(if use_dvb_device {
Stdio::null()
} else {
Stdio::piped()
})
.stderr(Stdio::inherit())
.kill_on_drop(true);
let mut tuner = command.spawn().map_err(|source| CoreError::Read {
kind: "tuner process",
path: program.clone().into(),
source,
})?;
if let Some(device_path) = &config.dvb_device_path {
let output =
tokio::fs::File::open(device_path)
.await
.map_err(|source| CoreError::Read {
kind: "DVB device",
path: device_path.into(),
source,
})?;
Ok((
ProcessGroup {
children: vec![tuner],
},
TunerOutput::Dvb(output),
display,
))
} else {
let output = tuner
.stdout
.take()
.ok_or_else(|| CoreError::InvalidConfig("tuner did not expose stdout".into()))?;
Ok((
ProcessGroup {
children: vec![tuner],
},
TunerOutput::Process(output),
display,
))
}
}
fn spawn_remote_tuner(
config: &ConfigTuner,
channel: &ConfigChannel,
host: &str,
) -> Result<(ProcessGroup, ChildStdout, String)> {
let executable = std::env::current_exe().map_err(|source| CoreError::Read {
kind: "current executable",
path: "<current executable>".into(),
source,
})?;
let port = config.remote_mirakurun_port.unwrap_or(40_772);
let mut command = Command::new(&executable);
command
.arg("remote")
.arg(host)
.arg(port.to_string())
.arg(channel.channel_type.as_str())
.arg(&channel.channel);
if config.remote_mirakurun_decoder == Some(true) {
command.arg("--decode");
}
command
.stdout(Stdio::piped())
.stderr(Stdio::inherit())
.kill_on_drop(true);
let decode_argument = if config.remote_mirakurun_decoder == Some(true) {
" --decode"
} else {
""
};
let display = format!(
"{} remote {} {} {} {}{}",
executable.display(),
host,
port,
channel.channel_type.as_str(),
channel.channel,
decode_argument
);
let mut child = command.spawn().map_err(|source| CoreError::Read {
kind: "remote tuner process",
path: executable,
source,
})?;
let output = child
.stdout
.take()
.ok_or_else(|| CoreError::InvalidConfig("remote tuner did not expose stdout".into()))?;
Ok((
ProcessGroup {
children: vec![child],
},
output,
display,
))
}
enum TunerOutput {
Process(ChildStdout),
Dvb(tokio::fs::File),
}
async fn read_output(output: TunerOutput, broadcaster: broadcast::Sender<Bytes>) {
match output {
TunerOutput::Process(output) => broadcast_output(output, broadcaster).await,
TunerOutput::Dvb(output) => broadcast_output(output, broadcaster).await,
}
}
async fn broadcast_output<R>(mut output: R, broadcaster: broadcast::Sender<Bytes>)
where
R: AsyncRead + Unpin,
{
let mut buffer = vec![0_u8; 32 * 1024];
loop {
match output.read(&mut buffer).await {
Ok(0) => break,
Ok(length) => {
if broadcaster
.send(Bytes::copy_from_slice(&buffer[..length]))
.is_err()
{
tokio::task::yield_now().await;
}
}
Err(error) => {
tracing::warn!(%error, "failed to read tuner output");
break;
}
}
}
}
fn replace_command_template(template: &str, channel: &ConfigChannel) -> String {
let mut variables: HashMap<&str, String> = HashMap::new();
variables.insert("channel", channel.channel.clone());
variables.insert("type", channel.channel_type.as_str().into());
variables.insert(
"satelite",
channel
.command_vars
.as_ref()
.and_then(|vars| vars.get("satellite"))
.map_or_else(String::new, ToString::to_string),
);
variables.insert("space", "0".into());
if let Some(command_vars) = &channel.command_vars {
for (key, value) in command_vars {
variables.insert(key, command_variable_to_string(value));
}
}
let mut result = String::with_capacity(template.len());
let mut rest = template;
while let Some(start) = rest.find('<') {
result.push_str(&rest[..start]);
let after_start = &rest[start + 1..];
let Some(end) = after_start.find('>') else {
result.push_str(&rest[start..]);
return result;
};
let key = &after_start[..end];
result.push_str(variables.get(key).map_or("", String::as_str));
rest = &after_start[end + 1..];
}
result.push_str(rest);
result
}
fn command_variable_to_string(value: &CommandVariable) -> String {
value.to_string()
}
fn parse_command(command: &str) -> Result<(String, Vec<String>)> {
let mut in_quote = false;
let mut quote = '\0';
let mut current = String::new();
let mut parts = Vec::new();
for character in command.chars() {
if matches!(character, '"' | '\'') && (!in_quote || character == quote) {
in_quote = !in_quote;
quote = if in_quote { character } else { '\0' };
} else if character.is_ascii_whitespace() && !in_quote {
if !current.is_empty() {
parts.push(std::mem::take(&mut current));
}
} else {
current.push(character);
}
}
if in_quote {
return Err(CoreError::InvalidConfig(
"tuner command has an unterminated quote".into(),
));
}
if !current.is_empty() {
parts.push(current);
}
let mut parts = parts.into_iter();
let program = parts
.next()
.ok_or_else(|| CoreError::InvalidConfig("tuner command is empty".into()))?;
Ok((program, parts.collect()))
}
#[cfg(test)]
mod tests {
use std::collections::BTreeMap;
use mirakurun_types::{ChannelType, CommandVariable, ConfigChannel};
use super::{parse_command, replace_command_template};
#[test]
fn command_parser_preserves_quoted_arguments() {
let (program, args) =
parse_command("command --name 'Tokyo MX' \"two words\"").expect("parse command");
assert_eq!(program, "command");
assert_eq!(args, ["--name", "Tokyo MX", "two words"]);
}
#[test]
fn command_template_uses_channel_and_custom_variables() {
let channel = ConfigChannel {
name: "test".into(),
channel_type: ChannelType::Bs,
channel: "BS01_0".into(),
service_id: None,
tsmf_rel_ts: None,
command_vars: Some(BTreeMap::from([(
"satellite".into(),
CommandVariable::String("JCSAT".into()),
)])),
is_disabled: None,
satelite: None,
satellite: None,
space: None,
freq: None,
polarity: None,
};
assert_eq!(
replace_command_template("rec <type> <channel> <satelite>", &channel),
"rec BS BS01_0 JCSAT"
);
}
}