candid_server/
client_handler.rs

1use std::io::{BufReader, Read, Write};
2use std::net::TcpStream;
3use std::sync::mpsc;
4use std::thread;
5
6use byteorder::{NetworkEndian, ReadBytesExt, WriteBytesExt};
7use socketcan::{CanFrame, CanSocket, EmbeddedFrame, ExtendedId, Socket};
8
9/// Spawns two threads: one to read incoming frames from the client (and write them to the
10/// CanSocket), and one to read incoming frames from the `CANReader` (and write them to the client)
11pub fn new_client_handler(
12    mut connection: TcpStream,
13    can_interface: String,
14    can_reader: mpsc::Receiver<CanFrame>,
15) -> (thread::JoinHandle<()>, thread::JoinHandle<()>) {
16    let connection_clone = connection.try_clone().unwrap();
17    // Thread to pass frames from the server to the client
18    let tx_handle = thread::spawn(move || {
19        let ip = connection.peer_addr().unwrap();
20
21        loop {
22            // Get an incoming frame from the reader
23            let frame = can_reader.recv().unwrap();
24
25            // Relay to the client
26            if send_frame(&mut connection, frame).is_err() {
27                println!("Connection to client {:?} dropped.", ip);
28                break;
29            }
30        }
31    });
32
33    // Thread to pass frames from the client to the CAN socket
34    let rx_handle = thread::spawn(move || {
35        let ip = connection_clone.peer_addr().unwrap();
36        let mut connection = BufReader::new(connection_clone);
37
38        // Possible issue: There is a new socket for each thread
39        let can_socket = CanSocket::open(&can_interface).unwrap();
40
41        loop {
42            // Get an incoming frame from the client
43            let Some(id) = connection
44                .read_u32::<NetworkEndian>()
45                .ok()
46                .and_then(ExtendedId::new)
47            else {
48                println!("Connection to client {:?} dropped.", ip);
49                break;
50            };
51
52            let mut data = [0 as u8; 8];
53            if connection.read_exact(&mut data).is_err() {
54                println!("Connection to client {:?} dropped.", ip);
55                break;
56            }
57
58            // Write the frame to the CAN Socket
59            let frame = CanFrame::new(id, &data).unwrap();
60            can_socket.write_frame_insist(&frame).unwrap();
61        }
62    });
63
64    (tx_handle, rx_handle)
65}
66
67fn send_frame(connection: &mut TcpStream, frame: CanFrame) -> Result<(), std::io::Error> {
68    let id = match frame.id() {
69        socketcan::Id::Standard(id) => id.as_raw() as u32,
70        socketcan::Id::Extended(id) => id.as_raw(),
71    };
72    connection.write_u32::<NetworkEndian>(id)?;
73    connection.write_all(frame.data())?;
74    connection.flush().unwrap();
75    Ok(())
76}
candid_server/client_handler.rs

candid_server/
client_handler.rs
