improve error messages

13 files changed, 207 insertions, 124 deletions

diff --git a/src/client.rs b/src/client.rs
index ce7ce77..80a7455 100644
--- a/src/client.rs
+++ b/src/client.rs
@@ -308,11 +308,11 @@ impl<S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + 'static>
 
         let auth_type = self.auth.auth_type();
         let id = id.to_string();
-        let addr = OAUTH_LISTEN_ADDRESS
+        let address = OAUTH_LISTEN_ADDRESS
             .parse()
             .context(crate::error::ParseAddr)?;
-        let listener = tokio::net::TcpListener::bind(&addr)
-            .context(crate::error::Bind)?;
+        let listener = tokio::net::TcpListener::bind(&address)
+            .context(crate::error::Bind { address })?;
         Ok(Box::new(
             listener
                 .incoming()
diff --git a/src/cmd/play.rs b/src/cmd/play.rs
index c99a697..22bd220 100644
--- a/src/cmd/play.rs
+++ b/src/cmd/play.rs
@@ -56,6 +56,7 @@ enum FileState {
         filename: String,
     },
     Opening {
+        filename: String,
         fut: tokio::fs::file::OpenFuture<String>,
     },
     Open {
@@ -100,13 +101,17 @@ impl PlaySession {
         match &mut self.file {
             FileState::Closed { filename } => {
                 self.file = FileState::Opening {
+                    filename: filename.to_string(),
                     fut: tokio::fs::File::open(filename.to_string()),
                 };
                 Ok(crate::component_future::Async::DidWork)
             }
-            FileState::Opening { fut } => {
-                let file =
-                    try_ready!(fut.poll().context(crate::error::OpenFile));
+            FileState::Opening { filename, fut } => {
+                let file = try_ready!(fut.poll().with_context(|| {
+                    crate::error::OpenFile {
+                        filename: filename.to_string(),
+                    }
+                }));
                 let file = crate::ttyrec::File::new(file);
                 self.file = FileState::Open { file };
                 Ok(crate::component_future::Async::DidWork)
diff --git a/src/cmd/record.rs b/src/cmd/record.rs
index 7481f92..4a1ab6e 100644
--- a/src/cmd/record.rs
+++ b/src/cmd/record.rs
@@ -97,6 +97,7 @@ enum FileState {
         filename: String,
     },
     Opening {
+        filename: String,
         fut: tokio::fs::file::CreateFuture<String>,
     },
     Open {
@@ -166,13 +167,17 @@ impl RecordSession {
         match &mut self.file {
             FileState::Closed { filename } => {
                 self.file = FileState::Opening {
+                    filename: filename.to_string(),
                     fut: tokio::fs::File::create(filename.to_string()),
                 };
                 Ok(crate::component_future::Async::DidWork)
             }
-            FileState::Opening { fut } => {
-                let file =
-                    try_ready!(fut.poll().context(crate::error::OpenFile));
+            FileState::Opening { filename, fut } => {
+                let file = try_ready!(fut.poll().with_context(|| {
+                    crate::error::OpenFile {
+                        filename: filename.clone(),
+                    }
+                }));
                 let mut file = crate::ttyrec::File::new(file);
                 file.write_frame(self.buffer.contents())?;
                 self.file = FileState::Open { file };
diff --git a/src/cmd/server.rs b/src/cmd/server.rs
index 134cb9d..f6fabfa 100644
--- a/src/cmd/server.rs
+++ b/src/cmd/server.rs
@@ -165,7 +165,7 @@ fn create_server(
 )> {
     let (mut sock_w, sock_r) = tokio::sync::mpsc::channel(100);
     let listener = tokio::net::TcpListener::bind(&address)
-        .context(crate::error::Bind)?;
+        .context(crate::error::Bind { address })?;
     let acceptor = listener
         .incoming()
         .context(crate::error::Acceptor)
@@ -197,13 +197,16 @@ fn create_server_tls(
 )> {
     let (mut sock_w, sock_r) = tokio::sync::mpsc::channel(100);
     let listener = tokio::net::TcpListener::bind(&address)
-        .context(crate::error::Bind)?;
+        .context(crate::error::Bind { address })?;
 
-    let mut file = std::fs::File::open(tls_identity_file)
-        .context(crate::error::OpenIdentityFile)?;
+    let mut file = std::fs::File::open(tls_identity_file).context(
+        crate::error::OpenFileSync {
+            filename: tls_identity_file,
+        },
+    )?;
     let mut identity = vec![];
     file.read_to_end(&mut identity)
-        .context(crate::error::ReadIdentityFile)?;
+        .context(crate::error::ReadFileSync)?;
     let identity = native_tls::Identity::from_pkcs12(&identity, "")
         .context(crate::error::ParseIdentity)?;
     let acceptor = native_tls::TlsAcceptor::new(identity)
diff --git a/src/cmd/stream.rs b/src/cmd/stream.rs
index 95c1191..11634d3 100644
--- a/src/cmd/stream.rs
+++ b/src/cmd/stream.rs
@@ -95,13 +95,15 @@ impl crate::config::Config for Config {
                 let connector = connector.clone();
                 let connector = tokio_tls::TlsConnector::from(connector);
                 let stream = tokio::net::tcp::TcpStream::connect(&address);
-                Box::new(stream.context(crate::error::Connect).and_then(
-                    move |stream| {
-                        connector
-                            .connect(&host, stream)
-                            .context(crate::error::ConnectTls)
-                    },
-                ))
+                Box::new(
+                    stream
+                        .context(crate::error::Connect { address })
+                        .and_then(move |stream| {
+                            connector
+                                .connect(&host, stream)
+                                .context(crate::error::ConnectTls { host })
+                        }),
+                )
             });
             Box::new(StreamSession::new(
                 &self.command,
@@ -114,7 +116,7 @@ impl crate::config::Config for Config {
             let connect: crate::client::Connector<_> = Box::new(move || {
                 Box::new(
                     tokio::net::tcp::TcpStream::connect(&address)
-                        .context(crate::error::Connect),
+                        .context(crate::error::Connect { address }),
                 )
             });
             Box::new(StreamSession::new(
diff --git a/src/cmd/watch.rs b/src/cmd/watch.rs
index 853e5ea..4666abc 100644
--- a/src/cmd/watch.rs
+++ b/src/cmd/watch.rs
@@ -77,13 +77,15 @@ impl crate::config::Config for Config {
                     let connector = tokio_tls::TlsConnector::from(connector);
                     let stream =
                         tokio::net::tcp::TcpStream::connect(&address);
-                    Box::new(stream.context(crate::error::Connect).and_then(
-                        move |stream| {
-                            connector
-                                .connect(&host, stream)
-                                .context(crate::error::ConnectTls)
-                        },
-                    ))
+                    Box::new(
+                        stream
+                            .context(crate::error::Connect { address })
+                            .and_then(move |stream| {
+                                connector.connect(&host, stream).context(
+                                    crate::error::ConnectTls { host },
+                                )
+                            }),
+                    )
                 })
             });
             Box::new(WatchSession::new(make_connector, &auth))
@@ -94,7 +96,7 @@ impl crate::config::Config for Config {
                 Box::new(move || {
                     Box::new(
                         tokio::net::tcp::TcpStream::connect(&address)
-                            .context(crate::error::Connect),
+                            .context(crate::error::Connect { address }),
                     )
                 })
             });
diff --git a/src/config.rs b/src/config.rs
index 86d72b4..5fed3d5 100644
--- a/src/config.rs
+++ b/src/config.rs
@@ -59,11 +59,14 @@ pub fn to_connect_address(
 ) -> Result<(String, std::net::SocketAddr)> {
     let mut address_parts = address.split(':');
     let host = address_parts.next().context(crate::error::ParseAddress)?;
-    let port = address_parts.next().context(crate::error::ParseAddress)?;
-    let port: u16 = port.parse().context(crate::error::ParsePort)?;
+    let port_str =
+        address_parts.next().context(crate::error::ParseAddress)?;
+    let port: u16 = port_str
+        .parse()
+        .context(crate::error::ParsePort { string: port_str })?;
     let socket_addr = (host, port)
         .to_socket_addrs()
-        .context(crate::error::ResolveAddress)?
+        .context(crate::error::ResolveAddress { host, port })?
         .next()
         .context(crate::error::HasResolvedAddr)?;
     Ok((host.to_string(), socket_addr))
diff --git a/src/dirs.rs b/src/dirs.rs
index d868a4c..44f828b 100644
--- a/src/dirs.rs
+++ b/src/dirs.rs
@@ -15,8 +15,12 @@ impl Dirs {
     }
 
     pub fn create_all(&self) -> Result<()> {
-        std::fs::create_dir_all(self.data_dir())
-            .context(crate::error::CreateDir)?;
+        let filename = self.data_dir();
+        std::fs::create_dir_all(filename).with_context(|| {
+            crate::error::CreateDir {
+                filename: filename.to_string_lossy(),
+            }
+        })?;
         Ok(())
     }
 
diff --git a/src/error.rs b/src/error.rs
index 4166cc4..7a3f226 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -4,19 +4,32 @@ pub enum Error {
     #[snafu(display("failed to accept: {}", source))]
     Acceptor { source: tokio::io::Error },
 
-    #[snafu(display("auth type not allowed: {:?}", ty))]
+    #[snafu(display("auth type {:?} not allowed", ty))]
     AuthTypeNotAllowed { ty: crate::protocol::AuthType },
 
-    #[snafu(display("failed to bind: {}", source))]
-    Bind { source: tokio::io::Error },
+    #[snafu(display("failed to bind to {}: {}", address, source))]
+    Bind {
+        address: std::net::SocketAddr,
+        source: tokio::io::Error,
+    },
 
-    #[snafu(display("failed to connect: {}", source))]
-    Connect { source: std::io::Error },
+    #[snafu(display("failed to connect to {}: {}", address, source))]
+    Connect {
+        address: std::net::SocketAddr,
+        source: std::io::Error,
+    },
 
-    #[snafu(display("failed to connect: {}", source))]
-    ConnectTls { source: native_tls::Error },
+    #[snafu(display(
+        "failed to make tls connection to {}: {}",
+        host,
+        source
+    ))]
+    ConnectTls {
+        host: String,
+        source: native_tls::Error,
+    },
 
-    #[snafu(display("couldn't find username"))]
+    #[snafu(display("couldn't determine the current username"))]
     CouldntFindUsername,
 
     #[snafu(display("failed to create tls acceptor: {}", source))]
@@ -25,11 +38,17 @@ pub enum Error {
     #[snafu(display("failed to create tls connector: {}", source))]
     CreateConnector { source: native_tls::Error },
 
-    #[snafu(display("failed to create directory: {}", source))]
-    CreateDir { source: std::io::Error },
+    #[snafu(display("failed to create directory {}: {}", filename, source))]
+    CreateDir {
+        filename: String,
+        source: std::io::Error,
+    },
 
-    #[snafu(display("failed to create file: {}", source))]
-    CreateFile { source: tokio::io::Error },
+    #[snafu(display("failed to create file {}: {}", filename, source))]
+    CreateFile {
+        filename: String,
+        source: tokio::io::Error,
+    },
 
     #[snafu(display("eof"))]
     EOF,
@@ -46,37 +65,40 @@ pub enum Error {
         // >
     },
 
-    #[snafu(display("failed to parse string: {:?}", data))]
+    #[snafu(display(
+        "failed to parse string {:?}: unexpected trailing data",
+        data
+    ))]
     ExtraMessageData { data: Vec<u8> },
 
     #[snafu(display("failed to write to stdout: {}", source))]
     FlushTerminal { source: tokio::io::Error },
 
-    #[snafu(display("failed to flush writes to terminal: {}", source))]
+    #[snafu(display("failed to write to stdout: {}", source))]
     FlushTerminalSync { source: std::io::Error },
 
     #[snafu(display(
         "failed to get recurse center profile data: {}",
         source
     ))]
-    GetProfile { source: reqwest::Error },
+    GetRecurseCenterProfile { source: reqwest::Error },
 
     #[snafu(display("failed to get terminal size: {}", source))]
     GetTerminalSize { source: crossterm::ErrorKind },
 
-    #[snafu(display("failed to find any resolved addresses"))]
+    #[snafu(display("failed to find any resolvable addresses"))]
     HasResolvedAddr,
 
-    #[snafu(display("invalid auth type: {}", ty))]
+    #[snafu(display("invalid auth type {}", ty))]
     InvalidAuthType { ty: u8 },
 
-    #[snafu(display("invalid auth type: {}", ty))]
+    #[snafu(display("invalid auth type {}", ty))]
     InvalidAuthTypeStr { ty: String },
 
-    #[snafu(display("invalid message type: {}", ty))]
+    #[snafu(display("invalid message type {}", ty))]
     InvalidMessageType { ty: u8 },
 
-    #[snafu(display("invalid watch id: {}", id))]
+    #[snafu(display("invalid watch id {}", id))]
     InvalidWatchId { id: String },
 
     #[snafu(display(
@@ -97,16 +119,22 @@ pub enum Error {
     MissingArgv,
 
     #[snafu(display(
-        "detected argv path was not a valid filename: {}",
+        "detected argv path {} was not a valid filename",
         path
     ))]
     NotAFileName { path: String },
 
-    #[snafu(display("failed to open file: {}", source))]
-    OpenFile { source: tokio::io::Error },
+    #[snafu(display("failed to open file {}: {}", filename, source))]
+    OpenFile {
+        filename: String,
+        source: tokio::io::Error,
+    },
 
-    #[snafu(display("failed to open identity file: {}", source))]
-    OpenIdentityFile { source: std::io::Error },
+    #[snafu(display("failed to open file {}: {}", filename, source))]
+    OpenFileSync {
+        filename: String,
+        source: std::io::Error,
+    },
 
     #[snafu(display("failed to open link in browser: {}", source))]
     OpenLink { source: std::io::Error },
@@ -123,7 +151,7 @@ pub enum Error {
     #[snafu(display("{}", source))]
     ParseArgs { source: clap::Error },
 
-    #[snafu(display("failed to parse buffer size '{}': {}", input, source))]
+    #[snafu(display("failed to parse buffer size {}: {}", input, source))]
     ParseBufferSize {
         input: String,
         source: std::num::ParseIntError,
@@ -151,32 +179,46 @@ pub enum Error {
     #[snafu(display("failed to parse identity file: {}", source))]
     ParseIdentity { source: native_tls::Error },
 
-    #[snafu(display("failed to parse int: {}", source))]
+    #[snafu(display(
+        "failed to parse int from buffer {:?}: {}",
+        buf,
+        source
+    ))]
     ParseInt {
+        buf: Vec<u8>,
         source: std::array::TryFromSliceError,
     },
 
     #[snafu(display("failed to parse response json: {}", source))]
     ParseJson { source: reqwest::Error },
 
-    #[snafu(display("failed to parse address: {}", source))]
-    ParsePort { source: std::num::ParseIntError },
-
     #[snafu(display(
-        "failed to parse read timeout '{}': {}",
-        input,
+        "failed to parse port {} from address: {}",
+        string,
         source
     ))]
+    ParsePort {
+        string: String,
+        source: std::num::ParseIntError,
+    },
+
+    #[snafu(display("failed to parse read timeout {}: {}", input, source))]
     ParseReadTimeout {
         input: String,
         source: std::num::ParseIntError,
     },
 
-    #[snafu(display("failed to parse string: {}", source))]
-    ParseString { source: std::string::FromUtf8Error },
+    #[snafu(display("failed to parse string {:?}: {}", string, source))]
+    ParseString {
+        string: Vec<u8>,
+        source: std::string::FromUtf8Error,
+    },
 
-    #[snafu(display("failed to parse url: {}", source))]
-    ParseUrl { source: url::ParseError },
+    #[snafu(display("failed to parse url {}: {}", url, source))]
+    ParseUrl {
+        url: String,
+        source: url::ParseError,
+    },
 
     #[snafu(display("failed to poll for process exit: {}", source))]
     ProcessExitPoll { source: std::io::Error },
@@ -197,8 +239,8 @@ pub enum Error {
     #[snafu(display("failed to read from file: {}", source))]
     ReadFile { source: tokio::io::Error },
 
-    #[snafu(display("failed to read identity file: {}", source))]
-    ReadIdentityFile { source: std::io::Error },
+    #[snafu(display("failed to read from file: {}", source))]
+    ReadFileSync { source: std::io::Error },
 
     #[snafu(display("{}", source))]
     ReadMessageWithTimeout {
@@ -207,10 +249,10 @@ pub enum Error {
     },
 
     #[snafu(display("failed to read packet: {}", source))]
-    ReadPacket { source: std::io::Error },
+    ReadPacketSync { source: std::io::Error },
 
     #[snafu(display("failed to read packet: {}", source))]
-    ReadPacketAsync { source: tokio::io::Error },
+    ReadPacket { source: tokio::io::Error },
 
     #[snafu(display("failed to read from pty: {}", source))]
     ReadPty { source: std::io::Error },
@@ -224,8 +266,17 @@ pub enum Error {
     #[snafu(display("failed to resize pty: {}", source))]
     ResizePty { source: std::io::Error },
 
-    #[snafu(display("failed to resolve address: {}", source))]
-    ResolveAddress { source: std::io::Error },
+    #[snafu(display(
+        "failed to resolve address {}:{}: {}",
+        host,
+        port,
+        source
+    ))]
+    ResolveAddress {
+        host: String,
+        port: u16,
+        source: std::io::Error,
+    },
 
     #[snafu(display(
         "failed to send oauth result back to main thread: {}",
@@ -328,10 +379,10 @@ pub enum Error {
     },
 
     #[snafu(display("failed to write packet: {}", source))]
-    WritePacket { source: std::io::Error },
+    WritePacketSync { source: std::io::Error },
 
     #[snafu(display("failed to write packet: {}", source))]
-    WritePacketAsync { source: tokio::io::Error },
+    WritePacket { source: tokio::io::Error },
 
     #[snafu(display("failed to write to pty: {}", source))]
     WritePty { source: std::io::Error },
diff --git a/src/oauth.rs b/src/oauth.rs
index 6264ce2..78ca392 100644
--- a/src/oauth.rs
+++ b/src/oauth.rs
@@ -79,8 +79,10 @@ pub fn save_client_auth_id(
 ) -> impl futures::future::Future<Item = (), Error = Error> {
     let id_file = client_id_file(auth);
     let id = id.to_string();
-    tokio::fs::File::create(id_file)
-        .context(crate::error::CreateFile)
+    tokio::fs::File::create(id_file.clone())
+        .with_context(move || crate::error::CreateFile {
+            filename: id_file.to_string_lossy().to_string(),
+        })
         .and_then(|file| {
             tokio::io::write_all(file, id).context(crate::error::WriteFile)
         })
@@ -113,8 +115,10 @@ fn cache_refresh_token(
         token.refresh_token().unwrap().secret(),
         token.access_token().secret(),
     );
-    let fut = tokio::fs::File::create(token_cache_file)
-        .context(crate::error::CreateFile)
+    let fut = tokio::fs::File::create(token_cache_file.clone())
+        .with_context(move || crate::error::CreateFile {
+            filename: token_cache_file.to_string_lossy().to_string(),
+        })
         .and_then(|file| {
             tokio::io::write_all(file, token_data)
                 .context(crate::error::WriteFile)
diff --git a/src/oauth/recurse_center.rs b/src/oauth/recurse_center.rs
index 273de66..b12b968 100644
--- a/src/oauth/recurse_center.rs
+++ b/src/oauth/recurse_center.rs
@@ -53,7 +53,7 @@ impl super::Oauth for RecurseCenter {
             .get("https://www.recurse.com/api/v1/profiles/me")
             .bearer_auth(token)
             .send()
-            .context(crate::error::GetProfile)
+            .context(crate::error::GetRecurseCenterProfile)
             .and_then(|mut res| res.json().context(crate::error::ParseJson))
             .map(|user: User| user.name());
         Box::new(fut)
diff --git a/src/protocol.rs b/src/protocol.rs
index 2827f76..f0b02fb 100644
--- a/src/protocol.rs
+++ b/src/protocol.rs
@@ -398,7 +398,7 @@ impl Packet {
     ) -> impl futures::future::Future<Item = (Self, FramedReader<T>), Error = Error>
     {
         r.0.into_future()
-            .map_err(|(e, _)| Error::ReadPacketAsync { source: e })
+            .map_err(|(e, _)| Error::ReadPacket { source: e })
             .and_then(|(data, r)| match data {
                 Some(data) => Ok((data, r)),
                 None => Err(Error::EOF),
@@ -434,7 +434,7 @@ impl Packet {
     {
         w.0.send(bytes::Bytes::from(self.as_bytes()))
             .map(FramedWriter)
-            .context(crate::error::WritePacketAsync)
+            .context(crate::error::WritePacket)
     }
 
     fn as_bytes(&self) -> Vec<u8> {
@@ -563,7 +563,7 @@ impl std::convert::TryFrom<Packet> for Message {
             }
             let (buf, rest) = data.split_at(std::mem::size_of::<u32>());
             let val = u32::from_be_bytes(
-                buf.try_into().context(crate::error::ParseInt)?,
+                buf.try_into().context(crate::error::ParseInt { buf })?,
             );
             Ok((val, rest))
         }
@@ -576,7 +576,7 @@ impl std::convert::TryFrom<Packet> for Message {
             }
             let (buf, rest) = data.split_at(std::mem::size_of::<u16>());
             let val = u16::from_be_bytes(
-                buf.try_into().context(crate::error::ParseInt)?,
+                buf.try_into().context(crate::error::ParseInt { buf })?,
             );
             Ok((val, rest))
         }
@@ -589,7 +589,7 @@ impl std::convert::TryFrom<Packet> for Message {
             }
             let (buf, rest) = data.split_at(std::mem::size_of::<u8>());
             let val = u8::from_be_bytes(
-                buf.try_into().context(crate::error::ParseInt)?,
+                buf.try_into().context(crate::error::ParseInt { buf })?,
             );
             Ok((val, rest))
         }
@@ -607,8 +607,11 @@ impl std::convert::TryFrom<Packet> for Message {
         }
         fn read_str(data: &[u8]) -> Result<(String, &[u8])> {
             let (bytes, rest) = read_bytes(data)?;
-            let val = String::from_utf8(bytes)
-                .context(crate::error::ParseString)?;
+            let val =
+                String::from_utf8(bytes).map_err(|e| Error::ParseString {
+                    string: e.as_bytes().to_vec(),
+                    source: e,
+                })?;
             Ok((val, rest))
         }
         fn read_size(data: &[u8]) -> Result<(crate::term::Size, &[u8])> {
diff --git a/src/server.rs b/src/server.rs
index 9b48381..b49a564 100644
--- a/src/server.rs
+++ b/src/server.rs
@@ -443,41 +443,42 @@ impl<S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + 'static>
                     let term_type = term_type.to_string();
                     let client = conn.oauth_client.take().unwrap();
                     let mut new_state = conn.state.clone();
-                    let fut =
-                        tokio::fs::File::open(client.server_token_file())
-                            .context(crate::error::OpenFile)
-                            .and_then(|file| {
-                                tokio::io::lines(std::io::BufReader::new(
-                                    file,
-                                ))
+                    let token_filename = client.server_token_file();
+                    let fut = tokio::fs::File::open(token_filename.clone())
+                        .with_context(move || crate::error::OpenFile {
+                            filename: token_filename
+                                .to_string_lossy()
+                                .to_string(),
+                        })
+                        .and_then(|file| {
+                            tokio::io::lines(std::io::BufReader::new(file))
                                 .into_future()
                                 .map_err(|(e, _)| e)
                                 .context(crate::error::ReadFile)
-                            })
-                            .and_then(|(refresh_token, _)| {
-                                // XXX unwrap here isn't super safe
-                                let refresh_token = refresh_token.unwrap();
-                                client
-                                    .get_access_token_from_refresh_token(
-                                        refresh_token.trim(),
-                                    )
-                                    .and_then(|access_token| {
-                                        client.get_username_from_access_token(
-                                            &access_token,
-                                        )
-                                    })
-                            })
-                            .map(move |username| {
-                                new_state.login_oauth(
-                                    &term_type, &size, &username,
-                                );
-                                (
-                                    new_state,
-                                    crate::protocol::Message::logged_in(
-                                        &username,
-                                    ),
+                        })
+                        .and_then(|(refresh_token, _)| {
+                            // XXX unwrap here isn't super safe
+                            let refresh_token = refresh_token.unwrap();
+                            client
+                                .get_access_token_from_refresh_token(
+                                    refresh_token.trim(),
                                 )
-                            });
+                                .and_then(|access_token| {
+                                    client.get_username_from_access_token(
+                                        &access_token,
+                                    )
+                                })
+                        })
+                        .map(move |username| {
+                            new_state
+                                .login_oauth(&term_type, &size, &username);
+                            (
+                                new_state,
+                                crate::protocol::Message::logged_in(
+                                    &username,
+                                ),
+                            )
+                        });
                     return Ok(Some(Box::new(fut)));
                 } else {
                     conn.state.login_oauth_start(term_type, &size);
@@ -1055,10 +1056,10 @@ fn classify_connection_error(
     if source.is_inner() {
         let source = source.into_inner().unwrap();
         let tokio_err = match source {
-            Error::ReadPacketAsync {
+            Error::ReadPacket {
                 source: ref tokio_err,
             } => tokio_err,
-            Error::WritePacketAsync {
+            Error::WritePacket {
                 source: ref tokio_err,
             } => tokio_err,
             Error::EOF => {