| ... | @@ -935,13 +935,13 @@ impl Outbox { | ... | @@ -935,13 +935,13 @@ impl Outbox { |
| 935 | } | 935 | } |
| 936 | } | 936 | } |
| 937 | | 937 | |
| 938 | /// Everything queued, waiting up to `wait` for something; none once closed. | 938 | /// Everything queued, waiting up to `wait` for something; none once closed and empty. |
| 939 | fn take(&self, wait: Duration) -> Option<Vec<u8>> { | 939 | fn take(&self, wait: Duration) -> Option<Vec<u8>> { |
| 940 | let mut queue = self.queue.lock().unwrap(); | 940 | let mut queue = self.queue.lock().unwrap(); |
| 941 | if queue.frames.is_empty() && !queue.closed { | 941 | if queue.frames.is_empty() && !queue.closed { |
| 942 | queue = self.ready.wait_timeout(queue, wait).unwrap().0; | 942 | queue = self.ready.wait_timeout(queue, wait).unwrap().0; |
| 943 | } | 943 | } |
| 944 | if queue.closed { | 944 | if queue.closed && queue.frames.is_empty() { |
| 945 | return None; | 945 | return None; |
| 946 | } | 946 | } |
| 947 | queue.bytes = 0; | 947 | queue.bytes = 0; |
| ... | @@ -956,7 +956,7 @@ impl Outbox { | ... | @@ -956,7 +956,7 @@ impl Outbox { |
| 956 | } | 956 | } |
| 957 | if queue.bytes + frame.len() > self.most { | 957 | if queue.bytes + frame.len() > self.most { |
| 958 | drop(queue); | 958 | drop(queue); |
| 959 | self.close(); | 959 | self.abort(); |
| 960 | return; | 960 | return; |
| 961 | } | 961 | } |
| 962 | queue.bytes += frame.len(); | 962 | queue.bytes += frame.len(); |
| ... | @@ -964,14 +964,19 @@ impl Outbox { | ... | @@ -964,14 +964,19 @@ impl Outbox { |
| 964 | self.ready.notify_one(); | 964 | self.ready.notify_one(); |
| 965 | } | 965 | } |
| 966 | | 966 | |
| | 967 | /// Hangs up after writing what is queued, which tells a refused guest its code was wrong. |
| 967 | fn close(&self) { | 968 | fn close(&self) { |
| 968 | let mut queue = self.queue.lock().unwrap(); | 969 | self.queue.lock().unwrap().closed = true; |
| 969 | *queue = Queue { | 970 | self.ready.notify_one(); |
| | 971 | } |
| | 972 | |
| | 973 | /// Hangs up now, dropping what is queued. |
| | 974 | fn abort(&self) { |
| | 975 | *self.queue.lock().unwrap() = Queue { |
| 970 | closed: true, | 976 | closed: true, |
| 971 | ..Queue::default() | 977 | ..Queue::default() |
| 972 | }; | 978 | }; |
| 973 | self.ready.notify_one(); | 979 | self.ready.notify_one(); |
| 974 | drop(queue); | | |
| 975 | if let Some(stream) = &self.stream { | 980 | if let Some(stream) = &self.stream { |
| 976 | let _ = stream.shutdown(Shutdown::Both); | 981 | let _ = stream.shutdown(Shutdown::Both); |
| 977 | } | 982 | } |
| ... | @@ -983,13 +988,14 @@ impl Outbox { | ... | @@ -983,13 +988,14 @@ impl Outbox { |
| 983 | let frame = { | 988 | let frame = { |
| 984 | let mut queue = self.queue.lock().unwrap(); | 989 | let mut queue = self.queue.lock().unwrap(); |
| 985 | loop { | 990 | loop { |
| 986 | if queue.closed { | | |
| 987 | return; | | |
| 988 | } | | |
| 989 | if let Some(frame) = queue.frames.pop_front() { | 991 | if let Some(frame) = queue.frames.pop_front() { |
| 990 | queue.bytes -= frame.len(); | 992 | queue.bytes -= frame.len(); |
| 991 | break frame; | 993 | break frame; |
| 992 | } | 994 | } |
| | 995 | if queue.closed { |
| | 996 | drop(queue); |
| | 997 | return self.abort(); |
| | 998 | } |
| 993 | queue = self.ready.wait(queue).unwrap(); | 999 | queue = self.ready.wait(queue).unwrap(); |
| 994 | } | 1000 | } |
| 995 | }; | 1001 | }; |
| ... | @@ -997,8 +1003,7 @@ impl Outbox { | ... | @@ -997,8 +1003,7 @@ impl Outbox { |
| 997 | return; | 1003 | return; |
| 998 | }; | 1004 | }; |
| 999 | if stream.write_all(&frame).is_err() { | 1005 | if stream.write_all(&frame).is_err() { |
| 1000 | self.close(); | 1006 | return self.abort(); |
| 1001 | return; | | |
| 1002 | } | 1007 | } |
| 1003 | } | 1008 | } |
| 1004 | } | 1009 | } |