polish broadcast stuff + refactor and fix examples

Cargo.toml

@@ -2,7 +2,6 @@
 resolver = "2"
 members = [
   "crates/*",
-  "examples/*",
   "tools/*"
 ]
 
@@ -12,7 +11,7 @@ edition = "2021"
 [workspace.dependencies]
 # internal
 haste = { path = "." }
-haste_broadcast = { path = "crates/haste_broadcast" }
+haste_broadcast = { path = "crates/haste_broadcast", default-features = false }
 haste_core = { path = "crates/haste_core" }
 haste_vartype = { path = "crates/haste_vartype" }
 # external
@@ -39,24 +38,51 @@ thiserror = "1.0.64"
 tokio = { version = "1.40.0", default-features = false }
 valveprotos = { git = "https://github.com/blukai/valveprotos-rs.git", rev = "63620c46236ed0f03a9e8b75954a4dafd0fe8e53" }
 
+# enable more optimizations in dev (/debug) builds for dependencies
+[profile.dev.package."*"]
+opt-level = 3
+
 [package]
 name = "haste"
 version = "0.0.0"
 edition.workspace = true
 
 [dependencies]
+haste_broadcast = { workspace = true, optional = true }
 haste_core.workspace = true
 
+[dev-dependencies]
+anyhow.workspace = true
+rand.workspace = true
+
 [features]
+broadcast = ["haste_broadcast/reqwest", "haste_broadcast/tokio"]
 deadlock = ["haste_core/deadlock"]
 dota2 = ["haste_core/dota2"]
 # TODO(blukai): rename preserve-metadata feature into something more meaningful,
 # or get rid of it all together and preserve symbols only in debug builds.
 preserve-metadata = ["haste_core/preserve-metadata"]
 protobuf-src = ["haste_core/protobuf-src"]
-# TODO(blukai): re-export broadcast from haste
-# broadcast = ["haste_core/broadcast"]
 
-# enable more optimizations in dev (/debug) builds for dependencies
-[profile.dev.package."*"]
-opt-level = 3
+[[example]]
+name = "deadlock-gametime"
+
+[[example]]
+name = "deadlock-position"
+required-features = ["deadlock"]
+
+[[example]]
+name = "dota2-allchat"
+required-features = ["dota2"]
+
+[[example]]
+name = "messagehandler-experiment"
+required-features = ["deadlock"]
+
+[[example]]
+name = "lifestate"
+required-features = ["preserve-metadata"]
+
+[[example]]
+name = "seek"
+

crates/haste_broadcast/Cargo.toml

@@ -4,6 +4,7 @@ version = "0.0.0"
 edition.workspace = true
 
 [dependencies]
+anyhow.workspace = true
 bytes.workspace = true
 haste_core.workspace = true
 http.workspace = true
@@ -21,7 +22,6 @@ valveprotos.workspace = true
 default = ["reqwest", "tokio"]
 # "standard" http client
 reqwest = ["dep:reqwest", "tokio"]
-# reqwest is built on top of hyper, hyper need tokio; but that is not all, tokio
-# also provides async-friendly sleep function and block_on for dealing with
-# async rust in sync code (DemoStream)
+# reqwest is built on top of hyper, hyper needs tokio; also tokio also provides
+# async-friendly sleep function
 tokio = ["dep:tokio"] 

crates/haste_broadcast/src/broadcastfile.rs

@@ -0,0 +1,112 @@
+use std::io::{self, Read, Seek, SeekFrom};
+
+use haste_core::demofile::DEMO_RECORD_BUFFER_SIZE;
+use haste_core::demostream::{
+    CmdHeader, DecodeCmdError, DemoStream, ReadCmdError, ReadCmdHeaderError,
+};
+use valveprotos::common::{CDemoClassInfo, CDemoFullPacket, CDemoPacket, CDemoSendTables};
+
+use crate::demostream::{
+    decode_cmd_class_info, decode_cmd_full_packet, decode_cmd_packet, decode_cmd_send_tables,
+    read_cmd_header, scan_for_last_tick,
+};
+
+/// allows to read recorded broadcasts.
+///
+/// the format is:
+/// - `/0/start`
+/// - `/<signup_fragment>/full`
+/// - `/<signup_fragment>/delta`
+/// - `/<signup_fragment + 1>/delta`
+/// - and so on...
+pub struct BroadcastFile<R: Read + Seek> {
+    rdr: R,
+    buf: Vec<u8>,
+    total_ticks: Option<i32>,
+}
+
+impl<R: Read + Seek> BroadcastFile<R> {
+    /// creates a new [`BroadcastFile`] instance from the given reader.
+    ///
+    /// # performance note
+    ///
+    /// for optimal performance make sure to provide a reader that implements buffering (for
+    /// example [`std::io::BufReader`]).
+    pub fn start_reading(rdr: R) -> Self {
+        Self {
+            rdr,
+            buf: vec![0u8; DEMO_RECORD_BUFFER_SIZE],
+            total_ticks: None,
+        }
+    }
+}
+
+impl<R: Read + Seek> DemoStream for BroadcastFile<R> {
+    // stream ops
+    // ----
+
+    /// delegated from [`std::io::Seek`].
+    #[inline]
+    fn seek(&mut self, pos: SeekFrom) -> Result<u64, io::Error> {
+        self.rdr.seek(pos)
+    }
+
+    /// delegated from [`std::io::Seek`].
+    ///
+    /// # note
+    ///
+    /// be aware that this method can be quite expensive. it might be best to make sure not to call
+    /// it too frequently.
+    #[inline]
+    fn stream_position(&mut self) -> Result<u64, io::Error> {
+        self.rdr.stream_position()
+    }
+
+    // cmd header
+    // ----
+
+    fn read_cmd_header(&mut self) -> Result<CmdHeader, ReadCmdHeaderError> {
+        read_cmd_header(&mut self.rdr)
+    }
+
+    // cmd body
+    // ----
+
+    fn read_cmd(&mut self, cmd_header: &CmdHeader) -> Result<&[u8], ReadCmdError> {
+        assert!(!cmd_header.body_compressed);
+
+        let data = &mut self.buf[..cmd_header.body_size as usize];
+        self.rdr.read_exact(data)?;
+        Ok(data)
+    }
+
+    #[inline(always)]
+    fn decode_cmd_send_tables(data: &[u8]) -> Result<CDemoSendTables, DecodeCmdError> {
+        decode_cmd_send_tables(data)
+    }
+
+    #[inline(always)]
+    fn decode_cmd_class_info(data: &[u8]) -> Result<CDemoClassInfo, DecodeCmdError> {
+        decode_cmd_class_info(data)
+    }
+
+    #[inline(always)]
+    fn decode_cmd_packet(data: &[u8]) -> Result<CDemoPacket, DecodeCmdError> {
+        decode_cmd_packet(data)
+    }
+
+    #[inline(always)]
+    fn decode_cmd_full_packet(data: &[u8]) -> Result<CDemoFullPacket, DecodeCmdError> {
+        decode_cmd_full_packet(data)
+    }
+
+    // other
+    // ----
+
+    fn total_ticks(&mut self) -> Result<i32, anyhow::Error> {
+        if self.total_ticks.is_none() {
+            self.total_ticks = Some(scan_for_last_tick(self)?);
+        }
+        Ok(self.total_ticks.unwrap())
+    }
+}