Allow concurrent TCP clients

.github/workflows/rust.yml

@@ -10,13 +10,13 @@ jobs:
     name: Default features
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/checkout@v2
+    - uses: actions/checkout@v4
     - name: Format
       run: cargo fmt --all -- --check
     - name: Clippy
-      run: cargo clippy -- -D warnings
+      run: cargo clippy --all-targets -- -D warnings
     - name: Doc
-      run: cargo doc
+      run: RUSTDOCFLAGS="-D warnings" cargo doc
     - name: Build
       run: cargo build --verbose
     - name: Run tests
@@ -26,13 +26,13 @@ jobs:
     name: No features
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/checkout@v2
+    - uses: actions/checkout@v4
     - name: Format
       run: cargo fmt --all -- --check
     - name: Clippy
-      run: cargo clippy --no-default-features -- -D warnings
+      run: cargo clippy --all-targets --no-default-features -- -D warnings
     - name: Doc
-      run: cargo doc
+      run: RUSTDOCFLAGS="-D warnings" cargo doc
     - name: Build
       run: cargo build --no-default-features --verbose
     - name: Run tests

CHANGELOG.md

@@ -7,6 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.6.2] - 2024-05-01
+
+### Added
+
+- Support for handling concurrent TCP connections with `--input TCP`.
+
+### Changed
+
+- Log BBFRAMEs and GSE packets in hex format.
+
 ## [0.6.1] - 2024-04-08
 
 ### Fixed
@@ -113,7 +123,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Initial release.
 
-[unreleased]: https://github.com/daniestevez/dvb-gse/compare/v0.6.1...HEAD
+[unreleased]: https://github.com/daniestevez/dvb-gse/compare/v0.6.2...HEAD
+[0.6.2]: https://github.com/daniestevez/dvb-gse/compare/v0.6.1...v0.6.2
 [0.6.1]: https://github.com/daniestevez/dvb-gse/compare/v0.6.0...v0.6.1
 [0.6.0]: https://github.com/daniestevez/dvb-gse/compare/v0.5.0...v0.6.0
 [0.5.0]: https://github.com/daniestevez/dvb-gse/compare/v0.4.4...v0.5.0

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "dvb-gse"
-version = "0.6.1"
+version = "0.6.2"
 edition = "2021"
 authors = ["Daniel Estevez <[email protected]>"]
 description = "DVB-GSE (Digital Video Brodcast Generic Stream Encapsulation)"
@@ -21,9 +21,10 @@ cli = ["anyhow", "clap", "env_logger", "libc", "tun-tap"]
 anyhow = { version = "1", features = ["std"], optional = true }
 bitvec = "1"
 bytes = "1.2"
-clap = { version = "4.0", features = ["derive"], optional = true }
-crc = "3.0"
-env_logger = { version = "0.10", optional = true }
+clap = { version = "4", features = ["derive"], optional = true }
+crc = "3"
+env_logger = { version = "0.11", optional = true }
+faster-hex = "0.9"
 lazy_static = "1.4"
 libc = { version = "0.2", optional = true }
 log = "0.4"

src/bbframe.rs

@@ -363,7 +363,7 @@ impl<R: RecvFragment> BBFrameReceiver for BBFrameDefrag<R> {
                 self.recv()?;
             }
             let bbframe = Bytes::copy_from_slice(&self.buffer[..bbframe_len]);
-            log::trace!("completed BBFRAME {:?}", bbframe);
+            log::trace!("completed BBFRAME {}", faster_hex::hex_string(&bbframe));
             return Ok(bbframe);
         }
     }
@@ -476,7 +476,7 @@ impl<R: RecvBBFrame> BBFrameReceiver for BBFrameRecv<R> {
             ));
         }
         let bbframe = Bytes::copy_from_slice(&self.buffer[..bbframe_len]);
-        log::trace!("completed BBFRAME {:?}", bbframe);
+        log::trace!("completed BBFRAME {}", faster_hex::hex_string(&bbframe));
         Ok(bbframe)
     }
 }
@@ -548,7 +548,7 @@ impl<R: RecvStream> BBFrameReceiver for BBFrameStream<R> {
         self.recv_stream
             .recv_stream(&mut self.buffer[BBHeader::LEN..bbframe_len])?;
         let bbframe = Bytes::copy_from_slice(&self.buffer[..bbframe_len]);
-        log::trace!("completed BBFRAME {:?}", bbframe);
+        log::trace!("completed BBFRAME {}", faster_hex::hex_string(&bbframe));
         Ok(bbframe)
     }
 }
@@ -676,7 +676,7 @@ mod test {
          30 31 32 33 34 35 36 37"
     );
 
