medius 1.6.0

Host control library for the medius transparent mouse passthrough box
Documentation 
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
114
115
116
117
118
119
120

use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::thread::JoinHandle;
use std::time::Duration;

use parking_lot::Mutex;

use crate::protocol::{DecodedFrame, FrameDecoder, FrameType, parse_log};
use crate::types::LogLine;

use super::catch::{self, CatchReg};
use super::correlation::{self, PendingEntry};
use super::counters::Counters;
use super::logs;
use super::reconnect::{self, ReconnectCtx};
use super::slot::TransportSlot;

const READER_IDLE_POLL: Duration = Duration::from_millis(2);

#[allow(clippy::too_many_arguments)]
pub(crate) fn spawn_reader(
    transport: Arc<TransportSlot>,
    pending: Arc<Mutex<HashMap<u8, PendingEntry>>>,
    logs_tx: flume::Sender<LogLine>,
    logs_rx: flume::Receiver<LogLine>,
    events: Arc<Mutex<CatchReg>>,
    counters: Arc<Counters>,
    stop: Arc<AtomicBool>,
    reconnect_ctx: ReconnectCtx,
) -> JoinHandle<()> {
    std::thread::Builder::new()
        .name("medius-reader".into())
        .spawn(move || {
            reader_loop(
                &transport,
                &pending,
                &logs_tx,
                &logs_rx,
                &events,
                &counters,
                &stop,
                &reconnect_ctx,
            )
        })
        .expect("spawn medius-reader thread")
}

#[allow(clippy::too_many_arguments)]
fn reader_loop(
    transport: &TransportSlot,
    pending: &Mutex<HashMap<u8, PendingEntry>>,
    logs_tx: &flume::Sender<LogLine>,
    logs_rx: &flume::Receiver<LogLine>,
    events: &Mutex<CatchReg>,
    counters: &Counters,
    stop: &AtomicBool,
    reconnect_ctx: &ReconnectCtx,
) {
    let mut decoder = FrameDecoder::new();
    let mut buf = [0u8; 1024];
    let mut seen_generation = transport.generation();

    loop {
        if stop.load(Ordering::SeqCst) {
            return;
        }
        let generation = transport.generation();
        if generation != seen_generation {
            decoder = FrameDecoder::new();
            seen_generation = generation;
        }
        let current = transport.current();
        match current.read(&mut buf) {
            Ok(0) => {
                std::thread::sleep(READER_IDLE_POLL);
            }
            Ok(n) => {
                decoder.feed(&buf[..n], |frame| {
                    route_frame(frame, pending, logs_tx, logs_rx, events, counters);
                });
                counters.set_crc_drops(decoder.crc_error_count());
            }
            Err(_) => {
                drop(current);
                reconnect::auto_reconnect(reconnect_ctx, stop);
            }
        }
    }
}

fn route_frame(
    frame: DecodedFrame,
    pending: &Mutex<HashMap<u8, PendingEntry>>,
    logs_tx: &flume::Sender<LogLine>,
    logs_rx: &flume::Receiver<LogLine>,
    events: &Mutex<CatchReg>,
    counters: &Counters,
) {
    counters.inc_rx();
    trace_event!(
        target: "medius::transport",
        tracing::Level::TRACE,
        dir = "rx",
        opcode = u8::from(frame.ty),
        seq = frame.seq,
        len = frame.payload.len(),
    );
    match frame.ty {
        FrameType::Resp => correlation::deliver(pending, frame.seq, frame.payload),
        FrameType::Log => {
            let line = parse_log(&frame.payload);
            #[cfg(feature = "tracing")]
            crate::trace::emit_device_log(&line);
            logs::push(logs_tx, logs_rx, line);
        }
        FrameType::Event => catch::deliver_event(events, &frame.payload),
        _ => {}
    }
}