twitch-hls-client 0.3.2

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

//    Copyright (C) 2023 2bc4
//
//    This program is free software: you can redistribute it and/or modify
//    it under the terms of the GNU General Public License as published by
//    the Free Software Foundation, either version 3 of the License, or
//    (at your option) any later version.
//
//    This program is distributed in the hope that it will be useful,
//    but WITHOUT ANY WARRANTY; without even the implied warranty of
//    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
//    GNU General Public License for more details.
//
//    You should have received a copy of the GNU General Public License
//    along with this program.  If not, see <https://www.gnu.org/licenses/>.

use std::{
    io,
    io::{ErrorKind::BrokenPipe, Read, Write},
    process,
    sync::mpsc::{channel, sync_channel, Receiver, Sender, SyncSender},
    thread::Builder,
};

use anyhow::{Context, Result};
use log::info;

use crate::{http::Request, Player};

pub struct Worker {
    url_tx: Sender<String>,
    sync_rx: Receiver<()>,
}

impl Worker {
    pub fn new(player_path: String, player_args: String) -> Result<Self> {
        let (url_tx, url_rx): (Sender<String>, Receiver<String>) = channel();
        let (sync_tx, sync_rx): (SyncSender<()>, Receiver<()>) = sync_channel(1);

        Builder::new()
            .name(String::from("Segment Worker"))
            .spawn(move || {
                // :(

                if let Err(e) = Self::thread_main(&url_rx, &sync_tx, &player_path, &player_args) {
                    eprintln!("Error: {e}");
                    process::exit(1);
                }
            })
            .context("Failed to spawn segment worker thread")?;

        Ok(Self { url_tx, sync_rx })
    }

    pub fn send(&self, url: &str) -> Result<()> {
        self.url_tx
            .send(url.to_owned())
            .context("Failed to send URL to segment worker thread")
    }

    pub fn sync(&self) -> Result<()> {
        self.sync_rx
            .recv()
            .context("Failed to receive sync state from segment worker thread")
    }

    fn thread_main(
        url_rx: &Receiver<String>,
        sync_tx: &SyncSender<()>,
        player_path: &str,
        player_args: &str,
    ) -> Result<()> {
        let mut player = Player::spawn(player_path, player_args)?;
        let mut pipe = player.stdin()?;

        let mut request = match url_rx.recv() {
            Ok(url) => {
                let mut request = Request::get(&url)?;
                copy_segment(&mut request.reader()?, &mut pipe)?;

                request
            }
            _ => return Ok(()),
        };

        sync_tx
            .send(())
            .context("Failed to send sync state from segment worker thread")?;

        loop {
            match url_rx.recv() {
                Ok(url) => {
                    request.set_url(&url)?;
                    copy_segment(&mut request.reader()?, &mut pipe)?;
                }
                _ => return Ok(()),
            }
        }
    }
}

#[inline]
fn copy_segment(reader: &mut impl Read, writer: &mut impl Write) -> Result<()> {
    match io::copy(reader, writer) {
        Ok(_) => Ok(()),
        Err(e) => match e.kind() {
            BrokenPipe => {
                info!("Player closed, exiting...");
                process::exit(0);
            }
            _ => Err(e.into()),
        },
    }
}