eqtui 0.1.3

Terminal-native(TUI) audio effects processor for PipeWire
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

// Copyright (C) 2026 SiputBiru <radityamahatma23@gmail.com>
// SPDX-License-Identifier: GPL-2.0-only

//! Per-client connection handling: bounded request framing and responses.

use std::io::{BufRead, BufReader, Read, Write};
use std::os::unix::net::UnixStream;
use std::sync::Arc;

use pipewire::channel;
use tracing::{debug, error, warn};

use crate::protocol::{Request, Response};

use super::dispatch::dispatch_request;
use super::state::DaemonState;

/// Serves one client connection: reads bounded JSON-lines requests, dispatches
/// them, and writes responses until the client disconnects.
pub(super) fn handle_client(
    stream: UnixStream,
    state: Arc<DaemonState>,
    cmd_tx: channel::Sender<crate::state::PwCommand>,
    client_id: u64,
) {
    let read_stream = match stream.try_clone() {
        Ok(s) => s,
        Err(e) => {
            error!(%e, client_id, "Failed to clone client stream for reading");
            return;
        }
    };

    state.register_client(&stream, client_id);

    // 64 KiB covers ~31 bands with generous headroom; anything larger is hostile.
    const MAX_REQUEST_LINE: u64 = 64 * 1024;

    let mut reader = BufReader::new(read_stream);
    loop {
        let mut buf = Vec::new();
        // `take` caps bytes read from the stream, so a missing newline can't
        // grow `buf` past the limit.
        let n = match reader
            .by_ref()
            .take(MAX_REQUEST_LINE + 1)
            .read_until(b'\n', &mut buf)
        {
            Ok(n) => n,
            Err(e) => {
                debug!(%e, client_id, "Read error; closing connection");
                break;
            }
        };
        if n == 0 {
            break; // client closed
        }
        if buf.len() as u64 > MAX_REQUEST_LINE || !buf.ends_with(b"\n") {
            let _ = send_resp(
                &stream,
                Response {
                    ok: false,
                    error: Some(format!("request line exceeds {MAX_REQUEST_LINE} bytes")),
                    status: None,
                },
            );
            warn!(client_id, "Oversized request line; disconnecting client");
            break;
        }

        let trimmed = String::from_utf8_lossy(&buf);
        let trimmed = trimmed.trim();
        if trimmed.is_empty() {
            continue;
        }

        let req: Request = match serde_json::from_str(trimmed) {
            Ok(r) => r,
            Err(e) => {
                let _ = send_resp(
                    &stream,
                    Response {
                        ok: false,
                        error: Some(format!("Invalid JSON: {e}")),
                        status: None,
                    },
                );
                continue;
            }
        };

        let resp = dispatch_request(req, &state, &cmd_tx);
        let _ = send_resp(&stream, resp);
    }

    state.unregister_client(client_id);
}

fn send_resp(mut stream: &UnixStream, resp: Response) -> std::io::Result<()> {
    let json = serde_json::to_string(&resp)
        .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
    stream.write_all(json.as_bytes())?;
    stream.write_all(b"\n")?;
    Ok(())
}