pipedash_plugin_api/logs/
rendered.rs

1use bytes::Bytes;
2use serde::{
3    Deserialize,
4    Serialize,
5};
6
7use super::{
8    content_hash,
9    LogCursor,
10    LogRead,
11    LogSink,
12};
13use crate::error::PluginResult;
14
15#[derive(Debug, Serialize, Deserialize)]
16pub struct RenderedCursor {
17    length: usize,
18    hash: u64,
19}
20
21impl RenderedCursor {
22    pub fn of(rendered: &[u8]) -> Self {
23        Self {
24            length: rendered.len(),
25            hash: content_hash(rendered),
26        }
27    }
28
29    pub fn resume(&self, rendered: &[u8]) -> Option<usize> {
30        let prefix = rendered.get(..self.length)?;
31        (content_hash(prefix) == self.hash).then_some(self.length)
32    }
33
34    pub async fn read(
35        rendered: Bytes, from: Option<&LogCursor>, sink: &mut LogSink,
36    ) -> PluginResult<LogRead> {
37        let resumed = match from {
38            None => None,
39            Some(cursor) => {
40                let resumed = cursor
41                    .parse::<RenderedCursor>()
42                    .and_then(|cursor| cursor.resume(&rendered));
43                if resumed.is_none() {
44                    sink.restart()?;
45                }
46                resumed
47            }
48        };
49        let next = LogCursor::new(&RenderedCursor::of(&rendered))?;
50        match resumed {
51            Some(offset) => {
52                sink.write_chunks(rendered.slice(offset..)).await?;
53                Ok(LogRead::append(next))
54            }
55            None => {
56                sink.write_chunks(rendered).await?;
57                Ok(LogRead::replace(Some(next)))
58            }
59        }
60    }
61}
62
63#[cfg(test)]
64mod tests;
pipedash_plugin_api/logs/rendered.rs

pipedash_plugin_api/logs/
rendered.rs
