github-mirrors/netris-nestri
1
0
Fork 0
mirror of https://github.com/nestriness/nestri.git synced 2025-12-11 00:05:36 +02:00
Code Issues Packages Projects Releases Wiki Activity

⭐feat(runner): Improve robustness and argument handling (#285)

Browse Source 
## Description
Made argument parsing and handling much nicer with clap features.
Changed to tracing package for logging and made other improvements
around to hopefully make things more robust and logical.

Default audio-capture-method is now PipeWire since it seems to work
perfectly fine with latest gstreamer 🎉

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

- **New Features**
- Improved command-line argument parsing with stricter validation, type
safety, and clearer help messages.
- Enhanced GPU selection and logging, including explicit GPU info
logging and support for negative GPU indices for auto-selection.
- Added support for new audio and video codec and encoder enums,
providing safer and more flexible codec handling.

- **Bug Fixes**
- Improved error handling and logging throughout the application,
unifying logs under the `tracing` system for better diagnostics.
- Fixed issues with directory ownership and environment variable
handling in startup scripts.

- **Refactor**
- Replaced string-based parsing and manual conversions with strongly
typed enums and value parsers.
- Updated logging from `println!` and `log` macros to the `tracing`
crate for consistency.
- Simplified and unified the handling of pipeline and element references
in the signaling and data channel logic.

- **Chores**
- Updated and cleaned up dependencies, including switching from `log` to
`tracing` and upgrading the `webrtc` crate.
- Removed unused or redundant code and environment variables for
improved maintainability.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Co-authored-by: DatCaptainHorse <DatCaptainHorse@users.noreply.github.com>
This commit is contained in:
Kristian Ollikainen
2025-05-23 11:33:40 +03:00
committed by GitHub
parent d7e6da12ac
commit ae364f69bd
17 changed files with 653 additions and 617 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
containers/runner.Containerfile
View File

@@ -168,8 +168,7 @@ ENV USER="nestri" \
USER_PWD="nestri1234" \
XDG_RUNTIME_DIR=/run/user/1000 \
HOME=/home/nestri \
NVIDIA_DRIVER_CAPABILITIES=all \
NVIDIA_VISIBLE_DEVICES=all
NVIDIA_DRIVER_CAPABILITIES=all
RUN mkdir -p /home/${USER} && \
groupadd -g ${GID} ${USER} && \

26
packages/scripts/entrypoint.sh
View File

@@ -10,6 +10,16 @@ log() {
echo "[$(date +'%Y-%m-%d %H:%M:%S')] $1"
}
# Ensures user directory ownership
chown_user_directory() {
local user_group="${USER}:${GID}"
if ! chown -R -h --no-preserve-root "$user_group" "${HOME}" 2>/dev/null; then
echo "Error: Failed to change ownership of ${HOME} to ${user_group}" >&2
return 1
fi
return 0
}
# Waits for a given socket to be ready
wait_for_socket() {
local socket_path="$1"
@@ -110,6 +120,13 @@ install_nvidia_driver() {
return 0
}
function log_gpu_info {
log "Detected GPUs:"
for vendor in "${!vendor_devices[@]}"; do
log "> $vendor: ${vendor_devices[$vendor]}"
done
}
main() {
# Wait for required sockets
wait_for_socket "/run/dbus/system_bus_socket" "DBus" || exit 1
@@ -126,10 +143,11 @@ main() {
log "Error: Failed to detect GPU information."
exit 1
}
log_gpu_info
# Handle NVIDIA GPU
if [[ -n "${vendor_devices[nvidia]:-}" ]]; then
log "NVIDIA GPU detected, applying driver fix..."
log "NVIDIA GPU(s) detected, applying driver fix..."
# Determine NVIDIA driver version
local nvidia_driver_version=""
@@ -180,8 +198,6 @@ main() {
fi
}
fi
else
log "No NVIDIA GPU detected, skipping driver fix."
fi
# Make sure gamescope has CAP_SYS_NICE capabilities if available
@@ -195,6 +211,10 @@ main() {
log "Skipping CAP_SYS_NICE for gamescope, capability not available..."
fi
# Handle user directory permissions
log "Ensuring user directory permissions..."
chown_user_directory || exit 1
# Switch to nestri user
log "Switching to nestri user for application startup..."
if [[ ! -x /etc/nestri/entrypoint_nestri.sh ]]; then

11
packages/scripts/entrypoint_nestri.sh
View File

@@ -5,16 +5,6 @@ log() {
echo "[$(date +'%Y-%m-%d %H:%M:%S')] $1"
}
# Ensures user directory ownership
chown_user_directory() {
local user_group="$(id -nu):$(id -ng)"
chown -f "$user_group" ~ 2>/dev/null ||
sudo chown -f "$user_group" ~ 2>/dev/null ||
chown -R -f -h --no-preserve-root "$user_group" ~ 2>/dev/null ||
sudo chown -R -f -h --no-preserve-root "$user_group" ~ 2>/dev/null ||
log "Warning: Failed to change user directory permissions, there may be permission issues, continuing..."
}
# Parses resolution string
parse_resolution() {
local resolution="$1"
@@ -156,7 +146,6 @@ main_loop() {
}
main() {
chown_user_directory
load_envs
parse_resolution "${RESOLUTION:-1920x1080}" || exit 1
restart_chain

3
packages/scripts/envs.sh
View File

@@ -11,6 +11,3 @@ export PROTON_NO_FSYNC=1
# Sleeker Mangohud preset :)
export MANGOHUD_CONFIG=preset=2
# Our preferred prefix
export WINEPREFIX=/home/${USER}/.nestripfx/

529
packages/server/Cargo.lock generated
View File

File diff suppressed because it is too large Load Diff

9
packages/server/Cargo.toml
View File

@@ -10,17 +10,18 @@ path = "src/main.rs"
[dependencies]
gst = { package = "gstreamer", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs", branch = "main", features = ["v1_26"] }
gst-webrtc = { package = "gstreamer-webrtc", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs", branch = "main", features = ["v1_26"] }
gstrswebrtc = { package = "gst-plugin-webrtc", git = "https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs", branch = "main", features = ["v1_22"] }
gstrswebrtc = { package = "gst-plugin-webrtc", git = "https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs", branch = "main" }
serde = {version = "1.0", features = ["derive"] }
tokio = { version = "1.44", features = ["full"] }
clap = { version = "4.5", features = ["env"] }
serde_json = "1.0"
webrtc = "0.12"
webrtc = "0.13"
regex = "1.11"
rand = "0.9"
rustls = { version = "0.23", features = ["ring"] }
tokio-tungstenite = { version = "0.26", features = ["native-tls"] }
log = { version = "0.4", features = ["std"] }
tracing = "0.1"
tracing-subscriber = "0.3"
chrono = "0.4"
futures-util = "0.3"
num-derive = "0.4"
@@ -28,4 +29,4 @@ num-traits = "0.2"
prost = "0.13"
prost-types = "0.13"
parking_lot = "0.12"
atomic_refcell = "0.1"
atomic_refcell = "0.1"

55
packages/server/src/args.rs
View File

@@ -1,4 +1,7 @@
use clap::{Arg, Command};
use crate::args::encoding_args::AudioCaptureMethod;
use crate::enc_helper::{AudioCodec, EncoderType, VideoCodec};
use clap::{Arg, Command, value_parser};
use clap::builder::{BoolishValueParser, NonEmptyStringValueParser};
pub mod app_args;
pub mod device_args;
@@ -19,6 +22,7 @@ impl Args {
.long("verbose")
.env("VERBOSE")
.help("Enable verbose output")
.value_parser(BoolishValueParser::new())
.default_value("false"),
)
.arg(
@@ -26,7 +30,8 @@ impl Args {
.short('d')
.long("debug")
.env("DEBUG")
.help("Enable additional debugging information and features")
.help("Enable additional debugging features")
.value_parser(BoolishValueParser::new())
.default_value("false"),
)
.arg(
@@ -34,6 +39,7 @@ impl Args {
.short('u')
.long("relay-url")
.env("RELAY_URL")
.value_parser(NonEmptyStringValueParser::new())
.help("Nestri relay URL"),
)
.arg(
@@ -42,6 +48,7 @@ impl Args {
.long("resolution")
.env("RESOLUTION")
.help("Display/stream resolution in 'WxH' format")
.value_parser(NonEmptyStringValueParser::new())
.default_value("1280x720"),
)
.arg(
@@ -50,6 +57,7 @@ impl Args {
.long("framerate")
.env("FRAMERATE")
.help("Display/stream framerate")
.value_parser(value_parser!(u32).range(5..240))
.default_value("60"),
)
.arg(
@@ -63,7 +71,7 @@ impl Args {
.short('g')
.long("gpu-vendor")
.env("GPU_VENDOR")
.help("GPU to find by vendor (e.g. 'nvidia')")
.help("GPU to use by vendor")
.required(false),
)
.arg(
@@ -71,7 +79,7 @@ impl Args {
.short('n')
.long("gpu-name")
.env("GPU_NAME")
.help("GPU to find by name (e.g. 'rtx 3060')")
.help("GPU to use by name")
.required(false),
)
.arg(
@@ -79,14 +87,15 @@ impl Args {
.short('i')
.long("gpu-index")
.env("GPU_INDEX")
.help("GPU index, if multiple similar GPUs are present")
.default_value("0"),
.help("GPU to use by index")
.value_parser(value_parser!(i32).range(-1..))
.default_value("-1")
)
.arg(
Arg::new("gpu-card-path")
.long("gpu-card-path")
.env("GPU_CARD_PATH")
.help("Force a specific GPU by card/render path (e.g. '/dev/dri/card0')")
.help("Force a specific GPU by /dev/dri/ card or render path")
.required(false)
.conflicts_with_all(["gpu-vendor", "gpu-name", "gpu-index"]),
)
@@ -95,27 +104,30 @@ impl Args {
.short('c')
.long("video-codec")
.env("VIDEO_CODEC")
.help("Preferred video codec ('h264', 'h265', 'av1')")
.help("Preferred video codec")
.value_parser(value_parser!(VideoCodec))
.default_value("h264"),
)
.arg(
Arg::new("video-encoder")
.long("video-encoder")
.env("VIDEO_ENCODER")
.help("Override video encoder (e.g. 'vah264enc')"),
.help("Override video encoder"),
)
.arg(
Arg::new("video-rate-control")
.long("video-rate-control")
.env("VIDEO_RATE_CONTROL")
.help("Rate control method ('cqp', 'vbr', 'cbr')")
.default_value("vbr"),
.help("Rate control method")
.value_parser(value_parser!(encoding_args::RateControlMethod))
.default_value("cbr"),
)
.arg(
Arg::new("video-cqp")
.long("video-cqp")
.env("VIDEO_CQP")
.help("Constant Quantization Parameter (CQP) quality")
.value_parser(value_parser!(u32).range(1..51))
.default_value("26"),
)
.arg(
@@ -123,6 +135,7 @@ impl Args {
.long("video-bitrate")
.env("VIDEO_BITRATE")
.help("Target bitrate in kbps")
.value_parser(value_parser!(u32).range(1..))
.default_value("6000"),
)
.arg(
@@ -130,27 +143,31 @@ impl Args {
.long("video-bitrate-max")
.env("VIDEO_BITRATE_MAX")
.help("Maximum bitrate in kbps")
.value_parser(value_parser!(u32).range(1..))
.default_value("8000"),
)
.arg(
Arg::new("video-encoder-type")
.long("video-encoder-type")
.env("VIDEO_ENCODER_TYPE")
.help("Encoder type ('hardware', 'software')")
.help("Encoder type")
.value_parser(value_parser!(EncoderType))
.default_value("hardware"),
)
.arg(
Arg::new("audio-capture-method")
.long("audio-capture-method")
.env("AUDIO_CAPTURE_METHOD")
.help("Audio capture method ('pipewire', 'pulseaudio', 'alsa')")
.default_value("pulseaudio"),
.help("Audio capture method")
.value_parser(value_parser!(AudioCaptureMethod))
.default_value("pipewire"),
)
.arg(
Arg::new("audio-codec")
.long("audio-codec")
.env("AUDIO_CODEC")
.help("Preferred audio codec ('opus', 'aac')")
.help("Preferred audio codec")
.value_parser(value_parser!(AudioCodec))
.default_value("opus"),
)
.arg(
@@ -163,14 +180,16 @@ impl Args {
Arg::new("audio-rate-control")
.long("audio-rate-control")
.env("AUDIO_RATE_CONTROL")
.help("Rate control method ('cqp', 'vbr', 'cbr')")
.default_value("vbr"),
.help("Rate control method")
.value_parser(value_parser!(encoding_args::RateControlMethod))
.default_value("cbr"),
)
.arg(
Arg::new("audio-bitrate")
.long("audio-bitrate")
.env("AUDIO_BITRATE")
.help("Target bitrate in kbps")
.value_parser(value_parser!(u32).range(1..))
.default_value("128"),
)
.arg(
@@ -178,6 +197,7 @@ impl Args {
.long("audio-bitrate-max")
.env("AUDIO_BITRATE_MAX")
.help("Maximum bitrate in kbps")
.value_parser(value_parser!(u32).range(1..))
.default_value("192"),
)
.arg(
@@ -185,6 +205,7 @@ impl Args {
.long("dma-buf")
.env("DMA_BUF")
.help("Use DMA-BUF for pipeline")
.value_parser(BoolishValueParser::new())
.default_value("false"),
)
.get_matches();

31
packages/server/src/args/app_args.rs
View File

@@ -20,10 +20,8 @@ pub struct AppArgs {
impl AppArgs {
pub fn from_matches(matches: &clap::ArgMatches) -> Self {
Self {
verbose: matches.get_one::<String>("verbose").unwrap() == "true"
|| matches.get_one::<String>("verbose").unwrap() == "1",
debug: matches.get_one::<String>("debug").unwrap() == "true"
|| matches.get_one::<String>("debug").unwrap() == "1",
verbose: matches.get_one::<bool>("verbose").unwrap_or(&false).clone(),
debug: matches.get_one::<bool>("debug").unwrap_or(&false).clone(),
resolution: {
let res = matches
.get_one::<String>("resolution")
@@ -39,11 +37,7 @@ impl AppArgs {
(1280, 720)
}
},
framerate: matches
.get_one::<String>("framerate")
.unwrap()
.parse::<u32>()
.unwrap_or(60),
framerate: matches.get_one::<u32>("framerate").unwrap_or(&60).clone(),
relay_url: matches
.get_one::<String>("relay-url")
.expect("relay url cannot be empty")
@@ -53,19 +47,18 @@ impl AppArgs {
.get_one::<String>("room")
.unwrap_or(&rand::random::<u32>().to_string())
.clone(),
dma_buf: matches.get_one::<String>("dma-buf").unwrap() == "true"
|| matches.get_one::<String>("dma-buf").unwrap() == "1",
dma_buf: matches.get_one::<bool>("dma-buf").unwrap_or(&false).clone(),
}
}
pub fn debug_print(&self) {
println!("AppArgs:");
println!("> verbose: {}", self.verbose);
println!("> debug: {}", self.debug);
println!("> resolution: {}x{}", self.resolution.0, self.resolution.1);
println!("> framerate: {}", self.framerate);
println!("> relay_url: {}", self.relay_url);
println!("> room: {}", self.room);
println!("> dma_buf: {}", self.dma_buf);
tracing::info!("AppArgs:");
tracing::info!("> verbose: {}", self.verbose);
tracing::info!("> debug: {}", self.debug);
tracing::info!("> resolution: '{}x{}'", self.resolution.0, self.resolution.1);
tracing::info!("> framerate: {}", self.framerate);
tracing::info!("> relay_url: '{}'", self.relay_url);
tracing::info!("> room: '{}'", self.room);
tracing::info!("> dma_buf: {}", self.dma_buf);
}
}

28
packages/server/src/args/device_args.rs
View File

@@ -3,8 +3,8 @@ pub struct DeviceArgs {
pub gpu_vendor: String,
/// GPU name (e.g. "a770")
pub gpu_name: String,
/// GPU index, if multiple same GPUs are present
pub gpu_index: u32,
/// GPU index, if multiple same GPUs are present, -1 for auto-selection
pub gpu_index: i32,
/// GPU card/render path, sets card explicitly from such path
pub gpu_card_path: String,
}
@@ -20,10 +20,9 @@ impl DeviceArgs {
.unwrap_or(&"".to_string())
.clone(),
gpu_index: matches
.get_one::<String>("gpu-index")
.unwrap()
.parse::<u32>()
.unwrap(),
.get_one::<i32>("gpu-index")
.unwrap_or(&-1)
.clone(),
gpu_card_path: matches
.get_one::<String>("gpu-card-path")
.unwrap_or(&"".to_string())
@@ -32,17 +31,10 @@ impl DeviceArgs {
}
pub fn debug_print(&self) {
println!("DeviceArgs:");
println!("> gpu_vendor: {}", self.gpu_vendor);
println!("> gpu_name: {}", self.gpu_name);
println!("> gpu_index: {}", self.gpu_index);
println!(
"> gpu_card_path: {}",
if self.gpu_card_path.is_empty() {
"Auto-Selection"
} else {
&self.gpu_card_path
}
);
tracing::info!("DeviceArgs:");
tracing::info!("> gpu_vendor: '{}'", self.gpu_vendor);
tracing::info!("> gpu_name: '{}'", self.gpu_name);
tracing::info!("> gpu_index: {}", self.gpu_index);
tracing::info!("> gpu_card_path: '{}'", self.gpu_card_path);
}
}

188
packages/server/src/args/encoding_args.rs
View File

@@ -1,4 +1,8 @@
use crate::enc_helper::Codec::{Audio, Video};
use crate::enc_helper::{AudioCodec, Codec, EncoderType, VideoCodec};
use clap::ValueEnum;
use std::ops::Deref;
use std::str::FromStr;
#[derive(Debug, PartialEq, Eq, Clone)]
pub struct RateControlCQP {
@@ -8,14 +12,42 @@ pub struct RateControlCQP {
#[derive(Debug, PartialEq, Eq, Clone)]
pub struct RateControlVBR {
/// Target bitrate in kbps
pub target_bitrate: i32,
pub target_bitrate: u32,
/// Maximum bitrate in kbps
pub max_bitrate: i32,
pub max_bitrate: u32,
}
#[derive(Debug, PartialEq, Eq, Clone)]
pub struct RateControlCBR {
/// Target bitrate in kbps
pub target_bitrate: i32,
pub target_bitrate: u32,
}
#[derive(Debug, PartialEq, Eq, Clone, ValueEnum)]
pub enum RateControlMethod {
CQP,
VBR,
CBR,
}
impl RateControlMethod {
pub fn as_str(&self) -> &str {
match self {
RateControlMethod::CQP => "cqp",
RateControlMethod::VBR => "vbr",
RateControlMethod::CBR => "cbr",
}
}
}
impl FromStr for RateControlMethod {
type Err = String;
fn from_str(s: &str) -> Result<Self, Self::Err> {
match s {
"cqp" => Ok(RateControlMethod::CQP),
"vbr" => Ok(RateControlMethod::VBR),
"cbr" => Ok(RateControlMethod::CBR),
_ => Err(format!("Invalid rate control method: {}", s)),
}
}
}
#[derive(Debug, PartialEq, Eq, Clone)]
@@ -30,7 +62,7 @@ pub enum RateControl {
pub struct EncodingOptionsBase {
/// Codec (e.g. "h264", "opus" etc.)
pub codec: String,
pub codec: Codec,
/// Overridable encoder (e.g. "vah264lpenc", "opusenc" etc.)
pub encoder: String,
/// Rate control method (e.g. "cqp", "vbr", "cbr")
@@ -38,28 +70,21 @@ pub struct EncodingOptionsBase {
}
impl EncodingOptionsBase {
pub fn debug_print(&self) {
println!("> Codec: {}", self.codec);
println!(
"> Encoder: {}",
if self.encoder.is_empty() {
"Auto-Selection"
} else {
&self.encoder
}
);
tracing::info!("> Codec: '{}'", self.codec.as_str());
tracing::info!("> Encoder: '{}'", self.encoder);
match &self.rate_control {
RateControl::CQP(cqp) => {
println!("> Rate Control: CQP");
println!("-> Quality: {}", cqp.quality);
tracing::info!("> Rate Control: CQP");
tracing::info!("-> Quality: {}", cqp.quality);
}
RateControl::VBR(vbr) => {
println!("> Rate Control: VBR");
println!("-> Target Bitrate: {}", vbr.target_bitrate);
println!("-> Max Bitrate: {}", vbr.max_bitrate);
tracing::info!("> Rate Control: VBR");
tracing::info!("-> Target Bitrate: {}", vbr.target_bitrate);
tracing::info!("-> Max Bitrate: {}", vbr.max_bitrate);
}
RateControl::CBR(cbr) => {
println!("> Rate Control: CBR");
println!("-> Target Bitrate: {}", cbr.target_bitrate);
tracing::info!("> Rate Control: CBR");
tracing::info!("-> Target Bitrate: {}", cbr.target_bitrate);
}
}
}
@@ -67,63 +92,62 @@ impl EncodingOptionsBase {
pub struct VideoEncodingOptions {
pub base: EncodingOptionsBase,
/// Encoder type (e.g. "hardware", "software")
pub encoder_type: String,
pub encoder_type: EncoderType,
}
impl VideoEncodingOptions {
pub fn from_matches(matches: &clap::ArgMatches) -> Self {
Self {
base: EncodingOptionsBase {
codec: matches.get_one::<String>("video-codec").unwrap().clone(),
codec: Video(
matches
.get_one::<VideoCodec>("video-codec")
.unwrap_or(&VideoCodec::H264)
.clone(),
),
encoder: matches
.get_one::<String>("video-encoder")
.unwrap_or(&"".to_string())
.clone(),
rate_control: match matches
.get_one::<String>("video-rate-control")
.unwrap()
.as_str()
.get_one::<RateControlMethod>("video-rate-control")
.unwrap_or(&RateControlMethod::CBR)
{
"cqp" => RateControl::CQP(RateControlCQP {
RateControlMethod::CQP => RateControl::CQP(RateControlCQP {
quality: matches
.get_one::<String>("video-cqp")
.unwrap()
.parse::<u32>()
.unwrap(),
}),
"cbr" => RateControl::CBR(RateControlCBR {
RateControlMethod::CBR => RateControl::CBR(RateControlCBR {
target_bitrate: matches
.get_one::<String>("video-bitrate")
.get_one::<u32>("video-bitrate")
.unwrap()
.parse::<i32>()
.unwrap(),
.clone(),
}),
"vbr" => RateControl::VBR(RateControlVBR {
RateControlMethod::VBR => RateControl::VBR(RateControlVBR {
target_bitrate: matches
.get_one::<String>("video-bitrate")
.get_one::<u32>("video-bitrate")
.unwrap()
.parse::<i32>()
.unwrap(),
.clone(),
max_bitrate: matches
.get_one::<String>("video-bitrate-max")
.get_one::<u32>("video-bitrate-max")
.unwrap()
.parse::<i32>()
.unwrap(),
.clone(),
}),
_ => panic!("Invalid rate control method for video"),
},
},
encoder_type: matches
.get_one::<String>("video-encoder-type")
.unwrap_or(&"hardware".to_string())
.get_one::<EncoderType>("video-encoder-type")
.unwrap_or(&EncoderType::HARDWARE)
.clone(),
}
}
pub fn debug_print(&self) {
println!("Video Encoding Options:");
tracing::info!("Video Encoding Options:");
self.base.debug_print();
println!("> Encoder Type: {}", self.encoder_type);
tracing::info!("> Encoder Type: {}", self.encoder_type.as_str());
}
}
impl Deref for VideoEncodingOptions {
@@ -134,18 +158,30 @@ impl Deref for VideoEncodingOptions {
}
}
#[derive(Debug, PartialEq, Eq)]
#[derive(Debug, PartialEq, Eq, Clone, ValueEnum)]
pub enum AudioCaptureMethod {
PulseAudio,
PipeWire,
PULSEAUDIO,
PIPEWIRE,
ALSA,
}
impl AudioCaptureMethod {
pub fn as_str(&self) -> &str {
match self {
AudioCaptureMethod::PulseAudio => "pulseaudio",
AudioCaptureMethod::PipeWire => "pipewire",
AudioCaptureMethod::ALSA => "alsa",
AudioCaptureMethod::PULSEAUDIO => "PulseAudio",
AudioCaptureMethod::PIPEWIRE => "PipeWire",
AudioCaptureMethod::ALSA => "ALSA",
}
}
}
impl FromStr for AudioCaptureMethod {
type Err = String;
fn from_str(s: &str) -> Result<Self, Self::Err> {
match s.to_lowercase().as_str() {
"pulseaudio" => Ok(AudioCaptureMethod::PULSEAUDIO),
"pipewire" => Ok(AudioCaptureMethod::PIPEWIRE),
"alsa" => Ok(AudioCaptureMethod::ALSA),
_ => Err(format!("Invalid audio capture method: {}", s)),
}
}
}
@@ -158,56 +194,50 @@ impl AudioEncodingOptions {
pub fn from_matches(matches: &clap::ArgMatches) -> Self {
Self {
base: EncodingOptionsBase {
codec: matches.get_one::<String>("audio-codec").unwrap().clone(),
codec: Audio(
matches
.get_one::<AudioCodec>("audio-codec")
.unwrap_or(&AudioCodec::OPUS)
.clone(),
),
encoder: matches
.get_one::<String>("audio-encoder")
.unwrap_or(&"".to_string())
.clone(),
rate_control: match matches
.get_one::<String>("audio-rate-control")
.unwrap()
.as_str()
.get_one::<RateControlMethod>("audio-rate-control")
.unwrap_or(&RateControlMethod::CBR)
{
"cbr" => RateControl::CBR(RateControlCBR {
RateControlMethod::CBR => RateControl::CBR(RateControlCBR {
target_bitrate: matches
.get_one::<String>("audio-bitrate")
.get_one::<u32>("audio-bitrate")
.unwrap()
.parse::<i32>()
.unwrap(),
.clone(),
}),
"vbr" => RateControl::VBR(RateControlVBR {
RateControlMethod::VBR => RateControl::VBR(RateControlVBR {
target_bitrate: matches
.get_one::<String>("audio-bitrate")
.get_one::<u32>("audio-bitrate")
.unwrap()
.parse::<i32>()
.unwrap(),
.clone(),
max_bitrate: matches
.get_one::<String>("audio-bitrate-max")
.get_one::<u32>("audio-bitrate-max")
.unwrap()
.parse::<i32>()
.unwrap(),
.clone(),
}),
_ => panic!("Invalid rate control method for audio"),
wot => panic!("Invalid rate control method for audio: {}", wot.as_str()),
},
},
capture_method: match matches
.get_one::<String>("audio-capture-method")
.unwrap()
.as_str()
{
"pulseaudio" => AudioCaptureMethod::PulseAudio,
"pipewire" => AudioCaptureMethod::PipeWire,
"alsa" => AudioCaptureMethod::ALSA,
// Default to PulseAudio
_ => AudioCaptureMethod::PulseAudio,
},
capture_method: matches
.get_one::<AudioCaptureMethod>("audio-capture-method")
.unwrap_or(&AudioCaptureMethod::PIPEWIRE)
.clone(),
}
}
pub fn debug_print(&self) {
println!("Audio Encoding Options:");
tracing::info!("Audio Encoding Options:");
self.base.debug_print();
println!("> Capture Method: {}", self.capture_method.as_str());
tracing::info!("> Capture Method: {}", self.capture_method.as_str());
}
}
impl Deref for AudioEncodingOptions {
@@ -233,7 +263,7 @@ impl EncodingArgs {
}
pub fn debug_print(&self) {
println!("Encoding Arguments:");
tracing::info!("Encoding Arguments:");
self.video.debug_print();
self.audio.debug_print();
}

128
packages/server/src/enc_helper.rs
View File

@@ -1,31 +1,70 @@
use crate::args::encoding_args::RateControl;
use crate::gpu::{self, GPUInfo, get_gpu_by_card_path, get_gpus_by_vendor};
use clap::ValueEnum;
use gst::prelude::*;
use std::error::Error;
use std::str::FromStr;
#[derive(Debug, Eq, PartialEq, Clone)]
#[derive(Debug, Eq, PartialEq, Clone, ValueEnum)]
pub enum AudioCodec {
OPUS,
}
impl AudioCodec {
pub fn as_str(&self) -> &'static str {
match self {
Self::OPUS => "Opus",
}
}
}
impl FromStr for AudioCodec {
type Err = String;
fn from_str(s: &str) -> Result<Self, Self::Err> {
match s.to_lowercase().as_str() {
"opus" => Ok(Self::OPUS),
_ => Err(format!("Invalid audio codec: {}", s)),
}
}
}
#[derive(Debug, Eq, PartialEq, Clone, ValueEnum)]
pub enum VideoCodec {
H264,
H265,
AV1,
UNKNOWN,
}
impl VideoCodec {
pub fn to_str(&self) -> &'static str {
pub fn as_str(&self) -> &'static str {
match self {
Self::H264 => "H.264",
Self::H265 => "H.265",
Self::AV1 => "AV1",
Self::UNKNOWN => "Unknown",
}
}
}
impl FromStr for VideoCodec {
type Err = String;
pub fn from_str(s: &str) -> Self {
fn from_str(s: &str) -> Result<Self, Self::Err> {
match s.to_lowercase().as_str() {
"h264" | "h.264" | "avc" => Self::H264,
"h265" | "h.265" | "hevc" | "hev1" => Self::H265,
"av1" => Self::AV1,
_ => Self::UNKNOWN,
"h264" | "h.264" | "avc" => Ok(Self::H264),
"h265" | "h.265" | "hevc" | "hev1" => Ok(Self::H265),
"av1" => Ok(Self::AV1),
_ => Err(format!("Invalid video codec: {}", s)),
}
}
}
#[derive(Debug, Eq, PartialEq, Clone)]
pub enum Codec {
Audio(AudioCodec),
Video(VideoCodec),
}
impl Codec {
pub fn as_str(&self) -> &'static str {
match self {
Self::Audio(codec) => codec.as_str(),
Self::Video(codec) => codec.as_str(),
}
}
}
@@ -53,27 +92,17 @@ impl EncoderAPI {
}
}
#[derive(Debug, Eq, PartialEq, Clone)]
#[derive(Debug, Eq, PartialEq, Clone, ValueEnum)]
pub enum EncoderType {
SOFTWARE,
HARDWARE,
UNKNOWN,
}
impl EncoderType {
pub fn to_str(&self) -> &'static str {
pub fn as_str(&self) -> &'static str {
match self {
Self::SOFTWARE => "Software",
Self::HARDWARE => "Hardware",
Self::UNKNOWN => "Unknown",
}
}
pub fn from_str(s: &str) -> Self {
match s.to_lowercase().as_str() {
"software" => Self::SOFTWARE,
"hardware" => Self::HARDWARE,
_ => Self::UNKNOWN,
}
}
}
@@ -121,7 +150,7 @@ impl VideoEncoderInfo {
for (key, value) in &self.parameters {
if element.has_property(key) {
if verbose {
println!("Setting property {} to {}", key, value);
tracing::debug!("Setting property {} to {}", key, value);
}
element.set_property_from_str(key, value);
}
@@ -145,19 +174,15 @@ fn get_encoder_api(encoder: &str, encoder_type: &EncoderType) -> EncoderAPI {
}
}
EncoderType::SOFTWARE => EncoderAPI::SOFTWARE,
_ => EncoderAPI::UNKNOWN,
}
}
fn codec_from_encoder_name(name: &str) -> Option<VideoCodec> {
if name.contains("h264") {
Some(VideoCodec::H264)
} else if name.contains("h265") {
Some(VideoCodec::H265)
} else if name.contains("av1") {
Some(VideoCodec::AV1)
} else {
None
match name.to_lowercase() {
n if n.contains("h264") => Some(VideoCodec::H264),
n if n.contains("h265") => Some(VideoCodec::H265),
n if n.contains("av1") => Some(VideoCodec::AV1),
_ => None,
}
}
@@ -272,7 +297,6 @@ pub fn encoder_low_latency_params(
let usage = match encoder_optz.codec {
VideoCodec::H264 | VideoCodec::H265 => "ultra-low-latency",
VideoCodec::AV1 => "low-latency",
_ => "",
};
if !usage.is_empty() {
encoder_optz.set_parameter("usage", usage);
@@ -378,8 +402,8 @@ pub fn get_compatible_encoders() -> Vec<VideoEncoderInfo> {
}
})
.unwrap_or_else(|_| {
log::error!(
"Panic occurred while querying properties for {}",
tracing::error!(
"Error occurred while querying properties for {}",
encoder_name
);
None
@@ -401,16 +425,20 @@ pub fn get_compatible_encoders() -> Vec<VideoEncoderInfo> {
/// * `encoders` - A vector containing information about each encoder.
/// * `name` - A string slice that holds the encoder name.
/// # Returns
/// * `Option<EncoderInfo>` - A reference to an EncoderInfo struct if found.
/// * `Result<EncoderInfo, Box<dyn Error>>` - A Result containing EncoderInfo if found, or an error.
pub fn get_encoder_by_name(
encoders: &Vec<VideoEncoderInfo>,
name: &str,
) -> Option<VideoEncoderInfo> {
) -> Result<VideoEncoderInfo, Box<dyn Error>> {
let name = name.to_lowercase();
encoders
if let Some(encoder) = encoders
.iter()
.find(|encoder| encoder.name.to_lowercase() == name)
.cloned()
{
Ok(encoder.clone())
} else {
Err(format!("Encoder '{}' not found", name).into())
}
}
/// Helper to get encoders from vector by video codec.
@@ -453,15 +481,23 @@ pub fn get_encoders_by_type(
/// * `codec` - Desired codec.
/// * `encoder_type` - Desired encoder type.
/// # Returns
/// * `Option<EncoderInfo>` - Best-case compatible encoder.
/// * `Result<VideoEncoderInfo, Box<dyn Error>>` - A Result containing the best compatible encoder if found, or an error.
pub fn get_best_compatible_encoder(
encoders: &Vec<VideoEncoderInfo>,
codec: VideoCodec,
encoder_type: EncoderType,
) -> Option<VideoEncoderInfo> {
codec: &Codec,
encoder_type: &EncoderType,
) -> Result<VideoEncoderInfo, Box<dyn Error>> {
let mut best_encoder: Option<VideoEncoderInfo> = None;
let mut best_score: i32 = 0;
let codec = match codec {
Codec::Video(c) => c.clone(),
Codec::Audio(_) => {
// Only for video currently
return Err("Attempted to get best compatible video encoder with audio codec".into());
}
};
// Filter by codec and type first
let encoders = get_encoders_by_videocodec(encoders, &codec);
let encoders = get_encoders_by_type(&encoders, &encoder_type);
@@ -498,5 +534,9 @@ pub fn get_best_compatible_encoder(
}
}
best_encoder
if let Some(encoder) = best_encoder {
Ok(encoder)
} else {
Err("No compatible encoder found".into())
}
}

8
packages/server/src/gpu.rs
View File

@@ -161,3 +161,11 @@ pub fn get_gpu_by_card_path(gpus: &[GPUInfo], path: &str) -> Option<GPUInfo> {
})
.cloned()
}
pub fn get_gpu_by_index(gpus: &[GPUInfo], index: i32) -> Option<GPUInfo> {
if index < 0 || index as usize >= gpus.len() {
None
} else {
Some(gpus[index as usize].clone())
}
}

151
packages/server/src/main.rs
View File

@@ -8,6 +8,7 @@ mod proto;
mod websocket;
use crate::args::encoding_args;
use crate::enc_helper::EncoderType;
use crate::gpu::GPUVendor;
use crate::nestrisink::NestriSignaller;
use crate::websocket::NestriWebSocket;
@@ -20,16 +21,16 @@ use std::str::FromStr;
use std::sync::Arc;
// Handles gathering GPU information and selecting the most suitable GPU
fn handle_gpus(args: &args::Args) -> Option<gpu::GPUInfo> {
println!("Gathering GPU information..");
fn handle_gpus(args: &args::Args) -> Result<gpu::GPUInfo, Box<dyn Error>> {
tracing::info!("Gathering GPU information..");
let gpus = gpu::get_gpus();
if gpus.is_empty() {
println!("No GPUs found");
return None;
return Err("No GPUs found".into());
}
for gpu in &gpus {
println!(
"> [GPU] Vendor: '{}', Card Path: '{}', Render Path: '{}', Device Name: '{}'",
for (i, gpu) in gpus.iter().enumerate() {
tracing::info!(
"> [GPU:{}] Vendor: '{}', Card Path: '{}', Render Path: '{}', Device Name: '{}'",
i,
gpu.vendor_string(),
gpu.card_path(),
gpu.render_path(),
@@ -50,9 +51,12 @@ fn handle_gpus(args: &args::Args) -> Option<gpu::GPUInfo> {
if !args.device.gpu_name.is_empty() {
filtered_gpus = gpu::get_gpus_by_device_name(&filtered_gpus, &args.device.gpu_name);
}
if args.device.gpu_index != 0 {
if args.device.gpu_index > -1 {
// get single GPU by index
gpu = filtered_gpus.get(args.device.gpu_index as usize).cloned();
gpu = gpu::get_gpu_by_index(&filtered_gpus, args.device.gpu_index).or_else(|| {
tracing::warn!("GPU index {} is out of range", args.device.gpu_index);
None
});
} else {
// get first GPU
gpu = filtered_gpus
@@ -61,35 +65,33 @@ fn handle_gpus(args: &args::Args) -> Option<gpu::GPUInfo> {
}
}
if gpu.is_none() {
println!(
return Err(format!(
"No GPU found with the specified parameters: vendor='{}', name='{}', index='{}', card_path='{}'",
args.device.gpu_vendor,
args.device.gpu_name,
args.device.gpu_index,
args.device.gpu_card_path
);
return None;
).into());
}
let gpu = gpu.unwrap();
println!("Selected GPU: '{}'", gpu.device_name());
Some(gpu)
tracing::info!("Selected GPU: '{}'", gpu.device_name());
Ok(gpu)
}
// Handles picking video encoder
fn handle_encoder_video(args: &args::Args) -> Option<enc_helper::VideoEncoderInfo> {
println!("Getting compatible video encoders..");
fn handle_encoder_video(args: &args::Args) -> Result<enc_helper::VideoEncoderInfo, Box<dyn Error>> {
tracing::info!("Getting compatible video encoders..");
let video_encoders = enc_helper::get_compatible_encoders();
if video_encoders.is_empty() {
println!("No compatible video encoders found");
return None;
return Err("No compatible video encoders found".into());
}
for encoder in &video_encoders {
println!(
tracing::info!(
"> [Video Encoder] Name: '{}', Codec: '{}', API: '{}', Type: '{}', Device: '{}'",
encoder.name,
encoder.codec.to_str(),
encoder.codec.as_str(),
encoder.encoder_api.to_str(),
encoder.encoder_type.to_str(),
encoder.encoder_type.as_str(),
if let Some(gpu) = &encoder.gpu_info {
gpu.device_name()
} else {
@@ -101,26 +103,16 @@ fn handle_encoder_video(args: &args::Args) -> Option<enc_helper::VideoEncoderInf
let video_encoder;
if !args.encoding.video.encoder.is_empty() {
video_encoder =
enc_helper::get_encoder_by_name(&video_encoders, &args.encoding.video.encoder);
enc_helper::get_encoder_by_name(&video_encoders, &args.encoding.video.encoder)?;
} else {
video_encoder = enc_helper::get_best_compatible_encoder(
&video_encoders,
enc_helper::VideoCodec::from_str(&args.encoding.video.codec),
enc_helper::EncoderType::from_str(&args.encoding.video.encoder_type),
);
&args.encoding.video.codec,
&args.encoding.video.encoder_type,
)?;
}
if video_encoder.is_none() {
println!(
"No video encoder found with the specified parameters: name='{}', vcodec='{}', type='{}'",
args.encoding.video.encoder,
args.encoding.video.codec,
args.encoding.video.encoder_type
);
return None;
}
let video_encoder = video_encoder.unwrap();
println!("Selected video encoder: '{}'", video_encoder.name);
Some(video_encoder)
tracing::info!("Selected video encoder: '{}'", video_encoder.name);
Ok(video_encoder)
}
// Handles picking preferred settings for video encoder
@@ -141,16 +133,16 @@ fn handle_encoder_video_settings(
encoding_args::RateControl::VBR(vbr) => {
optimized_encoder = enc_helper::encoder_vbr_params(
&optimized_encoder,
vbr.target_bitrate as u32,
vbr.max_bitrate as u32,
vbr.target_bitrate,
vbr.max_bitrate,
);
}
encoding_args::RateControl::CBR(cbr) => {
optimized_encoder =
enc_helper::encoder_cbr_params(&optimized_encoder, cbr.target_bitrate as u32);
enc_helper::encoder_cbr_params(&optimized_encoder, cbr.target_bitrate);
}
}
println!(
tracing::info!(
"Selected video encoder settings: '{}'",
optimized_encoder.get_parameters_string()
);
@@ -165,16 +157,23 @@ fn handle_encoder_audio(args: &args::Args) -> String {
} else {
args.encoding.audio.encoder.clone()
};
println!("Selected audio encoder: '{}'", audio_encoder);
tracing::info!("Selected audio encoder: '{}'", audio_encoder);
audio_encoder
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
// Parse command line arguments
let args = args::Args::new();
let mut args = args::Args::new();
if args.app.verbose {
// Make sure tracing has INFO level
tracing_subscriber::fmt()
.with_max_level(tracing::Level::INFO)
.init();
args.debug_print();
} else {
tracing_subscriber::fmt::init();
}
rustls::crypto::ring::default_provider()
@@ -192,33 +191,39 @@ async fn main() -> Result<(), Box<dyn Error>> {
// Setup our websocket
let nestri_ws = Arc::new(NestriWebSocket::new(ws_url).await?);
log::set_max_level(log::LevelFilter::Info);
log::set_boxed_logger(Box::new(nestri_ws.clone())).unwrap();
gst::init()?;
gstrswebrtc::plugin_register_static()?;
// Handle GPU selection
let gpu = handle_gpus(&args);
if gpu.is_none() {
log::error!("Failed to find a suitable GPU. Exiting..");
return Err("Failed to find a suitable GPU. Exiting..".into());
}
let gpu = gpu.unwrap();
let gpu = match handle_gpus(&args) {
Ok(gpu) => gpu,
Err(e) => {
tracing::error!("Failed to find a suitable GPU: {}", e);
return Err(e);
}
};
if args.app.dma_buf {
log::warn!(
"DMA-BUF is experimental, it may or may not improve performance, or even work at all."
);
if args.encoding.video.encoder_type != EncoderType::HARDWARE {
tracing::warn!("DMA-BUF is only supported with hardware encoders, disabling DMA-BUF..");
args.app.dma_buf = false;
} else {
tracing::warn!(
"DMA-BUF is experimental, it may or may not improve performance, or even work at all."
);
}
}
// Handle video encoder selection
let video_encoder_info = handle_encoder_video(&args);
if video_encoder_info.is_none() {
log::error!("Failed to find a suitable video encoder. Exiting..");
return Err("Failed to find a suitable video encoder. Exiting..".into());
}
let mut video_encoder_info = video_encoder_info.unwrap();
let mut video_encoder_info = match handle_encoder_video(&args) {
Ok(encoder) => encoder,
Err(e) => {
tracing::error!("Failed to find a suitable video encoder: {}", e);
return Err(e);
}
};
// Handle video encoder settings
video_encoder_info = handle_encoder_video_settings(&args, &video_encoder_info);
@@ -232,10 +237,10 @@ async fn main() -> Result<(), Box<dyn Error>> {
/* Audio */
// Audio Source Element
let audio_source = match args.encoding.audio.capture_method {
encoding_args::AudioCaptureMethod::PulseAudio => {
encoding_args::AudioCaptureMethod::PULSEAUDIO => {
gst::ElementFactory::make("pulsesrc").build()?
}
encoding_args::AudioCaptureMethod::PipeWire => {
encoding_args::AudioCaptureMethod::PIPEWIRE => {
gst::ElementFactory::make("pipewiresrc").build()?
}
encoding_args::AudioCaptureMethod::ALSA => gst::ElementFactory::make("alsasrc").build()?,
@@ -257,8 +262,8 @@ async fn main() -> Result<(), Box<dyn Error>> {
audio_encoder.set_property(
"bitrate",
&match &args.encoding.audio.rate_control {
encoding_args::RateControl::CBR(cbr) => cbr.target_bitrate * 1000i32,
encoding_args::RateControl::VBR(vbr) => vbr.target_bitrate * 1000i32,
encoding_args::RateControl::CBR(cbr) => cbr.target_bitrate.saturating_mul(1000) as i32,
encoding_args::RateControl::VBR(vbr) => vbr.target_bitrate.saturating_mul(1000) as i32,
_ => 128000i32,
},
);
@@ -269,7 +274,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
/* Video */
// Video Source Element
let video_source = gst::ElementFactory::make("waylanddisplaysrc").build()?;
let video_source = Arc::new(gst::ElementFactory::make("waylanddisplaysrc").build()?);
video_source.set_property_from_str("render-node", gpu.render_path());
// Caps Filter Element (resolution, fps)
@@ -323,7 +328,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
/* Output */
// WebRTC sink Element
let signaller = NestriSignaller::new(nestri_ws.clone(), pipeline.clone());
let signaller = NestriSignaller::new(nestri_ws.clone(), video_source.clone());
let webrtcsink = BaseWebRTCSink::with_signaller(Signallable::from(signaller.clone()));
webrtcsink.set_property_from_str("stun-server", "stun://stun.l.google.com:19302");
webrtcsink.set_property_from_str("congestion-control", "disabled");
@@ -456,9 +461,9 @@ async fn main() -> Result<(), Box<dyn Error>> {
let result = run_pipeline(pipeline.clone()).await;
match result {
Ok(_) => log::info!("All tasks finished"),
Ok(_) => tracing::info!("All tasks finished"),
Err(e) => {
log::error!("Error occurred in one of the tasks: {}", e);
tracing::error!("Error occurred in one of the tasks: {}", e);
return Err("Error occurred in one of the tasks".into());
}
}
@@ -471,7 +476,7 @@ async fn run_pipeline(pipeline: Arc<gst::Pipeline>) -> Result<(), Box<dyn Error>
{
if let Err(e) = pipeline.set_state(gst::State::Playing) {
log::error!("Failed to start pipeline: {}", e);
tracing::error!("Failed to start pipeline: {}", e);
return Err("Failed to start pipeline".into());
}
}
@@ -479,12 +484,12 @@ async fn run_pipeline(pipeline: Arc<gst::Pipeline>) -> Result<(), Box<dyn Error>
// Wait for EOS or error (don't lock the pipeline indefinitely)
tokio::select! {
_ = tokio::signal::ctrl_c() => {
log::info!("Pipeline interrupted via Ctrl+C");
tracing::info!("Pipeline interrupted via Ctrl+C");
}
result = listen_for_gst_messages(bus) => {
match result {
Ok(_) => log::info!("Pipeline finished with EOS"),
Err(err) => log::error!("Pipeline error: {}", err),
Ok(_) => tracing::info!("Pipeline finished with EOS"),
Err(err) => tracing::error!("Pipeline error: {}", err),
}
}
}
@@ -504,7 +509,7 @@ async fn listen_for_gst_messages(bus: gst::Bus) -> Result<(), Box<dyn Error>> {
while let Some(msg) = bus_stream.next().await {
match msg.view() {
gst::MessageView::Eos(_) => {
log::info!("Received EOS");
tracing::info!("Received EOS");
break;
}
gst::MessageView::Error(err) => {

1
packages/server/src/messages.rs
View File

@@ -107,7 +107,6 @@ pub fn encode_message<T: Serialize>(message: &T) -> Result<String, Box<dyn Error
}
pub fn decode_message(data: String) -> Result<MessageBase, Box<dyn Error + Send + Sync>> {
println!("Data: {}", data);
let base_message: MessageBase = serde_json::from_str(&data)?;
Ok(base_message)
}

36
packages/server/src/nestrisink/imp.rs
View File

@@ -21,14 +21,14 @@ use webrtc::peer_connection::sdp::session_description::RTCSessionDescription;
pub struct Signaller {
nestri_ws: PLRwLock<Option<Arc<NestriWebSocket>>>,
pipeline: PLRwLock<Option<Arc<gst::Pipeline>>>,
wayland_src: PLRwLock<Option<Arc<gst::Element>>>,
data_channel: AtomicRefCell<Option<gst_webrtc::WebRTCDataChannel>>,
}
impl Default for Signaller {
fn default() -> Self {
Self {
nestri_ws: PLRwLock::new(None),
pipeline: PLRwLock::new(None),
wayland_src: PLRwLock::new(None),
data_channel: AtomicRefCell::new(None),
}
}
@@ -38,12 +38,12 @@ impl Signaller {
*self.nestri_ws.write() = Some(nestri_ws);
}
pub fn set_pipeline(&self, pipeline: Arc<gst::Pipeline>) {
*self.pipeline.write() = Some(pipeline);
pub fn set_wayland_src(&self, wayland_src: Arc<gst::Element>) {
*self.wayland_src.write() = Some(wayland_src);
}
pub fn get_pipeline(&self) -> Option<Arc<gst::Pipeline>> {
self.pipeline.read().clone()
pub fn get_wayland_src(&self) -> Option<Arc<gst::Element>> {
self.wayland_src.read().clone()
}
pub fn set_data_channel(&self, data_channel: gst_webrtc::WebRTCDataChannel) {
@@ -159,8 +159,8 @@ impl Signaller {
);
if let Some(data_channel) = data_channel {
gst::info!(gst::CAT_DEFAULT, "Data channel created");
if let Some(pipeline) = signaller.imp().get_pipeline() {
setup_data_channel(&data_channel, &pipeline);
if let Some(wayland_src) = signaller.imp().get_wayland_src() {
setup_data_channel(&data_channel, &*wayland_src);
signaller.imp().set_data_channel(data_channel);
} else {
gst::error!(gst::CAT_DEFAULT, "Wayland display source not set");
@@ -201,7 +201,7 @@ impl SignallableImpl for Signaller {
// Wait for a reconnection notification
reconnected_notify.notified().await;
println!("Reconnected to relay, re-negotiating...");
tracing::warn!("Reconnected to relay, re-negotiating...");
gst::warning!(gst::CAT_DEFAULT, "Reconnected to relay, re-negotiating...");
// Emit "session-ended" first to make sure the element is cleaned up
@@ -255,7 +255,7 @@ impl SignallableImpl for Signaller {
};
if let Ok(encoded) = encode_message(&join_msg) {
if let Err(e) = nestri_ws.send_message(encoded) {
eprintln!("Failed to send join message: {:?}", e);
tracing::error!("Failed to send join message: {:?}", e);
gst::error!(gst::CAT_DEFAULT, "Failed to send join message: {:?}", e);
}
} else {
@@ -283,7 +283,7 @@ impl SignallableImpl for Signaller {
};
if let Ok(encoded) = encode_message(&sdp_message) {
if let Err(e) = nestri_ws.send_message(encoded) {
eprintln!("Failed to send SDP message: {:?}", e);
tracing::error!("Failed to send SDP message: {:?}", e);
gst::error!(gst::CAT_DEFAULT, "Failed to send SDP message: {:?}", e);
}
} else {
@@ -319,7 +319,7 @@ impl SignallableImpl for Signaller {
};
if let Ok(encoded) = encode_message(&ice_message) {
if let Err(e) = nestri_ws.send_message(encoded) {
eprintln!("Failed to send ICE message: {:?}", e);
tracing::error!("Failed to send ICE message: {:?}", e);
gst::error!(gst::CAT_DEFAULT, "Failed to send ICE message: {:?}", e);
}
} else {
@@ -361,8 +361,8 @@ impl ObjectImpl for Signaller {
}
}
fn setup_data_channel(data_channel: &gst_webrtc::WebRTCDataChannel, pipeline: &gst::Pipeline) {
let pipeline = pipeline.clone();
fn setup_data_channel(data_channel: &gst_webrtc::WebRTCDataChannel, wayland_src: &gst::Element) {
let wayland_src = wayland_src.clone();
data_channel.connect_on_message_data(move |_data_channel, data| {
if let Some(data) = data {
@@ -371,15 +371,15 @@ fn setup_data_channel(data_channel: &gst_webrtc::WebRTCDataChannel, pipeline: &g
if let Some(input_msg) = message_input.data {
// Process the input message and create an event
if let Some(event) = handle_input_message(input_msg) {
// Send the event to pipeline, result bool is ignored
let _ = pipeline.send_event(event);
// Send the event to wayland source, result bool is ignored
let _ = wayland_src.send_event(event);
}
} else {
eprintln!("Failed to parse InputMessage");
tracing::error!("Failed to parse InputMessage");
}
}
Err(e) => {
eprintln!("Failed to decode MessageInput: {:?}", e);
tracing::error!("Failed to decode MessageInput: {:?}", e);
}
}
}

4
packages/server/src/nestrisink/mod.rs
View File

@@ -11,10 +11,10 @@ glib::wrapper! {
}
impl NestriSignaller {
pub fn new(nestri_ws: Arc<NestriWebSocket>, pipeline: Arc<gst::Pipeline>) -> Self {
pub fn new(nestri_ws: Arc<NestriWebSocket>, wayland_src: Arc<gst::Element>) -> Self {
let obj: Self = glib::Object::new();
obj.imp().set_nestri_ws(nestri_ws);
obj.imp().set_pipeline(pipeline);
obj.imp().set_wayland_src(wayland_src);
obj
}
}

59
packages/server/src/websocket.rs
View File

@@ -1,8 +1,7 @@
use crate::messages::{MessageBase, MessageLog, decode_message, encode_message};
use crate::messages::decode_message;
use futures_util::StreamExt;
use futures_util::sink::SinkExt;
use futures_util::stream::{SplitSink, SplitStream};
use log::{Level, Log, Metadata, Record};
use std::collections::HashMap;
use std::error::Error;
use std::sync::{Arc, RwLock};
@@ -63,7 +62,7 @@ impl NestriWebSocket {
return Ok(ws_stream);
}
Err(e) => {
eprintln!("Failed to connect to WebSocket, retrying: {:?}", e);
tracing::error!("Failed to connect to WebSocket, retrying: {:?}", e);
sleep(Duration::from_secs(3)).await; // Wait before retrying
}
}
@@ -87,7 +86,7 @@ impl NestriWebSocket {
let mut ws_read = match ws_read_option {
Some(ws_read) => ws_read,
None => {
eprintln!("Reader is None, cannot proceed");
tracing::error!("Reader is None, cannot proceed");
return;
}
};
@@ -101,7 +100,7 @@ impl NestriWebSocket {
let base_message = match decode_message(data.to_string()) {
Ok(base_message) => base_message,
Err(e) => {
eprintln!("Failed to decode message: {:?}", e);
tracing::error!("Failed to decode message: {:?}", e);
continue;
}
};
@@ -113,7 +112,7 @@ impl NestriWebSocket {
}
}
Err(e) => {
eprintln!(
tracing::error!(
"Error receiving message: {:?}, reconnecting in 3 seconds...",
e
);
@@ -150,10 +149,10 @@ impl NestriWebSocket {
break;
}
Err(e) => {
eprintln!("Error sending message: {:?}", e);
tracing::error!("Error sending message: {:?}", e);
// Attempt to reconnect
if let Err(e) = self_clone.reconnect().await {
eprintln!("Error during reconnection: {:?}", e);
tracing::error!("Error during reconnection: {:?}", e);
// Wait before retrying
sleep(Duration::from_secs(3)).await;
continue;
@@ -161,10 +160,10 @@ impl NestriWebSocket {
}
}
} else {
eprintln!("Writer is None, cannot send message");
tracing::error!("Writer is None, cannot send message");
// Attempt to reconnect
if let Err(e) = self_clone.reconnect().await {
eprintln!("Error during reconnection: {:?}", e);
tracing::error!("Error during reconnection: {:?}", e);
// Wait before retrying
sleep(Duration::from_secs(3)).await;
continue;
@@ -196,7 +195,7 @@ impl NestriWebSocket {
return Ok(());
}
Err(e) => {
eprintln!("Failed to reconnect to WebSocket: {:?}", e);
tracing::error!("Failed to reconnect to WebSocket: {:?}", e);
sleep(Duration::from_secs(3)).await; // Wait before retrying
}
}
@@ -223,40 +222,4 @@ impl NestriWebSocket {
pub fn subscribe_reconnected(&self) -> Arc<Notify> {
self.reconnected_notify.clone()
}
}
impl Log for NestriWebSocket {
fn enabled(&self, metadata: &Metadata) -> bool {
metadata.level() <= Level::Info
}
fn log(&self, record: &Record) {
if self.enabled(record.metadata()) {
let level = record.level().to_string();
let message = record.args().to_string();
let time = chrono::Local::now().to_rfc3339();
// Print to console as well
println!("{}: {}", level, message);
// Encode and send the log message
let log_message = MessageLog {
base: MessageBase {
payload_type: "log".to_string(),
latency: None,
},
level,
message,
time,
};
if let Ok(encoded_message) = encode_message(&log_message) {
if let Err(e) = self.send_message(encoded_message) {
eprintln!("Failed to send log message: {:?}", e);
}
}
}
}
fn flush(&self) {
// No-op for this logger
}
}
}