Merge pull request #206 from lrbalt/master

Do not panic on connection reset

audio/src/decrypt.rs

@@ -8,7 +8,7 @@ use std::ops::Add;
 use core::audio_key::AudioKey;
 
 const AUDIO_AESIV: &'static [u8] = &[
-    0x72, 0xe0, 0x67, 0xfb, 0xdd, 0xcb, 0xcf, 0x77, 0xeb, 0xe8, 0xbc, 0x64, 0x3f, 0x63, 0x0d, 0x93
+    0x72, 0xe0, 0x67, 0xfb, 0xdd, 0xcb, 0xcf, 0x77, 0xeb, 0xe8, 0xbc, 0x64, 0x3f, 0x63, 0x0d, 0x93,
 ];
 
 pub struct AudioDecrypt<T: io::Read> {

audio/src/fetch.rs

@@ -1,7 +1,7 @@
 use bit_set::BitSet;
 use byteorder::{BigEndian, ByteOrder, WriteBytesExt};
-use futures::Stream;
 use futures::sync::{mpsc, oneshot};
+use futures::Stream;
 use futures::{Async, Future, Poll};
 use std::cmp::min;
 use std::fs;

audio/src/lewton_decoder.rs

@@ -25,10 +25,10 @@ where
     }
 
     pub fn next_packet(&mut self) -> Result<Option<VorbisPacket>, VorbisError> {
+        use self::lewton::audio::AudioReadError::AudioIsHeader;
         use self::lewton::OggReadError::NoCapturePatternFound;
         use self::lewton::VorbisError::BadAudio;
         use self::lewton::VorbisError::OggError;
-        use self::lewton::audio::AudioReadError::AudioIsHeader;
         loop {
             match self.0.read_dec_packet_itl() {
                 Ok(Some(packet)) => return Ok(Some(VorbisPacket(packet))),

connect/src/spirc.rs

@@ -301,6 +301,10 @@ impl Future for SpircTask {
         loop {
             let mut progress = false;
 
+            if self.session.is_invalid() {
+                return Ok(Async::Ready(()));
+            }
+
             if !self.shutdown {
                 match self.subscription.poll().unwrap() {
                     Async::Ready(Some(frame)) => {

core/src/session.rs

@@ -20,6 +20,7 @@ use mercury::MercuryManager;
 struct SessionData {
     country: String,
     canonical_username: String,
+    invalid: bool,
 }
 
 struct SessionInternal {
@@ -77,7 +78,9 @@ impl Session {
                 reusable_credentials.username.clone(),
             );
 
-            handle.spawn(task.map_err(|e| panic!(e)));
+            handle.spawn(task.map_err(|e| {
+                error!("{:?}", e);
+            }));
 
             session
         });
@@ -104,6 +107,7 @@ impl Session {
             data: RwLock::new(SessionData {
                 country: String::new(),
                 canonical_username: username,
+                invalid: false,
             }),
 
             tx_connection: sender_tx,
@@ -212,6 +216,15 @@ impl Session {
     pub fn session_id(&self) -> usize {
         self.0.session_id
     }
+
+    pub fn shutdown(&self) {
+        debug!("Invalidating session[{}]", self.0.session_id);
+        self.0.data.write().unwrap().invalid = true;
+    }
+
+    pub fn is_invalid(&self) -> bool {
+        self.0.data.read().unwrap().invalid
+    }
 }
 
 #[derive(Clone)]
@@ -240,6 +253,7 @@ where
 impl<S> Future for DispatchTask<S>
 where
     S: Stream<Item = (u8, Bytes)>,
+    <S as Stream>::Error: ::std::fmt::Debug,
 {
     type Item = ();
     type Error = S::Error;
@@ -251,7 +265,15 @@ where
         };
 
         loop {
-            let (cmd, data) = try_ready!(self.0.poll()).expect("connection closed");
+            let (cmd, data) = match self.0.poll() {
+                Ok(Async::Ready(t)) => t,
+                Ok(Async::NotReady) => return Ok(Async::NotReady),
+                Err(e) => {
+                    session.shutdown();
+                    return Err(From::from(e));
+                }
+            }.expect("connection closed");
+
             session.dispatch(cmd, data);
         }
     }

playback/src/mixer/softmixer.rs

@@ -1,5 +1,5 @@
-use std::sync::Arc;
 use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Arc;
 
 use super::AudioFilter;
 use super::Mixer;

playback/src/player.rs

@@ -341,6 +341,10 @@ impl PlayerInternal {
                     self.handle_packet(packet, current_normalisation_factor);
                 }
             }
+
+            if self.session.is_invalid() {
+                return;
+            }
         }
     }