1use ogg::{PacketReader, Packet};
17use header::*;
18use VorbisError;
19use std::io::{Read, Seek};
20use ::audio::{PreviousWindowRight, read_audio_packet,
21 read_audio_packet_generic};
22use ::header::HeaderSet;
23use ::samples::{Samples, InterleavedSamples};
24
25pub fn read_headers<'a, T: Read + Seek + 'a>(rdr: &mut PacketReader<T>) ->
31 Result<(HeaderSet, u32), VorbisError> {
32 let pck :Packet = try!(rdr.read_packet_expected());
33 let ident_hdr = try!(read_header_ident(&pck.data));
34 let stream_serial = pck.stream_serial();
35
36 let mut pck :Packet = try!(rdr.read_packet_expected());
37 while pck.stream_serial() != stream_serial {
38 pck = try!(rdr.read_packet_expected());
39 }
40 let comment_hdr = try!(read_header_comment(&pck.data));
41
42 let mut pck :Packet = try!(rdr.read_packet_expected());
43 while pck.stream_serial() != stream_serial {
44 pck = try!(rdr.read_packet_expected());
45 }
46 let setup_hdr = try!(read_header_setup(&pck.data, ident_hdr.audio_channels,
47 (ident_hdr.blocksize_0, ident_hdr.blocksize_1)));
48
49 rdr.delete_unread_packets();
50 return Ok(((ident_hdr, comment_hdr, setup_hdr), pck.stream_serial()));
51}
52
53pub struct OggStreamReader<T: Read + Seek> {
67 rdr :PacketReader<T>,
68 pwr :PreviousWindowRight,
69
70 stream_serial :u32,
71
72 pub ident_hdr :IdentHeader,
73 pub comment_hdr :CommentHeader,
74 pub setup_hdr :SetupHeader,
75
76 cur_absgp :Option<u64>,
77}
78
79impl<T: Read + Seek> OggStreamReader<T> {
80 pub fn new(rdr :T) ->
86 Result<Self, VorbisError> {
87 OggStreamReader::from_ogg_reader(PacketReader::new(rdr))
88 }
89 pub fn from_ogg_reader(mut rdr :PacketReader<T>) ->
98 Result<Self, VorbisError> {
99 let ((ident_hdr, comment_hdr, setup_hdr), stream_serial) =
100 try!(read_headers(&mut rdr));
101 return Ok(OggStreamReader {
102 rdr,
103 pwr : PreviousWindowRight::new(),
104 ident_hdr,
105 comment_hdr,
106 setup_hdr,
107 stream_serial,
108 cur_absgp : None,
109 });
110 }
111 pub fn into_inner(self) -> PacketReader<T> {
112 self.rdr
113 }
114 fn read_next_audio_packet(&mut self) -> Result<Option<Packet>, VorbisError> {
115 loop {
116 let pck = match try!(self.rdr.read_packet()) {
117 Some(p) => p,
118 None => return Ok(None),
119 };
120 if pck.stream_serial() != self.stream_serial {
121 if pck.first_in_stream() {
122 let ident_hdr = try!(read_header_ident(&pck.data));
125
126 let pck :Packet = try!(self.rdr.read_packet_expected());
127 let comment_hdr = try!(read_header_comment(&pck.data));
128
129 let pck :Packet = try!(self.rdr.read_packet_expected());
130 let setup_hdr = try!(read_header_setup(&pck.data, ident_hdr.audio_channels,
131 (ident_hdr.blocksize_0, ident_hdr.blocksize_1)));
132
133 self.pwr = PreviousWindowRight::new();
135 self.ident_hdr = ident_hdr;
136 self.comment_hdr = comment_hdr;
137 self.setup_hdr = setup_hdr;
138 self.stream_serial = pck.stream_serial();
139 self.cur_absgp = None;
140
141 let pck = match try!(self.rdr.read_packet()) {
144 Some(p) => p,
145 None => return Ok(None),
146 };
147 let _decoded_pck = try!(read_audio_packet(&self.ident_hdr,
148 &self.setup_hdr, &pck.data, &mut self.pwr));
149 self.cur_absgp = Some(pck.absgp_page());
150
151 return Ok(try!(self.rdr.read_packet()));
152 } else {
153 }
155 } else {
156 return Ok(Some(pck));
157 }
158 }
159 }
160 pub fn read_dec_packet(&mut self) ->
168 Result<Option<Vec<Vec<i16>>>, VorbisError> {
169 let pck = try!(self.read_dec_packet_generic());
170 Ok(pck)
171 }
172 pub fn read_dec_packet_generic<S :Samples>(&mut self) ->
180 Result<Option<S>, VorbisError> {
181 let pck = match try!(self.read_next_audio_packet()) {
182 Some(p) => p,
183 None => return Ok(None),
184 };
185 let mut decoded_pck :S = try!(read_audio_packet_generic(&self.ident_hdr,
186 &self.setup_hdr, &pck.data, &mut self.pwr));
187
188 if let (Some(absgp), true) = (self.cur_absgp, pck.last_in_stream()) {
194 let target_length = pck.absgp_page().saturating_sub(absgp) as usize;
195 decoded_pck.truncate(target_length);
196 }
197 if pck.last_in_page() {
198 self.cur_absgp = Some(pck.absgp_page());
199 } else if let &mut Some(ref mut absgp) = &mut self.cur_absgp {
200 *absgp += decoded_pck.num_samples() as u64;
201 }
202
203 return Ok(Some(decoded_pck));
204 }
205 pub fn read_dec_packet_itl(&mut self) ->
216 Result<Option<Vec<i16>>, VorbisError> {
217 let decoded_pck :InterleavedSamples<_> = match try!(self.read_dec_packet_generic()) {
218 Some(p) => p,
219 None => return Ok(None),
220 };
221 return Ok(Some(decoded_pck.samples));
222 }
223
224 pub fn stream_serial(&self) -> u32 {
228 self.stream_serial
229 }
230
231 pub fn get_last_absgp(&self) -> Option<u64> {
236 self.cur_absgp
237 }
238
239 pub fn seek_absgp_pg(&mut self, absgp :u64) -> Result<(), VorbisError> {
247 try!(self.rdr.seek_absgp(None, absgp));
248 self.cur_absgp = None;
250 self.pwr = PreviousWindowRight::new();
251 Ok(())
252 }
253}
254
255#[cfg(feature = "async_ogg")]
256pub mod async_api {
262
263 use super::*;
264 use ogg::OggReadError;
265 use ogg::reading::async_api::PacketReader;
266 use futures::stream::Stream;
267 use tokio_io::AsyncRead;
268 use futures::{Async, Future, Poll};
269 use std::io::{Error, ErrorKind};
270 use std::mem::replace;
271
272 pub struct HeadersReader<T: AsyncRead> {
277 pck_rd :PacketReader<T>,
278 ident_hdr :Option<IdentHeader>,
279 comment_hdr :Option<CommentHeader>,
280 }
281 impl<T: AsyncRead> HeadersReader<T> {
282 pub fn new(inner :T) -> Self {
283 HeadersReader::from_packet_reader(PacketReader::new(inner))
284 }
285 pub fn from_packet_reader(pck_rd :PacketReader<T>) -> Self {
286 HeadersReader {
287 pck_rd,
288 ident_hdr : None,
289 comment_hdr : None,
290 }
291 }
292 }
293 impl<T: AsyncRead> Future for HeadersReader<T> {
294 type Item = HeaderSet;
295 type Error = VorbisError;
296 fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
297 macro_rules! rd_pck {
298 () => {
299 if let Some(pck) = try_ready!(self.pck_rd.poll()) {
300 pck
301 } else {
302 try!(Err(OggReadError::ReadError(Error::new(ErrorKind::UnexpectedEof,
310 "Expected header packet but found end of stream"))))
311 }
312 }
313 }
314 if self.ident_hdr.is_none() {
315 let pck = rd_pck!();
316 self.ident_hdr = Some(try!(read_header_ident(&pck.data)));
317 }
318 if self.comment_hdr.is_none() {
319 let pck = rd_pck!();
320 self.comment_hdr = Some(try!(read_header_comment(&pck.data)));
321 }
322 let setup_hdr = {
323 let ident = self.ident_hdr.as_ref().unwrap();
324 let pck = rd_pck!();
325 try!(read_header_setup(&pck.data,
326 ident.audio_channels, (ident.blocksize_0, ident.blocksize_1)))
327 };
328 let ident_hdr = replace(&mut self.ident_hdr, None).unwrap();
329 let comment_hdr = replace(&mut self.comment_hdr, None).unwrap();
330 Ok(Async::Ready((ident_hdr, comment_hdr, setup_hdr)))
331 }
332 }
333 pub struct OggStreamReader<T :AsyncRead> {
345 pck_rd :PacketReader<T>,
346 pwr :PreviousWindowRight,
347
348 pub ident_hdr :IdentHeader,
349 pub comment_hdr :CommentHeader,
350 pub setup_hdr :SetupHeader,
351
352 absgp_of_last_read :Option<u64>,
353 }
354
355 impl<T :AsyncRead> OggStreamReader<T> {
356 pub fn new(hdr_rdr :HeadersReader<T>, hdrs :HeaderSet) -> Self {
358 OggStreamReader::from_pck_rdr(hdr_rdr.pck_rd, hdrs)
359 }
360 pub fn from_pck_rdr(pck_rd :PacketReader<T>, hdrs :HeaderSet) -> Self {
362 OggStreamReader {
363 pck_rd,
364 pwr : PreviousWindowRight::new(),
365
366 ident_hdr : hdrs.0,
367 comment_hdr : hdrs.1,
368 setup_hdr : hdrs.2,
369
370 absgp_of_last_read : None,
371 }
372 }
373 }
374
375 impl<T :AsyncRead> Stream for OggStreamReader<T> {
376 type Item = Vec<Vec<i16>>;
377 type Error = VorbisError;
378
379 fn poll(&mut self) -> Poll<Option<Vec<Vec<i16>>>, VorbisError> {
380 let pck = match try_ready!(self.pck_rd.poll()) {
381 Some(p) => p,
382 None => return Ok(Async::Ready(None)),
383 };
384 let decoded_pck = try!(read_audio_packet(&self.ident_hdr,
385 &self.setup_hdr, &pck.data, &mut self.pwr));
386 self.absgp_of_last_read = Some(pck.absgp_page());
387 Ok(Async::Ready(Some(decoded_pck)))
388 }
389 }
390}