stream_processor_test/
stream.rs

1// Copyright 2019 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use crate::buffer_set::*;
6use crate::elementary_stream::*;
7use crate::input_packet_stream::*;
8use crate::output_validator::*;
9use crate::{FatalError, Result};
10use fidl_fuchsia_media::*;
11use fidl_fuchsia_sysmem2::BufferCollectionConstraints;
12use fuchsia_stream_processors::*;
13use log::debug;
14use std::rc::Rc;
15
16pub type OrdinalSequence = <OrdinalPattern as IntoIterator>::IntoIter;
17
18#[derive(Debug, Clone)]
19pub struct StreamOptions {
20    /// When true, the stream runner will queue format details for each stream. Otherwise it will
21    /// inherit format details from the codec factory.
22    pub queue_format_details: bool,
23    pub release_input_buffers_at_end: bool,
24    pub release_output_buffers_at_end: bool,
25    pub input_buffer_collection_constraints: Option<BufferCollectionConstraints>,
26    pub output_buffer_collection_constraints: Option<BufferCollectionConstraints>,
27    pub stop_after_first_output: bool,
28    pub stop_after_n_output: Option<usize>,
29    pub close_on_stop: bool,
30}
31
32impl Default for StreamOptions {
33    fn default() -> Self {
34        Self {
35            queue_format_details: true,
36            release_input_buffers_at_end: false,
37            release_output_buffers_at_end: false,
38            input_buffer_collection_constraints: None,
39            output_buffer_collection_constraints: None,
40            stop_after_first_output: false,
41            stop_after_n_output: None,
42            close_on_stop: true,
43        }
44    }
45}
46
47pub struct Stream<'a> {
48    pub format_details_version_ordinal: u64,
49    pub stream_lifetime_ordinal: u64,
50    pub input_buffer_ordinals: &'a mut OrdinalSequence,
51    pub input_packet_stream:
52        Option<InputPacketStream<Box<dyn Iterator<Item = ElementaryStreamChunk> + 'a>>>,
53    pub output_buffer_ordinals: &'a mut OrdinalSequence,
54    pub output_buffer_set: Option<BufferSet>,
55    pub current_output_format: Option<Rc<ValidStreamOutputFormat>>,
56    pub stream_processor: &'a mut StreamProcessorProxy,
57    pub stream: &'a dyn ElementaryStream,
58    pub options: StreamOptions,
59    pub output: Vec<Output>,
60    pub closing: bool,
61}
62
63pub enum StreamControlFlow {
64    Continue,
65    Stop,
66}
67
68impl<'a: 'b, 'b> Stream<'a> {
69    pub async fn start(&'b mut self) -> Result<()> {
70        if self.options.queue_format_details && self.input_packet_stream.is_some() {
71            debug!("Sending input format details for follow-up stream.");
72            self.stream_processor.queue_input_format_details(
73                self.stream_lifetime_ordinal,
74                &self.stream.format_details(self.format_details_version_ordinal),
75            )?;
76        }
77
78        self.send_available_input()?;
79
80        Ok(())
81    }
82
83    pub async fn handle_event(
84        &'b mut self,
85        event: StreamProcessorEvent,
86    ) -> Result<StreamControlFlow> {
87        match event {
88            StreamProcessorEvent::OnInputConstraints { input_constraints } => {
89                debug!("Received input constraints.");
90                debug!("Input constraints are: {:#?}", input_constraints);
91
92                let buffer_set = Box::pin(BufferSetFactory::buffer_set(
93                    get_ordinal(self.input_buffer_ordinals),
94                    ValidStreamBufferConstraints::try_from(input_constraints)?,
95                    self.stream_processor,
96                    BufferSetType::Input,
97                    self.options.input_buffer_collection_constraints.clone(),
98                ))
99                .await?;
100
101                debug!("Sending input format details in response to input constraints.");
102                self.stream_processor.queue_input_format_details(
103                    self.stream_lifetime_ordinal,
104                    &self.stream.format_details(self.format_details_version_ordinal),
105                )?;
106
107                let chunk_stream = self.stream.capped_chunks(buffer_set.buffer_size);
108                self.input_packet_stream = Some(InputPacketStream::new(
109                    buffer_set,
110                    chunk_stream,
111                    self.stream_lifetime_ordinal,
112                ));
113                self.send_available_input()?;
114            }
115            StreamProcessorEvent::OnOutputConstraints { output_config } => {
116                debug!("Received output constraints.");
117                debug!("Output constraints are: {:#?}", output_config);
118
119                let constraints = ValidStreamOutputConstraints::try_from(output_config)?;
120                if constraints.stream_lifetime_ordinal < self.stream_lifetime_ordinal {
121                    debug!(
122                        "Ignoring stale output constraints for stream {}",
123                        constraints.stream_lifetime_ordinal
124                    );
125                    return Ok(StreamControlFlow::Continue);
126                }
127                if constraints.buffer_constraints_action_required {
128                    self.output_buffer_set = Some(
129                        Box::pin(BufferSetFactory::buffer_set(
130                            get_ordinal(self.output_buffer_ordinals),
131                            constraints.buffer_constraints,
132                            self.stream_processor,
133                            BufferSetType::Output,
134                            self.options.output_buffer_collection_constraints.clone(),
135                        ))
136                        .await?,
137                    );
138                }
139            }
140            StreamProcessorEvent::OnFreeInputPacket { free_input_packet } => {
141                debug!("Received freed input packet.");
142                debug!("Freed input packet is: {:#?}", free_input_packet);
143
144                let free_input_packet = ValidPacketHeader::try_from(free_input_packet)?;
145                let input_packet_stream = self.input_packet_stream.as_mut().expect(concat!(
146                    "Unwrapping packet stream; ",
147                    "it should be set before we ",
148                    "get free input packets back."
149                ));
150                input_packet_stream.add_free_packet(free_input_packet)?;
151                if self.closing {
152                    if self.all_input_packets_free() {
153                        debug!("All input packets are free after OnFreeInputPacket. Stopping.");
154                        return Ok(StreamControlFlow::Stop);
155                    }
156                } else {
157                    self.send_available_input()?;
158                }
159            }
160            StreamProcessorEvent::OnOutputFormat { output_format } => {
161                debug!("Received output format.");
162                debug!("Output format is: {:#?}", output_format);
163
164                let output_format = ValidStreamOutputFormat::try_from(output_format)?;
165                if output_format.stream_lifetime_ordinal < self.stream_lifetime_ordinal {
166                    debug!(
167                        "Ignoring stale output format for stream {}",
168                        output_format.stream_lifetime_ordinal
169                    );
170                    return Ok(StreamControlFlow::Continue);
171                }
172                assert_eq!(output_format.stream_lifetime_ordinal, self.stream_lifetime_ordinal);
173                self.current_output_format = Some(Rc::new(output_format));
174            }
175            StreamProcessorEvent::OnOutputPacket {
176                output_packet,
177                error_detected_before,
178                error_detected_during,
179            } => {
180                assert!(!error_detected_before);
181                assert!(!error_detected_during);
182                debug!("Received output packet.");
183                debug!("Output packet is: {:#?}", output_packet);
184
185                let output_packet = ValidPacket::try_from(output_packet)?;
186                if output_packet.stream_lifetime_ordinal < self.stream_lifetime_ordinal {
187                    debug!(
188                        "Ignoring stale output packet for stream {}",
189                        output_packet.stream_lifetime_ordinal
190                    );
191                    self.stream_processor.recycle_output_packet(&PacketHeader {
192                        buffer_lifetime_ordinal: Some(output_packet.header.buffer_lifetime_ordinal),
193                        packet_index: Some(output_packet.header.packet_index),
194                        ..Default::default()
195                    })?;
196                    return Ok(StreamControlFlow::Continue);
197                }
198
199                let reached_limit = if let Some(n) = self.options.stop_after_n_output {
200                    let packet_count =
201                        self.output.iter().filter(|o| matches!(o, Output::Packet(_))).count();
202                    packet_count >= n
203                } else {
204                    false
205                };
206
207                if !reached_limit {
208                    self.output.push(Output::Packet(OutputPacket {
209                        data: self
210                            .output_buffer_set
211                            .as_ref()
212                            .ok_or_else(|| {
213                                FatalError(String::from(concat!(
214                                    "There should be an output buffer set ",
215                                    "if we are receiving output packets"
216                                )))
217                            })?
218                            .read_packet(&output_packet)?,
219                        format: self.current_output_format.clone().ok_or_else(|| {
220                            FatalError(String::from(concat!(
221                                "There should be an output format set ",
222                                "if we are receiving output packets"
223                            )))
224                        })?,
225                        packet: output_packet,
226                    }));
227                }
228
229                self.stream_processor.recycle_output_packet(&PacketHeader {
230                    buffer_lifetime_ordinal: Some(output_packet.header.buffer_lifetime_ordinal),
231                    packet_index: Some(output_packet.header.packet_index),
232                    ..Default::default()
233                })?;
234
235                let should_stop = if let Some(n) = self.options.stop_after_n_output {
236                    let packet_count =
237                        self.output.iter().filter(|o| matches!(o, Output::Packet(_))).count();
238                    packet_count >= n
239                } else {
240                    self.options.stop_after_first_output
241                };
242
243                if should_stop && !self.closing {
244                    return self.close_stream_and_wait_for_free_packets().await;
245                }
246            }
247            StreamProcessorEvent::OnOutputEndOfStream {
248                stream_lifetime_ordinal,
249                error_detected_before,
250            } => {
251                assert!(!error_detected_before);
252                debug!("Received output end of stream.");
253                debug!("End of stream is for stream {}", stream_lifetime_ordinal);
254
255                if stream_lifetime_ordinal < self.stream_lifetime_ordinal {
256                    debug!("Ignoring stale end of stream for stream {}", stream_lifetime_ordinal);
257                    return Ok(StreamControlFlow::Continue);
258                }
259
260                // TODO(turnage): Enable the flush method of ending stream in options.
261                self.output.push(Output::Eos { stream_lifetime_ordinal });
262                if self.options.close_on_stop {
263                    self.stream_processor.close_current_stream(
264                        self.stream_lifetime_ordinal,
265                        self.options.release_input_buffers_at_end,
266                        self.options.release_output_buffers_at_end,
267                    )?;
268                    self.stream_processor.sync().await?;
269                }
270
271                // TODO(turnage): Some codecs return all input packets explicitly, not
272                //                implicitly. All codecs should return explicitly. For now
273                //                we forgive it but soon we want to check that all input
274                //                packets will come back.
275                return Ok(StreamControlFlow::Stop);
276            }
277            e => {
278                debug!("Got other event: {:#?}", e);
279            }
280        }
281
282        Ok(StreamControlFlow::Continue)
283    }
284
285    fn send_available_input(&'b mut self) -> Result<()> {
286        if self.closing {
287            return Ok(());
288        }
289        let input_packet_stream =
290            if let Some(input_packet_stream) = self.input_packet_stream.as_mut() {
291                input_packet_stream
292            } else {
293                return Ok(());
294            };
295
296        loop {
297            match input_packet_stream.next_packet()? {
298                PacketPoll::Ready(input_packet) => {
299                    debug!("Sending input packet. {:?}", input_packet.valid_length_bytes);
300                    self.stream_processor.queue_input_packet(&input_packet)?;
301                }
302                PacketPoll::Eos => {
303                    debug!("Sending end of stream.");
304                    break Ok(self
305                        .stream_processor
306                        .queue_input_end_of_stream(self.stream_lifetime_ordinal)?);
307                }
308                PacketPoll::NotReady => break Ok(()),
309            }
310        }
311    }
312
313    async fn close_stream_and_wait_for_free_packets(&mut self) -> Result<StreamControlFlow> {
314        if !self.options.close_on_stop {
315            return Ok(StreamControlFlow::Stop);
316        }
317        self.closing = true;
318        debug!(
319            "Closing stream {} early, waiting for input packets to be freed...",
320            self.stream_lifetime_ordinal
321        );
322        self.stream_processor.close_current_stream(
323            self.stream_lifetime_ordinal,
324            self.options.release_input_buffers_at_end,
325            self.options.release_output_buffers_at_end,
326        )?;
327        self.stream_processor.sync().await?;
328
329        if self.all_input_packets_free() {
330            debug!("All input packets are already free. Stopping.");
331            Ok(StreamControlFlow::Stop)
332        } else {
333            Ok(StreamControlFlow::Continue)
334        }
335    }
336
337    fn all_input_packets_free(&self) -> bool {
338        self.input_packet_stream.as_ref().map(|s| s.all_packets_free()).unwrap_or(true)
339    }
340}
stream_processor_test/stream.rs

stream_processor_test/
stream.rs
