refactor: Make the parser selectable via --parser

CHANGELOG.md

@@ -14,6 +14,8 @@ All notable changes to this project will be documented in this file.
   Enable it with `--enable-sink web`, it listens on `[::]:8080` unless told otherwise.
   The new `web` feature is enabled by default.
 - Support arbitrary strings as `--advertised-endpoint` ([#105]).
+- Add the CLI argument `--parser` to select the parser implementation: `original` (default) or `refactored` ([#109]).
+  `refactored` doesn't support the `binary-sync-pixels` feature, so breakwater refuses to start with that combination.
 
 ### Changed
 
@@ -43,6 +45,7 @@ All notable changes to this project will be documented in this file.
 [#104]: https://github.com/sbernauer/breakwater/pull/104
 [#105]: https://github.com/sbernauer/breakwater/pull/105
 [#107]: https://github.com/sbernauer/breakwater/pull/107
+[#109]: https://github.com/sbernauer/breakwater/pull/109
 
 ## [0.22.0] - 2026-06-20
 

README.md

@@ -102,6 +102,8 @@ Options:
           Listen address the Prometheus exporter should listen on [default: [::]:9100]
       --shared-memory-name <SHARED_MEMORY_NAME>
           Create (or use an existing) shared memory region for the framebuffer. This enables other applications to read and write Pixel values to the framebuffer or can be used to persist the canvas across restarts
+      --parser <PARSER>
+          The used parser implementation [default: original] [possible values: original, refactored]
   -h, --help
           Print help (see more with '--help')
   -V, --version
@@ -110,10 +112,10 @@ Options:
 Network listener options:
   -l, --listener-address <LISTEN_ADDRESSES>
           Listen address to bind to (multiple can be specified). The default value will listen on all interfaces for IPv4 and IPv6 packets [default: [::]:1234]
-      --advertised-endpoint <ADVERTISED_ENDPOINTS>
+      --advertised-endpoint <HOST:PORT>
           Specify one or more pixelflut endpoints to display to spectators
       --network-buffer-size <NETWORK_BUFFER_SIZE>
-          The size in bytes of the network buffer used for each open TCP connection. Please use at least 64 KB (64_000 bytes) [default: 262144]
+          The size in bytes of the network buffer used for each open TCP connection. Use at least 64 KB (64_000 bytes) [default: 262144]
   -c, --connections-per-ip <CONNECTIONS_PER_IP>
           Allow only a certain number of connections per ip address
 
@@ -152,6 +154,10 @@ VNC sink options:
 web sink options:
       --web-listen-address <WEB_LISTEN_ADDRESSES>
           Web server listen address to bind to (multiple can be specified). The default value will listen on all interfaces for IPv4 and IPv6 packets [default: [::]:8080]
+      --web-frame-compression-level <FRAME_COMPRESSION_LEVEL>
+          The compression level to use to compress frames before sending them to the connected clients [default: 1]
+      --web-frame-compression-chunks <FRAME_COMPRESSION_CHUNKS>
+          The number of independently-compressed chunks a frame is split into before compression [default: 16]
       --web-chat-messages-per-minute <CHAT_MESSAGES_PER_MINUTE>
           Maximum number of chat messages a single IP address may send per minute in the WebUI [default: 10]
 ```

breakwater-parser/Cargo.toml

@@ -12,6 +12,7 @@ name = "parsing"
 harness = false
 
 [dependencies]
+clap = { workspace = true, optional = true }
 color-eyre.workspace = true
 const_format.workspace = true
 memchr.workspace = true
@@ -25,6 +26,7 @@ rstest.workspace = true
 
 [features]
 alpha = []
+clap = ["dep:clap"]
 binary-set-pixel = []
 binary-sync-pixels = []
 # For ever pixel on the framebuffer the write timestamp is tracked

breakwater-parser/src/lib.rs

@@ -12,6 +12,7 @@ mod assembler;
 mod framebuffer;
 mod memchr;
 mod original;
+mod parser_implementation;
 mod refactored;
 
 #[cfg(target_arch = "x86_64")]
@@ -26,6 +27,7 @@ pub use framebuffer::{
 };
 pub use memchr::MemchrParser;
 pub use original::{OriginalParser, OriginalParserFrameBuffer};
+pub use parser_implementation::ParserImplementation;
 pub use refactored::RefactoredParser;
 
 pub const HELP_TEXT: &[u8] = formatcp!("\

breakwater-parser/src/parser_implementation.rs

@@ -0,0 +1,41 @@
+use std::fmt::{self, Display};
+
+use color_eyre::eyre::{self, bail};
+
+use crate::refactored;
+
+/// Different parser implementations, with their strengths and weaknesses.
+/// Not all parsers implement the same feature set.
+#[derive(Copy, Clone, Debug, Default, PartialEq, Eq)]
+#[cfg_attr(feature = "clap", derive(clap::ValueEnum))]
+pub enum ParserImplementation {
+    /// The OG parser, most feature rich. Ugly but performant.
+    #[default]
+    Original,
+
+    /// The original parser restructured into smaller functions. Doesn't support binary-sync-pixels.
+    Refactored,
+}
+
+impl ParserImplementation {
+    /// Checks that the parser supports everything this build enables, errors otherwise.
+    pub fn check_supported(self) -> eyre::Result<()> {
+        let unsupported_feature = match self {
+            Self::Original => None,
+            Self::Refactored => refactored::UNSUPPORTED_ENABLED_FEATURE,
+        };
+        if let Some(feature) = unsupported_feature {
+            bail!("the {self} parser doesn't support the {feature} feature this build enables");
+        }
+        Ok(())
+    }
+}
+
+impl Display for ParserImplementation {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str(match self {
+            Self::Original => "original",
+            Self::Refactored => "refactored",
+        })
+    }
+}

breakwater-parser/src/refactored.rs

@@ -11,6 +11,14 @@ use crate::{
 
 const PARSER_LOOKAHEAD: usize = "PX 1234 1234 rrggbbaa\n".len(); // Longest possible command
 
+/// A feature this build enables, but RefactoredParser doesn't implement
+pub(crate) const UNSUPPORTED_ENABLED_FEATURE: Option<&str> = if cfg!(feature = "binary-sync-pixels")
+{
+    Some("binary-sync-pixels")
+} else {
+    None
+};
+
 pub struct RefactoredParser<FB: FrameBuffer> {
     connection_x_offset: usize,
     connection_y_offset: usize,

breakwater/Cargo.toml

@@ -12,7 +12,7 @@ name = "breakwater"
 path = "src/main.rs"
 
 [dependencies]
-breakwater-parser.workspace = true
+breakwater-parser = { workspace = true, features = ["clap"] }
 breakwater-egui-overlay = { workspace = true, optional = true }
 
 async-trait.workspace = true

breakwater/src/cli_args.rs

@@ -3,6 +3,8 @@ use std::{
     net::{IpAddr, SocketAddr},
 };
 
+use breakwater_parser::ParserImplementation;
+
 use crate::sinks::cli_args::SinkCliArgs;
 
 pub const DEFAULT_NETWORK_BUFFER_SIZE: usize = 256 * 1024;
@@ -32,6 +34,10 @@ pub struct CliArgs {
     #[clap(long)]
     pub shared_memory_name: Option<String>,
 
+    /// The used parser implementation
+    #[clap(long, value_enum, default_value_t)]
+    pub parser: ParserImplementation,
+
     #[clap(flatten)]
     pub network_listener: NetworkListenerCliArgs,
 

breakwater/src/main.rs

@@ -59,6 +59,7 @@ async fn main() -> eyre::Result<()> {
         statistics_tx.clone(),
         args.network_listener.network_buffer_size,
         args.network_listener.connections_per_ip,
+        args.parser,
     )
     .await
     .context("failed to start pixelflut server")?;

breakwater/src/server.rs

@@ -7,13 +7,16 @@ use std::{
     time::Duration,
 };
 
-use breakwater_parser::{FrameBuffer, OriginalParser, OriginalParserFrameBuffer, Parser};
+use breakwater_parser::{
+    FrameBuffer, OriginalParser, OriginalParserFrameBuffer, Parser, ParserImplementation,
+    RefactoredParser,
+};
 use color_eyre::eyre::{self, Context};
 use futures::{StreamExt, stream::SelectAll};
 use memadvise::Advice;
 use tokio::{
     io::{AsyncReadExt, AsyncWriteExt},
-    net::{TcpListener, ToSocketAddrs},
+    net::{TcpListener, TcpStream, ToSocketAddrs},
     sync::mpsc,
     time::Instant,
 };
@@ -37,6 +40,7 @@ pub struct Server<FB: FrameBuffer> {
     network_buffer_size: usize,
     connections_per_ip: HashMap<IpAddr, u64>,
     max_connections_per_ip: Option<u64>,
+    parser: ParserImplementation,
 }
 
 impl<FB: OriginalParserFrameBuffer + Send + Sync + 'static> Server<FB> {
@@ -47,7 +51,10 @@ impl<FB: OriginalParserFrameBuffer + Send + Sync + 'static> Server<FB> {
         statistics_tx: mpsc::Sender<StatisticsEvent>,
         network_buffer_size: usize,
         max_connections_per_ip: Option<u64>,
+        parser: ParserImplementation,
     ) -> eyre::Result<Self> {
+        parser.check_supported()?;
+
         let mut listener_streams = Vec::with_capacity(listen_addresses.len());
         for addr in listen_addresses {
             let listener = TcpListener::bind(addr)
@@ -68,6 +75,7 @@ impl<FB: OriginalParserFrameBuffer + Send + Sync + 'static> Server<FB> {
             network_buffer_size,
             connections_per_ip: HashMap::new(),
             max_connections_per_ip,
+            parser,
         })
     }
 
@@ -122,38 +130,69 @@ impl<FB: OriginalParserFrameBuffer + Send + Sync + 'static> Server<FB> {
                 }
             }
 
-            let fb_for_thread = Arc::clone(&self.fb);
-            let statistics_tx_for_thread = self.statistics_tx.clone();
+            let fb = Arc::clone(&self.fb);
+            let statistics_tx = self.statistics_tx.clone();
             let network_buffer_size = self.network_buffer_size;
-            let connection_dropped_tx_clone = connection_dropped_tx.clone();
-            tokio::spawn(async move {
-                if let Err(error) = handle_connection(
+            let connection_dropped_tx = connection_dropped_tx.clone();
+            // Every parser gets its own task type, so its connection loop is compiled on its own,
+            // exactly as if the parser was hardcoded. No dynamic dispatch and no code of the other
+            // parser in the same function.
+            match self.parser {
+                ParserImplementation::Original => spawn_connection(
                     stream,
                     ip,
-                    fb_for_thread,
-                    statistics_tx_for_thread,
+                    OriginalParser::new(fb),
+                    statistics_tx,
                     network_buffer_size,
-                    connection_dropped_tx_clone,
-                )
-                .await
-                {
-                    tracing::error!(?error, %ip, "failed to handle connection");
-                }
-            });
+                    connection_dropped_tx,
+                ),
+                ParserImplementation::Refactored => spawn_connection(
+                    stream,
+                    ip,
+                    RefactoredParser::new(fb),
+                    statistics_tx,
+                    network_buffer_size,
+                    connection_dropped_tx,
+                ),
+            }
         }
 
         Ok(())
     }
 }
 
+fn spawn_connection(
+    stream: TcpStream,
+    ip: IpAddr,
+    parser: impl Parser + Send + 'static,
+    statistics_tx: mpsc::Sender<StatisticsEvent>,
+    network_buffer_size: usize,
+    connection_dropped_tx: Option<mpsc::UnboundedSender<IpAddr>>,
+) {
+    tokio::spawn(async move {
+        if let Err(error) = handle_connection(
+            stream,
+            ip,
+            parser,
+            statistics_tx,
+            network_buffer_size,
+            connection_dropped_tx,
+        )
+        .await
+        {
+            tracing::error!(?error, %ip, "failed to handle connection");
+        }
+    });
+}
+
 #[instrument(
-    skip(stream, fb, statistics_tx, connection_dropped_tx),
+    skip(stream, parser, statistics_tx, connection_dropped_tx),
     err(level = "debug")
 )]
-pub async fn handle_connection<FB: OriginalParserFrameBuffer>(
+pub async fn handle_connection(
     mut stream: impl AsyncReadExt + AsyncWriteExt + Send + Unpin,
     ip: IpAddr,
-    fb: Arc<FB>,
+    mut parser: impl Parser,
     statistics_tx: mpsc::Sender<StatisticsEvent>,
     network_buffer_size: usize,
     connection_dropped_tx: Option<mpsc::UnboundedSender<IpAddr>>,
@@ -173,9 +212,6 @@ pub async fn handle_connection<FB: OriginalParserFrameBuffer>(
     // Number bytes left over **on the first bytes of the buffer** from the previous loop iteration
     let mut leftover_bytes_in_buffer = 0;
 
-    // Not using `ParserImplementation` to avoid the dynamic dispatch.
-    // let mut parser = ParserImplementation::Simple(SimpleParser::new(fb));
-    let mut parser = OriginalParser::new(fb);
     let parser_lookahead = parser.parser_lookahead();
 
     // If we send e.g. an StatisticsEvent::BytesRead for every time we read something from the socket the statistics thread would go crazy.