From ab672e42b4f0e20ced6c9d7169fd97e3fbc2848a Mon Sep 17 00:00:00 2001 From: Steven Fackler Date: Wed, 21 Dec 2016 21:54:11 -0500 Subject: [PATCH] Shift functions around --- postgres-tokio/src/lib.rs | 76 +++++++++++++++++++-------------------- 1 file changed, 38 insertions(+), 38 deletions(-) diff --git a/postgres-tokio/src/lib.rs b/postgres-tokio/src/lib.rs index 9c3113de..46060f3c 100644 --- a/postgres-tokio/src/lib.rs +++ b/postgres-tokio/src/lib.rs @@ -332,6 +332,44 @@ impl Connection { .boxed() } + fn close_gc(self) -> BoxFuture { + let mut messages = vec![]; + while let Ok((type_, name)) = self.0.close_receiver.try_recv() { + let mut buf = vec![]; + frontend::close(type_, &name, &mut buf).unwrap(); // this can only fail on bad names + messages.push(buf); + } + if messages.is_empty() { + return Ok(self).into_future().boxed(); + } + + let mut buf = vec![]; + frontend::sync(&mut buf); + messages.push(buf); + self.0.send_all(futures::stream::iter(messages.into_iter().map(Ok::<_, io::Error>))) + .map_err(Error::Io) + .and_then(|s| Connection(s.0).finish_close_gc()) + .boxed() + } + + fn finish_close_gc(self) -> BoxFuture { + self.0.read() + .map_err(Error::Io) + .and_then(|(m, s)| { + match m { + backend::Message::ReadyForQuery(_) => { + Either::A(Ok(Connection(s)).into_future()) + } + backend::Message::CloseComplete => Either::B(Connection(s).finish_close_gc()), + backend::Message::ErrorResponse(body) => { + Either::B(Connection(s).ready_err(body)) + } + _ => Either::A(Err(bad_message()).into_future()), + } + }) + .boxed() + } + fn ready_err(self, body: ErrorResponseBody>) -> BoxFuture where T: 'static + Send { @@ -563,44 +601,6 @@ impl Connection { .boxed() } - fn close_gc(self) -> BoxFuture { - let mut messages = vec![]; - while let Ok((type_, name)) = self.0.close_receiver.try_recv() { - let mut buf = vec![]; - frontend::close(type_, &name, &mut buf).unwrap(); // this can only fail on bad names - messages.push(buf); - } - if messages.is_empty() { - return Ok(self).into_future().boxed(); - } - - let mut buf = vec![]; - frontend::sync(&mut buf); - messages.push(buf); - self.0.send_all(futures::stream::iter(messages.into_iter().map(Ok::<_, io::Error>))) - .map_err(Error::Io) - .and_then(|s| Connection(s.0).finish_close_gc()) - .boxed() - } - - fn finish_close_gc(self) -> BoxFuture { - self.0.read() - .map_err(Error::Io) - .and_then(|(m, s)| { - match m { - backend::Message::ReadyForQuery(_) => { - Either::A(Ok(Connection(s)).into_future()) - } - backend::Message::CloseComplete => Either::B(Connection(s).finish_close_gc()), - backend::Message::ErrorResponse(body) => { - Either::B(Connection(s).ready_err(body)) - } - _ => Either::A(Err(bad_message()).into_future()), - } - }) - .boxed() - } - pub fn close(self) -> BoxFuture<(), Error> { let mut terminate = vec![]; frontend::terminate(&mut terminate);