re_sdk 0.31.1

Rerun logging SDK
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104

use re_chunk::Chunk;
use re_lenses::{Lens, Lenses, OutputMode};
use re_log_types::{LogMsg, StoreId};

use crate::sink::LogSink;

/// A sink which can transform a [`LogMsg`] and forward the result to an underlying backing [`LogSink`].
///
/// The sink will only forward components that are matched by a lens specified via [`Self::with_lens`].
pub struct LensesSink<S: LogSink> {
    sink: S,
    lenses: Lenses,
    strict: bool,
}

impl<S: LogSink> LensesSink<S> {
    /// Creates a new sink without any lenses attached.
    ///
    /// Use [`Self::with_lens`] to add an additional lens to this sink.
    ///
    /// By default, the sink will do its best effort to produce chunks despite
    /// of errors in Lenses that it might encounter.
    pub fn new(sink: S) -> Self {
        Self {
            sink,
            lenses: Lenses::new(OutputMode::DropUnmatched),
            strict: false,
        }
    }

    /// Adds a [`Lens`] to this sink.
    pub fn with_lens(mut self, lens: Lens) -> Self {
        self.lenses.add_lens(lens);
        self
    }

    /// Configure how to handle matched and unmatched data.
    ///
    /// See [`OutputMode`] for more details.
    pub fn output_mode(mut self, mode: OutputMode) -> Self {
        self.lenses.set_output_mode(mode);
        self
    }

    /// When `strict` is `true` Lenses that encounter an error will not emit partial chunks.
    pub fn strict(mut self, strict: bool) -> Self {
        self.strict = strict;
        self
    }

    fn send_or_log_error(&self, store_id: StoreId, chunk: &Chunk) {
        match chunk.to_arrow_msg() {
            Ok(arrow_msg) => {
                self.sink.send(LogMsg::ArrowMsg(store_id, arrow_msg));
            }
            Err(err) => {
                re_log::error_once!("Failed to create log message from chunk: {err}");
            }
        }
    }
}

impl<S: LogSink> LogSink for LensesSink<S> {
    fn send(&self, msg: re_log_types::LogMsg) {
        match &msg {
            LogMsg::SetStoreInfo(_) | LogMsg::BlueprintActivationCommand(_) => {
                self.sink.send(msg);
            }
            LogMsg::ArrowMsg(store_id, arrow_msg) => match Chunk::from_arrow_msg(arrow_msg) {
                Ok(original_chunk) => {
                    let new_chunks = self.lenses.apply(&original_chunk);
                    for maybe_chunk in new_chunks {
                        match maybe_chunk {
                            Ok(new_chunk) => self.send_or_log_error(store_id.clone(), &new_chunk),
                            Err(partial_chunk) => {
                                for error in partial_chunk.errors() {
                                    // TODO(grtlr): Make this even more contextualized in the future!
                                    re_log::error_once!("Error encountered for lens: {error}");
                                }
                                if let Some(chunk) = partial_chunk.take()
                                    && self.strict
                                {
                                    self.send_or_log_error(store_id.clone(), &chunk);
                                }
                            }
                        }
                    }
                }

                Err(err) => {
                    re_log::error_once!("Failed to convert arrow message to chunk: {err}");
                    self.sink.send(msg);
                }
            },
        }
    }

    fn flush_blocking(
        &self,
        timeout: std::time::Duration,
    ) -> Result<(), crate::sink::SinkFlushError> {
        self.sink.flush_blocking(timeout)
    }
}