1use crate::error::Result;
10use crate::event::{ActionError, AfterEvent, BeforeEvent, InputEvent, LogEvent, Point, TraceEvent};
11use serde_json::Value;
12use std::collections::HashMap;
13
14#[derive(Debug, Clone)]
16pub struct Action {
17 pub call_id: String,
18 pub parent_id: Option<String>,
19 pub class: String,
20 pub method: String,
21 pub title: Option<String>,
22 pub page_id: Option<String>,
23 pub start_time: f64,
24 pub end_time: Option<f64>,
27 pub params: Value,
28 pub result: Option<Value>,
29 pub error: Option<ActionError>,
30 pub logs: Vec<LogLine>,
31 pub input: Option<InputEvent>,
32 pub before_snapshot: Option<String>,
36 pub after_snapshot: Option<String>,
39 pub point: Option<Point>,
40}
41
42#[derive(Debug, Clone)]
44pub struct LogLine {
45 pub time: f64,
46 pub message: String,
47}
48
49impl From<LogEvent> for LogLine {
50 fn from(value: LogEvent) -> Self {
51 Self {
52 time: value.time,
53 message: value.message,
54 }
55 }
56}
57
58pub struct ActionStream<I> {
62 events: I,
63 pending: HashMap<String, ActionBuilder>,
64 pending_order: Vec<String>,
67 upstream_done: bool,
68}
69
70impl<I> ActionStream<I>
71where
72 I: Iterator<Item = Result<TraceEvent>>,
73{
74 pub fn new(events: I) -> Self {
75 Self {
76 events,
77 pending: HashMap::new(),
78 pending_order: Vec::new(),
79 upstream_done: false,
80 }
81 }
82}
83
84impl<I> Iterator for ActionStream<I>
85where
86 I: Iterator<Item = Result<TraceEvent>>,
87{
88 type Item = Result<Action>;
89
90 fn next(&mut self) -> Option<Self::Item> {
91 loop {
92 if self.upstream_done {
93 while let Some(call_id) = self.pending_order.pop() {
95 if let Some(builder) = self.pending.remove(&call_id) {
96 return Some(Ok(builder.finalize_truncated()));
97 }
98 }
99 return None;
100 }
101
102 let event = match self.events.next() {
103 Some(Ok(e)) => e,
104 Some(Err(e)) => return Some(Err(e)),
105 None => {
106 self.upstream_done = true;
107 continue;
108 }
109 };
110
111 match event {
112 TraceEvent::Before(b) => {
113 let call_id = b.call_id.clone();
114 if !self.pending.contains_key(&call_id) {
115 self.pending_order.push(call_id.clone());
116 }
117 self.pending.insert(call_id, ActionBuilder::from_before(b));
118 }
119 TraceEvent::Input(i) => {
120 if let Some(builder) = self.pending.get_mut(&i.call_id) {
121 builder.input = Some(i);
122 }
123 }
127 TraceEvent::Log(l) => {
128 if let Some(builder) = self.pending.get_mut(&l.call_id) {
129 builder.logs.push(l.into());
130 }
131 }
132 TraceEvent::After(a) => {
133 if let Some(builder) = self.pending.remove(&a.call_id) {
134 return Some(Ok(builder.finalize(a)));
139 }
140 }
142 _ => {}
143 }
144 }
145 }
146}
147
148struct ActionBuilder {
149 call_id: String,
150 parent_id: Option<String>,
151 class: String,
152 method: String,
153 title: Option<String>,
154 page_id: Option<String>,
155 start_time: f64,
156 params: Value,
157 before_snapshot: Option<String>,
158 logs: Vec<LogLine>,
159 input: Option<InputEvent>,
160}
161
162impl ActionBuilder {
163 fn from_before(b: BeforeEvent) -> Self {
164 Self {
165 call_id: b.call_id,
166 parent_id: b.parent_id,
167 class: b.class,
168 method: b.method,
169 title: b.title,
170 page_id: b.page_id,
171 start_time: b.start_time,
172 params: b.params,
173 before_snapshot: b.before_snapshot,
174 logs: Vec::new(),
175 input: None,
176 }
177 }
178
179 fn finalize(self, a: AfterEvent) -> Action {
180 Action {
181 call_id: self.call_id,
182 parent_id: self.parent_id,
183 class: self.class,
184 method: self.method,
185 title: self.title,
186 page_id: self.page_id,
187 start_time: self.start_time,
188 end_time: Some(a.end_time),
189 params: self.params,
190 result: a.result,
191 error: a.error,
192 logs: self.logs,
193 input: self.input,
194 before_snapshot: self.before_snapshot,
195 after_snapshot: a.after_snapshot,
196 point: a.point,
197 }
198 }
199
200 fn finalize_truncated(self) -> Action {
201 Action {
202 call_id: self.call_id,
203 parent_id: self.parent_id,
204 class: self.class,
205 method: self.method,
206 title: self.title,
207 page_id: self.page_id,
208 start_time: self.start_time,
209 end_time: None,
210 params: self.params,
211 result: None,
212 error: None,
213 logs: self.logs,
214 input: self.input,
215 before_snapshot: self.before_snapshot,
216 after_snapshot: None,
217 point: None,
218 }
219 }
220}