actix/actix-extras
1
0
Fork 0
mirror of https://github.com/actix/actix-extras.git synced 2024-11-24 16:02:59 +01:00
Code Issues Releases Wiki Activity

simplify channels list management

Browse Source
This commit is contained in:
Nikolay Kim 2018-03-03 11:16:55 -08:00
parent f456be0309
commit 058630d041
2 changed files with 23 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
src/server/channel.rs
View File

@ -18,15 +18,6 @@ enum HttpProtocol<T: IoStream, H: 'static> {
Unknown(Rc<WorkerSettings<H>>, Option<SocketAddr>, T, BytesMut), Unknown(Rc<WorkerSettings<H>>, Option<SocketAddr>, T, BytesMut),
} }
impl<T: IoStream, H: 'static> HttpProtocol<T, H> {
fn is_unknown(&self) -> bool {
match *self {
HttpProtocol::Unknown(_, _, _, _) => true,
_ => false
}
}
}
enum ProtocolKind { enum ProtocolKind {
Http1, Http1,
Http2, Http2,
@ -44,15 +35,14 @@ impl<T, H> HttpChannel<T, H> where T: IoStream, H: HttpHandler + 'static
io: T, peer: Option<SocketAddr>, http2: bool) -> HttpChannel<T, H> io: T, peer: Option<SocketAddr>, http2: bool) -> HttpChannel<T, H>
{ {
settings.add_channel(); settings.add_channel();
if http2 { if http2 {
HttpChannel { HttpChannel {
node: None, node: None, proto: Some(HttpProtocol::H2(
proto: Some(HttpProtocol::H2(
h2::Http2::new(settings, io, peer, Bytes::new()))) } h2::Http2::new(settings, io, peer, Bytes::new()))) }
} else { } else {
HttpChannel { HttpChannel {
node: None, node: None, proto: Some(HttpProtocol::Unknown(
proto: Some(HttpProtocol::Unknown(
settings, peer, io, BytesMut::with_capacity(4096))) } settings, peer, io, BytesMut::with_capacity(4096))) }
} }
} }
@ -78,15 +68,18 @@ impl<T, H> Future for HttpChannel<T, H> where T: IoStream, H: HttpHandler + 'sta
type Error = (); type Error = ();
fn poll(&mut self) -> Poll<Self::Item, Self::Error> { fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
if !self.proto.as_ref().map(|p| p.is_unknown()).unwrap_or(false) && self.node.is_none() { if !self.node.is_none() {
self.node = Some(Node::new(self)); let el = self as *mut _;
match self.proto { self.node = Some(Node::new(el));
let _ = match self.proto {
Some(HttpProtocol::H1(ref mut h1)) => Some(HttpProtocol::H1(ref mut h1)) =>
h1.settings().head().insert(self.node.as_ref().unwrap()), self.node.as_ref().map(|n| h1.settings().head().insert(n)),
Some(HttpProtocol::H2(ref mut h2)) => Some(HttpProtocol::H2(ref mut h2)) =>
h2.settings().head().insert(self.node.as_ref().unwrap()), self.node.as_ref().map(|n| h2.settings().head().insert(n)),
_ => (), Some(HttpProtocol::Unknown(ref mut settings, _, _, _)) =>
} self.node.as_ref().map(|n| settings.head().insert(n)),
None => unreachable!(),
};
} }
let kind = match self.proto { let kind = match self.proto {
@ -95,7 +88,7 @@ impl<T, H> Future for HttpChannel<T, H> where T: IoStream, H: HttpHandler + 'sta
match result { match result {
Ok(Async::Ready(())) | Err(_) => { Ok(Async::Ready(())) | Err(_) => {
h1.settings().remove_channel(); h1.settings().remove_channel();
self.node.as_mut().unwrap().remove(); self.node.as_mut().map(|n| n.remove());
}, },
_ => (), _ => (),
} }
@ -106,7 +99,7 @@ impl<T, H> Future for HttpChannel<T, H> where T: IoStream, H: HttpHandler + 'sta
match result { match result {
Ok(Async::Ready(())) | Err(_) => { Ok(Async::Ready(())) | Err(_) => {
h2.settings().remove_channel(); h2.settings().remove_channel();
self.node.as_mut().unwrap().remove(); self.node.as_mut().map(|n| n.remove());
}, },
_ => (), _ => (),
} }
@ -117,6 +110,7 @@ impl<T, H> Future for HttpChannel<T, H> where T: IoStream, H: HttpHandler + 'sta
Ok(Async::Ready(0)) | Err(_) => { Ok(Async::Ready(0)) | Err(_) => {
debug!("Ignored premature client disconnection"); debug!("Ignored premature client disconnection");
settings.remove_channel(); settings.remove_channel();
self.node.as_mut().map(|n| n.remove());
return Err(()) return Err(())
}, },
_ => (), _ => (),
@ -163,11 +157,11 @@ pub(crate) struct Node<T>
impl<T> Node<T> impl<T> Node<T>
{ {
fn new(el: &mut T) -> Self { fn new(el: *mut T) -> Self {
Node { Node {
next: None, next: None,
prev: None, prev: None,
element: el as *mut _, element: el,
} }
} }
@ -187,14 +181,13 @@ impl<T> Node<T>
} }
fn remove(&mut self) { fn remove(&mut self) {
#[allow(mutable_transmutes)]
unsafe { unsafe {
let mut prev = self.prev.take(); self.element = ptr::null_mut();
let next = self.next.take(); let next = self.next.take();
let mut prev = self.prev.take();
if let Some(ref mut prev) = prev { if let Some(ref mut prev) = prev {
let p: &mut Node<()> = mem::transmute(prev.as_ref().unwrap()); prev.as_mut().unwrap().next = next;
p.next = next;
} }
} }
} }

4
src/server/settings.rs
View File

@ -63,7 +63,7 @@ pub(crate) struct WorkerSettings<H> {
bytes: Rc<SharedBytesPool>, bytes: Rc<SharedBytesPool>,
messages: Rc<helpers::SharedMessagePool>, messages: Rc<helpers::SharedMessagePool>,
channels: Cell<usize>, channels: Cell<usize>,
node: Node<()>, node: Box<Node<()>>,
} }
impl<H> WorkerSettings<H> { impl<H> WorkerSettings<H> {
@ -75,7 +75,7 @@ impl<H> WorkerSettings<H> {
bytes: Rc::new(SharedBytesPool::new()), bytes: Rc::new(SharedBytesPool::new()),
messages: Rc::new(helpers::SharedMessagePool::new()), messages: Rc::new(helpers::SharedMessagePool::new()),
channels: Cell::new(0), channels: Cell::new(0),
node: Node::head(), node: Box::new(Node::head()),
} }
} }