-    static MULTIPLE_FRAGMENTS: [&'static [u8]; 3] = [&FRAGMENT0, &FRAGMENT1, &FRAGMENT2];
+    static MULTIPLE_FRAGMENTS: [&[u8]; 3] = [&FRAGMENT0, &FRAGMENT1, &FRAGMENT2];
 
     #[test]
     fn single_fragment_defrag() {
@@ -861,7 +861,7 @@ mod proptests {
             // twice in order to flush a partial BBHEADER left
             // by the garbage.
             assert!(times_called.get() <= garbage_data.len() + 2);
-            if times_called.get() >= garbage_data.len() + 1 {
+            if times_called.get() > garbage_data.len() {
                 assert_eq!(bbframe, Bytes::from_static(&SINGLE_FRAGMENT));
             }
         }
@@ -894,7 +894,7 @@ mod proptests {
             while times_called.get() <= garbage_data.len() {
                 if let Ok(bbframe) = recv.get_bbframe() {
                     assert!(times_called.get() <= garbage_data.len() + 1);
-                    if times_called.get() >= garbage_data.len() + 1 {
+                    if times_called.get() > garbage_data.len() {
                         assert_eq!(bbframe, Bytes::from_static(&SINGLE_FRAGMENT));
                     }
                 }

src/cli.rs

@@ -8,13 +8,15 @@
 
 use crate::{
     bbframe::{BBFrameDefrag, BBFrameReceiver, BBFrameRecv, BBFrameStream},
-    gsepacket::GSEPacketDefrag,
+    gsepacket::{GSEPacketDefrag, PDU},
 };
 use anyhow::{Context, Result};
 use clap::Parser;
 use std::{
     net::{SocketAddr, TcpListener, UdpSocket},
     os::unix::io::AsRawFd,
+    sync::mpsc,
+    thread,
 };
 
 /// Receive DVB-GSE and send PDUs into a TUN device
@@ -116,16 +118,22 @@ fn set_reuseaddr(socket: &UdpSocket) -> Result<()> {
 
 #[derive(Debug)]
 struct AppLoop<D> {
-    bbframe_recv: Option<D>,
+    bbframe_recv: D,
     gsepacket_defrag: GSEPacketDefrag,
     tun: tun_tap::Iface,
     bbframe_recv_errors_fatal: bool,
 }
 
+fn write_pdu_tun(pdu: &PDU, tun: &mut tun_tap::Iface) {
+    if let Err(err) = tun.send(pdu.data()) {
+        log::error!("could not write packet to TUN device: {err}");
+    }
+}
+
 impl<D: BBFrameReceiver> AppLoop<D> {
     fn app_loop(&mut self) -> Result<()> {
         loop {
-            let bbframe = match self.bbframe_recv.as_mut().unwrap().get_bbframe() {
+            let bbframe = match self.bbframe_recv.get_bbframe() {
                 Ok(b) => b,
                 Err(err) => {
                     if self.bbframe_recv_errors_fatal {
@@ -136,26 +144,28 @@ impl<D: BBFrameReceiver> AppLoop<D> {
                 }
             };
             // the BBFRAME was validated by bbframe_recv, so we can unwrap here
-            let pdus = self.gsepacket_defrag.defragment(&bbframe).unwrap();
-            for pdu in pdus {
-                if let Err(err) = self.tun.send(pdu.data()) {
-                    log::error!("could not write packet to TUN device: {err}");
-                }
+            for pdu in self.gsepacket_defrag.defragment(&bbframe).unwrap() {
+                write_pdu_tun(&pdu, &mut self.tun);
             }
         }
     }
 }
 
+fn gsepacket_defragmenter(args: &Args) -> GSEPacketDefrag {
+    let mut defrag = GSEPacketDefrag::new();
+    defrag.set_skip_total_length_check(args.skip_total_length);
+    defrag
+}
+
 /// Main function of the CLI application.
 pub fn main() -> Result<()> {
     env_logger::init();
     let args = Args::parse();
-    let tun = tun_tap::Iface::without_packet_info(&args.tun, tun_tap::Mode::Tun)
+    let mut tun = tun_tap::Iface::without_packet_info(&args.tun, tun_tap::Mode::Tun)
         .context("failed to open TUN device")?;
-    let mut gsepacket_defrag = GSEPacketDefrag::new();
-    gsepacket_defrag.set_skip_total_length_check(args.skip_total_length);
     match args.input {
         InputFormat::UdpFragments | InputFormat::UdpComplete => {
+            let gsepacket_defrag = gsepacket_defragmenter(&args);
             let socket = UdpSocket::bind(args.listen).context("failed to bind to UDP socket")?;
             setup_multicast(&socket, &args.listen)?;
             match args.input {
@@ -164,7 +174,7 @@ pub fn main() -> Result<()> {
                     bbframe_recv.set_isi(args.isi);
                     bbframe_recv.set_header_bytes(args.header_length)?;
                     let mut app = AppLoop {
-                        bbframe_recv: Some(bbframe_recv),
+                        bbframe_recv,
                         gsepacket_defrag,
                         tun,
                         bbframe_recv_errors_fatal: true,
@@ -176,7 +186,7 @@ pub fn main() -> Result<()> {
                     bbframe_recv.set_isi(args.isi);
                     bbframe_recv.set_header_bytes(args.header_length)?;
                     let mut app = AppLoop {
-                        bbframe_recv: Some(bbframe_recv),
+                        bbframe_recv,
                         gsepacket_defrag,
                         tun,
                         bbframe_recv_errors_fatal: false,
@@ -189,34 +199,61 @@ pub fn main() -> Result<()> {
         InputFormat::Tcp => {
             let listener =
                 TcpListener::bind(args.listen).context("failed to bind to TCP socket")?;
-            let mut app = AppLoop {
-                bbframe_recv: None,
-                gsepacket_defrag,
-                tun,
-                bbframe_recv_errors_fatal: true,
-            };
-            for stream in listener.incoming() {
-                let stream = match stream {
-                    Ok(s) => s,
-                    Err(e) => {
-                        log::error!("connection error {e}");
-                        continue;
-                    }
-                };
-                match stream.peer_addr() {
-                    Ok(addr) => log::info!("TCP client connected from {addr}"),
-                    Err(err) => log::error!(
-                        "TCP client connected (but could not retrieve peer address): {err}"
-                    ),
+            // For TCP, the application runs each TCP connection in a dedicated
+            // thread. There is another thread that owns the TUN. The TCP
+            // connection threads are connected to the TUN thread by an mpsc
+            // channel.
+            let channel_capacity = 64;
+            let (tun_tx, tun_rx) = mpsc::sync_channel(channel_capacity);
+            thread::spawn(move || {
+                for pdu in tun_rx.iter() {
+                    write_pdu_tun(&pdu, &mut tun);
                 }
-                let mut bbframe_recv = BBFrameStream::new(stream);
-                bbframe_recv.set_isi(args.isi);
-                bbframe_recv.set_header_bytes(args.header_length)?;
-                app.bbframe_recv = Some(bbframe_recv);
-                if let Err(err) = app.app_loop() {
-                    log::error!("error; waiting for another client: {err:#}");
+            });
+            // use thread scope to pass args by reference
+            thread::scope(|s| {
+                for stream in listener.incoming() {
+                    let stream = match stream {
+                        Ok(s) => s,
+                        Err(e) => {
+                            log::error!("connection error {e}");
+                            continue;
+                        }
+                    };
+                    match stream.peer_addr() {
+                        Ok(addr) => log::info!("TCP client connected from {addr}"),
+                        Err(err) => log::error!(
+                            "TCP client connected (but could not retrieve peer address): {err}"
+                        ),
+                    }
+                    s.spawn({
+                        let args = &args;
+                        let tun_tx = tun_tx.clone();
+                        move || {
+                            let mut gsepacket_defrag = gsepacket_defragmenter(args);
+                            let mut bbframe_recv = BBFrameStream::new(stream);
+                            bbframe_recv.set_isi(args.isi);
+                            if let Err(err) = bbframe_recv.set_header_bytes(args.header_length) {
+                                eprintln!("could not set header length: {err}");
+                                std::process::exit(1);
+                            }
+                            loop {
+                                let bbframe = match bbframe_recv.get_bbframe() {
+                                    Ok(b) => b,
+                                    Err(err) => {
+                                        log::error!("failed to receive BBFRAME; terminating connection: {err}");
+                                        return;
+                                    }
+                                };
+                                // the BBFRAME was validated by bbframe_recv, so we can unwrap here
+                                for pdu in gsepacket_defrag.defragment(&bbframe).unwrap() {
+                                    tun_tx.send(pdu).unwrap();
+                                }
+                            }
+                        }
+                    });
                 }
-            }
+            });
         }
     }
     Ok(())

src/gseheader.rs

@@ -539,7 +539,8 @@ mod proptests {
                 header.protocol_type();
                 if let Some(label) = header.label() {
                     label.as_slice();
-                    assert_eq!(label.is_empty(), label.len() == 0);
+                    let len = label.len();
+                    assert_eq!(label.is_empty(), len == 0);
                 }
                 assert!(header.len() >= 3);
                 assert!(!header.is_empty());

src/gsepacket.rs

@@ -77,7 +77,10 @@ impl GSEPacket {
         Ok(std::iter::from_fn(move || {
             if let Some(packet) = GSEPacket::from_bytes(&remain, label.as_ref()) {
                 log::debug!("extracted GSE Packet with header {}", packet.header());
-                log::trace!("GSE Packet data field {:?}", packet.data());
+                log::trace!(
+                    "GSE Packet data field {}",
+                    faster_hex::hex_string(packet.data())
+                );
                 remain = remain.slice(packet.len()..);
                 if let Some(l) = packet.header.label() {
                     label = Some(l.clone());
@@ -376,7 +379,7 @@ mod proptests {
         fn defrag_garbage(garbage_bbframes in garbage()) {
             let mut defrag = GSEPacketDefrag::new();
             for bbframe in &garbage_bbframes {
-                if let Ok(pdus) = defrag.defragment(&bbframe) {
+                if let Ok(pdus) = defrag.defragment(bbframe) {
                     for pdu in pdus {
                         pdu.data();
                         pdu.protocol_type();