From 0583aba6bb7b901e907de68912480f4898b9aa3b Mon Sep 17 00:00:00 2001
From: Rasmus Thomsen <oss@cogitri.dev>
Date: Sat, 18 Jan 2020 11:16:24 +0100
Subject: [PATCH] refactor(*): use futures 0.3 & tokio 0.2

---
 Cargo.toml                |   7 +-
 src/client.rs             | 173 +++++++++++++++++++-------------------
 src/core.rs               |  12 +--
 src/frontend.rs           |  77 ++++++++---------
 src/protocol/client.rs    |  40 +++++----
 src/protocol/endpoint.rs  |  28 +++---
 src/protocol/server.rs    |  32 ++++---
 src/protocol/transport.rs |  22 ++---
 8 files changed, 195 insertions(+), 196 deletions(-)

diff --git a/Cargo.toml b/Cargo.toml
index 42ebe43..adb0c4b 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -16,12 +16,15 @@ edition = "2018"
 
 [dependencies]
 bytes = "0.4.12"
-futures = "0.1.27"
+futures = "0.3.1"
+futures-core = "0.3"
+futures-channel = "0.3"
+futures-util = "0.3"
 log = "0.4.6"
 serde = "1.0.92"
 serde_derive = "1.0.92"
 serde_json = "1.0.39"
-tokio = "0.1.21"
+tokio = { version="0.2.1", features=["rt-core"] }
 tokio-codec = "0.1.1"
 tokio-process = "0.2.3"
 syntect = { version = "3.2.0", default-features = false }
diff --git a/src/client.rs b/src/client.rs
index 0b44f6b..bd52b4b 100644
--- a/src/client.rs
+++ b/src/client.rs
@@ -36,7 +36,7 @@ impl Client {
         &self,
         method: &str,
         params: Value,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         info!(">>> notification: method={}, params={}", method, &params);
         self.0
             .notify(method, params)
@@ -50,7 +50,7 @@ impl Client {
         &self,
         method: &str,
         params: Value,
-    ) -> impl Future<Item = Value, Error = ClientError> {
+    ) -> impl Future<Output = Result<Value, ClientError>> {
         info!(">>> request : method={}, params={}", method, &params);
         self.0
             .request(method, params)
@@ -66,10 +66,10 @@ impl Client {
         view_id: ViewId,
         method: &str,
         params: Option<T>,
-    ) -> impl Future<Item = Value, Error = ClientError> {
+    ) -> impl Future<Output = Result<Value, ClientError>> {
         match get_edit_params(view_id, method, params) {
-            Ok(value) => Either::A(self.request("edit", value)),
-            Err(e) => Either::B(future::err(e)),
+            Ok(value) => Either::Left(self.request("edit", value)),
+            Err(e) => Either::Right(future::err(e)),
         }
     }
 
@@ -81,10 +81,10 @@ impl Client {
         view_id: ViewId,
         method: &str,
         params: Option<T>,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         match get_edit_params(view_id, method, params) {
-            Ok(value) => Either::A(self.notify("edit", value)),
-            Err(e) => Either::B(future::err(e)),
+            Ok(value) => Either::Left(self.notify("edit", value)),
+            Err(e) => Either::Right(future::err(e)),
         }
     }
 
@@ -98,7 +98,7 @@ impl Client {
         view_id: ViewId,
         first_line: u64,
         last_line: u64,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "scroll", Some(json!([first_line, last_line])))
     }
 
@@ -106,11 +106,11 @@ impl Client {
         &self,
         view_id: ViewId,
         line: u64,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "goto_line", Some(json!({ "line": line })))
     }
 
-    pub fn copy(&self, view_id: ViewId) -> impl Future<Item = Value, Error = ClientError> {
+    pub fn copy(&self, view_id: ViewId) -> impl Future<Output = Result<Value, ClientError>> {
         self.edit_request(view_id, "copy", None as Option<Value>)
     }
 
@@ -118,19 +118,19 @@ impl Client {
         &self,
         view_id: ViewId,
         buffer: &str,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "paste", Some(json!({ "chars": buffer })))
     }
 
-    pub fn cut(&self, view_id: ViewId) -> impl Future<Item = Value, Error = ClientError> {
+    pub fn cut(&self, view_id: ViewId) -> impl Future<Output = Result<Value, ClientError>> {
         self.edit_request(view_id, "cut", None as Option<Value>)
     }
 
-    pub fn undo(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn undo(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "undo", None as Option<Value>)
     }
 
-    pub fn redo(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn redo(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "redo", None as Option<Value>)
     }
 
@@ -141,7 +141,7 @@ impl Client {
         case_sensitive: bool,
         regex: bool,
         whole_words: bool,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(
             view_id,
             "find",
@@ -160,7 +160,7 @@ impl Client {
         wrap_around: bool,
         allow_same: bool,
         modify_selection: ModifySelection,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(
             view_id,
             command,
@@ -177,7 +177,7 @@ impl Client {
         wrap_around: bool,
         allow_same: bool,
         modify_selection: ModifySelection,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         self.find_other(
             view_id,
             "find_next",
@@ -193,7 +193,7 @@ impl Client {
         wrap_around: bool,
         allow_same: bool,
         modify_selection: ModifySelection,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         self.find_other(
             view_id,
             "find_previous",
@@ -203,7 +203,7 @@ impl Client {
         )
     }
 
-    pub fn find_all(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn find_all(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "find_all", None as Option<Value>)
     }
 
@@ -211,7 +211,7 @@ impl Client {
         &self,
         view_id: ViewId,
         visible: bool,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(
             view_id,
             "highlight_find",
@@ -219,11 +219,11 @@ impl Client {
         )
     }
 
-    pub fn left(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn left(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "move_left", None as Option<Value>)
     }
 
-    pub fn left_sel(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn left_sel(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(
             view_id,
             "move_left_and_modify_selection",
@@ -231,11 +231,11 @@ impl Client {
         )
     }
 
-    pub fn right(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn right(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "move_right", None as Option<Value>)
     }
 
-    pub fn right_sel(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn right_sel(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(
             view_id,
             "move_right_and_modify_selection",
@@ -243,11 +243,11 @@ impl Client {
         )
     }
 
-    pub fn up(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn up(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "move_up", None as Option<Value>)
     }
 
-    pub fn up_sel(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn up_sel(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(
             view_id,
             "move_up_and_modify_selection",
@@ -255,11 +255,11 @@ impl Client {
         )
     }
 
-    pub fn down(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn down(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "move_down", None as Option<Value>)
     }
 
-    pub fn down_sel(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn down_sel(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(
             view_id,
             "move_down_and_modify_selection",
@@ -267,30 +267,30 @@ impl Client {
         )
     }
 
-    pub fn backspace(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn backspace(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.del(view_id)
     }
 
-    pub fn delete(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn delete(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "delete_forward", None as Option<Value>)
     }
 
-    pub fn del(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn del(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "delete_backward", None as Option<Value>)
     }
 
     pub fn delete_word_backward(
         &self,
         view_id: ViewId,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "delete_word_backward", None as Option<Value>)
     }
 
-    pub fn page_up(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn page_up(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "scroll_page_up", None as Option<Value>)
     }
 
-    pub fn page_up_sel(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn page_up_sel(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(
             view_id,
             "page_up_and_modify_selection",
@@ -298,11 +298,11 @@ impl Client {
         )
     }
 
-    pub fn page_down(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn page_down(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "scroll_page_down", None as Option<Value>)
     }
 
-    pub fn page_down_sel(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn page_down_sel(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(
             view_id,
             "page_down_and_modify_selection",
@@ -310,11 +310,11 @@ impl Client {
         )
     }
 
-    pub fn line_start(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn line_start(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "move_to_left_end_of_line", None as Option<Value>)
     }
 
-    pub fn line_start_sel(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn line_start_sel(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(
             view_id,
             "move_to_left_end_of_line_and_modify_selection",
@@ -322,11 +322,11 @@ impl Client {
         )
     }
 
-    pub fn line_end(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn line_end(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "move_to_right_end_of_line", None as Option<Value>)
     }
 
-    pub fn line_end_sel(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn line_end_sel(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(
             view_id,
             "move_to_right_end_of_line_and_modify_selection",
@@ -334,7 +334,7 @@ impl Client {
         )
     }
 
-    pub fn document_begin(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn document_begin(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(
             view_id,
             "move_to_beginning_of_document",
@@ -345,7 +345,7 @@ impl Client {
     pub fn document_begin_sel(
         &self,
         view_id: ViewId,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(
             view_id,
             "move_to_beginning_of_document_and_modify_selection",
@@ -353,11 +353,14 @@ impl Client {
         )
     }
 
-    pub fn document_end(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn document_end(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "move_to_end_of_document", None as Option<Value>)
     }
 
-    pub fn document_end_sel(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn document_end_sel(
+        &self,
+        view_id: ViewId,
+    ) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(
             view_id,
             "move_to_end_of_document_and_modify_selection",
@@ -365,14 +368,14 @@ impl Client {
         )
     }
 
-    pub fn select_all(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn select_all(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "select_all", None as Option<Value>)
     }
 
     pub fn collapse_selections(
         &self,
         view_id: ViewId,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "collapse_selections", None as Option<Value>)
     }
 
@@ -380,27 +383,27 @@ impl Client {
         &self,
         view_id: ViewId,
         string: &str,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "insert", Some(json!({ "chars": string })))
     }
 
-    pub fn insert_newline(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn insert_newline(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "insert_newline", None as Option<Value>)
     }
 
-    pub fn insert_tab(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn insert_tab(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "insert_tab", None as Option<Value>)
     }
 
-    pub fn f1(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn f1(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "debug_rewrap", None as Option<Value>)
     }
 
-    pub fn f2(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn f2(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "debug_test_fg_spans", None as Option<Value>)
     }
 
-    pub fn char(&self, view_id: ViewId, ch: char) -> impl Future<Item = (), Error = ClientError> {
+    pub fn char(&self, view_id: ViewId, ch: char) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "insert", Some(json!({ "chars": ch })))
     }
 
@@ -410,7 +413,7 @@ impl Client {
         view_id: ViewId,
         line: u64,
         column: u64,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "click", Some(json!([line, column, 0, 1])))
     }
 
@@ -419,7 +422,7 @@ impl Client {
         view_id: ViewId,
         line: u64,
         column: u64,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         let ty = "point_select";
         self.edit_notify(
             view_id,
@@ -433,7 +436,7 @@ impl Client {
         view_id: ViewId,
         line: u64,
         column: u64,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         let ty = "toggle_sel";
         self.edit_notify(
             view_id,
@@ -447,7 +450,7 @@ impl Client {
         view_id: ViewId,
         line: u64,
         column: u64,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         let ty = "range_select";
         self.edit_notify(
             view_id,
@@ -461,7 +464,7 @@ impl Client {
         view_id: ViewId,
         line: u64,
         column: u64,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         let ty = "range_select";
         self.edit_notify(
             view_id,
@@ -475,7 +478,7 @@ impl Client {
         view_id: ViewId,
         line: u64,
         column: u64,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         let ty = "word_select";
         self.edit_notify(
             view_id,
@@ -489,7 +492,7 @@ impl Client {
         view_id: ViewId,
         line: u64,
         column: u64,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         let ty = "multi_line_select";
         self.edit_notify(
             view_id,
@@ -503,7 +506,7 @@ impl Client {
         view_id: ViewId,
         line: u64,
         column: u64,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         let ty = "multi_word_select";
         self.edit_notify(
             view_id,
@@ -517,7 +520,7 @@ impl Client {
         view_id: ViewId,
         line: u64,
         column: u64,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "drag", Some(json!([line, column, 0])))
     }
 
@@ -525,10 +528,7 @@ impl Client {
     /// ```ignore
     /// {"id":1,"method":"new_view","params":{"file_path":"foo/test.txt"}}
     /// ```
-    pub fn new_view(
-        &self,
-        file_path: Option<String>,
-    ) -> impl Future<Item = ViewId, Error = ClientError> {
+    pub fn new_view(&self, file_path: Option<String>) -> impl Future<Result<ViewId, ClientError>> {
         let params = if let Some(file_path) = file_path {
             json!({ "file_path": file_path })
         } else {
@@ -539,7 +539,7 @@ impl Client {
     }
 
     /// send a `"close_view"` notifycation to the core.
-    pub fn close_view(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn close_view(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.notify("close_view", json!({ "view_id": view_id }))
     }
 
@@ -547,12 +547,12 @@ impl Client {
         &self,
         view_id: ViewId,
         file_path: &str,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         let params = json!({"view_id": view_id, "file_path": file_path});
         self.notify("save", params).and_then(|_| Ok(()))
     }
 
-    pub fn set_theme(&self, theme: &str) -> impl Future<Item = (), Error = ClientError> {
+    pub fn set_theme(&self, theme: &str) -> impl Future<Output = Result<(), ClientError>> {
         let params = json!({ "theme_name": theme });
         self.notify("set_theme", params).and_then(|_| Ok(()))
     }
@@ -561,7 +561,7 @@ impl Client {
         &self,
         config_dir: Option<&str>,
         client_extras_dir: Option<&str>,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         let mut params = Map::new();
         if let Some(path) = config_dir {
             let _ = params.insert("config_dir".into(), json!(path));
@@ -576,7 +576,7 @@ impl Client {
         &self,
         view_id: ViewId,
         name: &str,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         let params = json!({"view_id": view_id, "plugin_name": name});
         self.notify("start", params).and_then(|_| Ok(()))
     }
@@ -585,7 +585,7 @@ impl Client {
         &self,
         view_id: ViewId,
         name: &str,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         let params = json!({"view_id": view_id, "plugin_name": name});
         self.notify("stop", params).and_then(|_| Ok(()))
     }
@@ -596,7 +596,7 @@ impl Client {
         plugin: &str,
         method: &str,
         params: &Value,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         let params = json!({
             "view_id": view_id,
             "receiver": plugin,
@@ -608,22 +608,25 @@ impl Client {
         self.notify("plugin_rpc", params).and_then(|_| Ok(()))
     }
 
-    pub fn outdent(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn outdent(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "outdent", None as Option<Value>)
     }
 
-    pub fn move_word_left(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn move_word_left(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "move_word_left", None as Option<Value>)
     }
 
-    pub fn move_word_right(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn move_word_right(
+        &self,
+        view_id: ViewId,
+    ) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "move_word_right", None as Option<Value>)
     }
 
     pub fn move_word_left_sel(
         &self,
         view_id: ViewId,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(
             view_id,
             "move_word_left_and_modify_selection",
@@ -634,7 +637,7 @@ impl Client {
     pub fn move_word_right_sel(
         &self,
         view_id: ViewId,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(
             view_id,
             "move_word_right_and_modify_selection",
@@ -647,7 +650,7 @@ impl Client {
         view_id: ViewId,
         width: i32,
         height: i32,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(
             view_id,
             "resize",
@@ -663,7 +666,7 @@ impl Client {
         view_id: ViewId,
         chars: &str,
         preserve_case: bool,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(
             view_id,
             "replace",
@@ -674,11 +677,11 @@ impl Client {
         )
     }
 
-    pub fn replace_next(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn replace_next(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "replace_next", None as Option<Value>)
     }
 
-    pub fn replace_all(&self, view_id: ViewId) -> impl Future<Item = (), Error = ClientError> {
+    pub fn replace_all(&self, view_id: ViewId) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(view_id, "replace_all", None as Option<Value>)
     }
 
@@ -686,7 +689,7 @@ impl Client {
         &self,
         view_id: ViewId,
         lang_name: &str,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         self.notify(
             "set_language",
             json!({ "view_id": view_id, "language_id": lang_name }),
@@ -698,7 +701,7 @@ impl Client {
         &self,
         domain: &str,
         changes: Value,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         self.notify(
             "modify_user_config",
             json!({
@@ -713,7 +716,7 @@ impl Client {
         view_id: ViewId,
         first_line: u64,
         last_line: u64,
-    ) -> impl Future<Item = (), Error = ClientError> {
+    ) -> impl Future<Output = Result<(), ClientError>> {
         self.edit_notify(
             view_id,
             "request_lines",
diff --git a/src/core.rs b/src/core.rs
index cba00e0..b2feed7 100644
--- a/src/core.rs
+++ b/src/core.rs
@@ -3,7 +3,8 @@ use crate::frontend::{Frontend, FrontendBuilder};
 use crate::protocol::Endpoint;
 use crate::ClientError;
 use bytes::BytesMut;
-use futures::{Future, Poll, Stream};
+use futures::{Future, Stream};
+use futures_core::task::{Context, Poll};
 use std::io::{self, Read, Write};
 use std::process::Command;
 use std::process::Stdio;
@@ -43,8 +44,8 @@ impl Write for Core {
 }
 
 impl AsyncWrite for Core {
-    fn shutdown(&mut self) -> Poll<(), io::Error> {
-        self.stdin.shutdown()
+    fn poll_shutdown(&mut self, cx: &mut Context) -> Poll<Result<(), io::Error>> {
+        self.stdin.poll_shutdown()
     }
 }
 
@@ -118,10 +119,9 @@ impl CoreStderr {
 }
 
 impl Stream for CoreStderr {
-    type Item = String;
-    type Error = io::Error;
+    type Item = Result<String, io::Error>;
 
-    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+    fn poll_next(&mut self, cx: &mut Context) -> Poll<Option<Self::Item>> {
         self.0.poll()
     }
 }
diff --git a/src/frontend.rs b/src/frontend.rs
index da0350d..8e6650c 100644
--- a/src/frontend.rs
+++ b/src/frontend.rs
@@ -6,7 +6,7 @@ use crate::structs::{
     ThemeChanged, Update, UpdateCmds,
 };
 use futures::{
-    future::{self, Either, FutureResult},
+    future::{self, Either},
     Future,
 };
 use serde_json::{from_value, to_value, Value};
@@ -34,10 +34,10 @@ pub enum XiNotification {
 /// The `Frontend` trait must be implemented by clients. It defines how the
 /// client handles notifications and requests coming from `xi-core`.
 pub trait Frontend {
-    type NotificationResult: IntoStaticFuture<Item = (), Error = ()>;
+    type NotificationResult: IntoStaticFuture<Output = Result<(), ()>>;
     fn handle_notification(&mut self, notification: XiNotification) -> Self::NotificationResult;
 
-    type MeasureWidthResult: IntoStaticFuture<Item = Vec<Vec<f32>>, Error = ()>;
+    type MeasureWidthResult: IntoStaticFuture<Output = Result<Vec<Vec<f32>>, ()>>;
     fn handle_measure_width(&mut self, request: MeasureWidth) -> Self::MeasureWidthResult;
 }
 
@@ -62,14 +62,11 @@ where
     }
 }
 
-impl<F: Frontend + Send> Service for F {
+impl<F: Frontend + Send + Sized> Service for F {
     type T = Value;
     type E = Value;
-    type RequestFuture = Box<dyn Future<Item = Self::T, Error = Self::E> + 'static + Send>;
-    type NotificationFuture = Either<
-        <<F as Frontend>::NotificationResult as IntoStaticFuture>::Future,
-        FutureResult<(), ()>,
-    >;
+    type RequestFuture = Box<dyn Future<Output = Result<Self::T, Self::E>> + 'static + Send>;
+    type NotificationFuture = <<F as Frontend>::NotificationResult as IntoStaticFuture>::Future;
 
     fn handle_request(&mut self, method: &str, params: Value) -> Self::RequestFuture {
         info!("<<< request: method={}, params={}", method, &params);
@@ -111,158 +108,158 @@ impl<F: Frontend + Send> Service for F {
         info!("<<< notification: method={}, params={}", method, &params);
         match method {
             "update" => match from_value::<Update>(params) {
-                Ok(update) => Either::A(
+                Ok(update) => Either::Left(
                     self.handle_notification(XiNotification::Update(update))
                         .into_static_future(),
                 ),
                 Err(e) => {
                     error!("received invalid update notification: {:?}", e);
-                    Either::B(future::err(()))
+                    Either::Right(future::err(()))
                 }
             },
 
             "scroll_to" => match from_value::<ScrollTo>(params) {
-                Ok(scroll_to) => Either::A(
+                Ok(scroll_to) => Either::Left(
                     self.handle_notification(XiNotification::ScrollTo(scroll_to))
                         .into_static_future(),
                 ),
                 Err(e) => {
                     error!("received invalid scroll_to notification: {:?}", e);
-                    Either::B(future::err(()))
+                    Either::Right(future::err(()))
                 }
             },
 
             "def_style" => match from_value::<Style>(params) {
-                Ok(style) => Either::A(
+                Ok(style) => Either::Left(
                     self.handle_notification(XiNotification::DefStyle(style))
                         .into_static_future(),
                 ),
                 Err(e) => {
                     error!("received invalid def_style notification: {:?}", e);
-                    Either::B(future::err(()))
+                    Either::Right(future::err(()))
                 }
             },
             "available_plugins" => match from_value::<AvailablePlugins>(params) {
-                Ok(plugins) => Either::A(
+                Ok(plugins) => Either::Left(
                     self.handle_notification(XiNotification::AvailablePlugins(plugins))
                         .into_static_future(),
                 ),
                 Err(e) => {
                     error!("received invalid available_plugins notification: {:?}", e);
-                    Either::B(future::err(()))
+                    Either::Right(future::err(()))
                 }
             },
             "plugin_started" => match from_value::<PluginStarted>(params) {
-                Ok(plugin) => Either::A(
+                Ok(plugin) => Either::Left(
                     self.handle_notification(XiNotification::PluginStarted(plugin))
                         .into_static_future(),
                 ),
                 Err(e) => {
                     error!("received invalid plugin_started notification: {:?}", e);
-                    Either::B(future::err(()))
+                    Either::Right(future::err(()))
                 }
             },
             "plugin_stoped" => match from_value::<PluginStoped>(params) {
-                Ok(plugin) => Either::A(
+                Ok(plugin) => Either::Left(
                     self.handle_notification(XiNotification::PluginStoped(plugin))
                         .into_static_future(),
                 ),
                 Err(e) => {
                     error!("received invalid plugin_stoped notification: {:?}", e);
-                    Either::B(future::err(()))
+                    Either::Right(future::err(()))
                 }
             },
             "update_cmds" => match from_value::<UpdateCmds>(params) {
-                Ok(cmds) => Either::A(
+                Ok(cmds) => Either::Left(
                     self.handle_notification(XiNotification::UpdateCmds(cmds))
                         .into_static_future(),
                 ),
                 Err(e) => {
                     error!("received invalid update_cmds notification: {:?}", e);
-                    Either::B(future::err(()))
+                    Either::Right(future::err(()))
                 }
             },
             "config_changed" => match from_value::<ConfigChanged>(params) {
-                Ok(config) => Either::A(
+                Ok(config) => Either::Left(
                     self.handle_notification(XiNotification::ConfigChanged(config))
                         .into_static_future(),
                 ),
                 Err(e) => {
                     error!("received invalid config_changed notification: {:?}", e);
-                    Either::B(future::err(()))
+                    Either::Right(future::err(()))
                 }
             },
             "theme_changed" => match from_value::<ThemeChanged>(params) {
-                Ok(theme) => Either::A(
+                Ok(theme) => Either::Left(
                     self.handle_notification(XiNotification::ThemeChanged(theme))
                         .into_static_future(),
                 ),
                 Err(e) => {
                     error!("received invalid theme_changed notification: {:?}", e);
-                    Either::B(future::err(()))
+                    Either::Right(future::err(()))
                 }
             },
             "alert" => match from_value::<Alert>(params) {
-                Ok(alert) => Either::A(
+                Ok(alert) => Either::Left(
                     self.handle_notification(XiNotification::Alert(alert))
                         .into_static_future(),
                 ),
                 Err(e) => {
                     error!("received invalid alert notification: {:?}", e);
-                    Either::B(future::err(()))
+                    Either::Right(future::err(()))
                 }
             },
             "available_themes" => match from_value::<AvailableThemes>(params) {
-                Ok(themes) => Either::A(
+                Ok(themes) => Either::Left(
                     self.handle_notification(XiNotification::AvailableThemes(themes))
                         .into_static_future(),
                 ),
                 Err(e) => {
                     error!("received invalid available_themes notification: {:?}", e);
-                    Either::B(future::err(()))
+                    Either::Right(future::err(()))
                 }
             },
             "find_status" => match from_value::<FindStatus>(params) {
-                Ok(find_status) => Either::A(
+                Ok(find_status) => Either::Left(
                     self.handle_notification(XiNotification::FindStatus(find_status))
                         .into_static_future(),
                 ),
                 Err(e) => {
                     error!("received invalid find_status notification: {:?}", e);
-                    Either::B(future::err(()))
+                    Either::Right(future::err(()))
                 }
             },
             "replace_status" => match from_value::<ReplaceStatus>(params) {
-                Ok(replace_status) => Either::A(
+                Ok(replace_status) => Either::Left(
                     self.handle_notification(XiNotification::ReplaceStatus(replace_status))
                         .into_static_future(),
                 ),
                 Err(e) => {
                     error!("received invalid replace_status notification: {:?}", e);
-                    Either::B(future::err(()))
+                    Either::Right(future::err(()))
                 }
             },
             "available_languages" => match from_value::<AvailableLanguages>(params) {
-                Ok(available_langs) => Either::A(
+                Ok(available_langs) => Either::Left(
                     self.handle_notification(XiNotification::AvailableLanguages(available_langs))
                         .into_static_future(),
                 ),
                 Err(e) => {
                     error!("received invalid available_languages notification: {:?}", e);
-                    Either::B(future::err(()))
+                    Either::Right(future::err(()))
                 }
             },
             "language_changed" => match from_value::<LanguageChanged>(params) {
-                Ok(lang) => Either::A(
+                Ok(lang) => Either::Left(
                     self.handle_notification(XiNotification::LanguageChanged(lang))
                         .into_static_future(),
                 ),
                 Err(e) => {
                     error!("received invalid language_changed notification: {:?}", e);
-                    Either::B(future::err(()))
+                    Either::Right(future::err(()))
                 }
             },
-            _ => Either::B(future::err(())),
+            _ => Either::Right(future::err(())),
         }
     }
 }
diff --git a/src/protocol/client.rs b/src/protocol/client.rs
index b8073e9..ea270c4 100644
--- a/src/protocol/client.rs
+++ b/src/protocol/client.rs
@@ -1,8 +1,9 @@
 use std::collections::HashMap;
 use std::io;
 
-use futures::sync::{mpsc, oneshot};
-use futures::{Async, Future, Poll, Stream};
+use futures::{Future, Stream};
+use futures_channel::{mpsc, oneshot};
+use futures_core::task::Poll;
 use serde_json::Value;
 use tokio::io::{AsyncRead, AsyncWrite};
 
@@ -23,10 +24,9 @@ type AckTx = oneshot::Sender<()>;
 pub struct Response(oneshot::Receiver<Result<Value, Value>>);
 
 impl Future for Response {
-    type Item = Result<Value, Value>;
-    type Error = RpcError;
+    type Output = Result<Result<Value, Value>, RpcError>;
 
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+    fn poll(&mut self) -> Poll<Self::Output> {
         self.0
             .poll()
             .map_err(|oneshot::Canceled| RpcError::ResponseCanceled)
@@ -39,10 +39,9 @@ impl Future for Response {
 pub struct Ack(oneshot::Receiver<()>);
 
 impl Future for Ack {
-    type Item = ();
-    type Error = RpcError;
+    type Output = Result<(), RpcError>;
 
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+    fn poll(&mut self) -> Poll<Self::Output> {
         self.0
             .poll()
             .map_err(|oneshot::Canceled| RpcError::AckCanceled)
@@ -93,7 +92,7 @@ impl InnerClient {
         trace!("polling shutdown signal channel");
         loop {
             match self.shutdown_rx.poll() {
-                Ok(Async::Ready(Some(()))) => {
+                Ok(Poll::Ready(Some(()))) => {
                     info!("Received shutdown signal");
                     self.shutdown();
                     // Note that in theory, we should continue polling
@@ -103,12 +102,12 @@ impl InnerClient {
                     // are being polled or not.
                     break;
                 }
-                Ok(Async::Ready(None)) => {
+                Ok(Poll::Ready(None)) => {
                     warn!("client closed the shutdown signal channel");
                     self.shutdown();
                     break;
                 }
-                Ok(Async::NotReady) => {
+                Ok(Poll::NotReady) => {
                     trace!("no shutdown signal from client");
                     break;
                 }
@@ -124,16 +123,16 @@ impl InnerClient {
         trace!("polling client notifications channel");
         loop {
             match self.notifications_rx.poll() {
-                Ok(Async::Ready(Some((notification, ack_sender)))) => {
+                Ok(Poll::Ready(Some((notification, ack_sender)))) => {
                     trace!("sending notification: {:?}", notification);
                     stream.send(Message::Notification(notification));
                     self.pending_notifications.push(ack_sender);
                 }
-                Ok(Async::NotReady) => {
+                Ok(Poll::NotReady) => {
                     trace!("no new notification from client");
                     break;
                 }
-                Ok(Async::Ready(None)) => {
+                Ok(Poll::Ready(None)) => {
                     warn!("client closed the notifications channel");
                     self.shutdown();
                     break;
@@ -152,7 +151,7 @@ impl InnerClient {
         trace!("polling client requests channel");
         loop {
             match self.requests_rx.poll() {
-                Ok(Async::Ready(Some((mut request, response_sender)))) => {
+                Ok(Poll::Ready(Some((mut request, response_sender)))) => {
                     self.request_id += 1;
                     trace!("sending request: {:?}", request);
                     request.id = self.request_id;
@@ -160,12 +159,12 @@ impl InnerClient {
                     self.pending_requests
                         .insert(self.request_id, response_sender);
                 }
-                Ok(Async::Ready(None)) => {
+                Ok(Poll::Ready(None)) => {
                     warn!("client closed the requests channel.");
                     self.shutdown();
                     break;
                 }
-                Ok(Async::NotReady) => {
+                Ok(Poll::NotReady) => {
                     trace!("no new request from client");
                     break;
                 }
@@ -273,10 +272,9 @@ impl Client {
 }
 
 impl Future for Client {
-    type Item = ();
-    type Error = io::Error;
+    type Output = Result<(), io::Error>;
 
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        Ok(Async::Ready(()))
+    fn poll(&mut self) -> Poll<Self::Output> {
+        Ok(Poll::Ready(()))
     }
 }
diff --git a/src/protocol/endpoint.rs b/src/protocol/endpoint.rs
index eb8ff03..ba3cf97 100644
--- a/src/protocol/endpoint.rs
+++ b/src/protocol/endpoint.rs
@@ -1,6 +1,7 @@
 use std::io;
 
-use futures::{Async, Future, Poll, Sink, Stream};
+use futures::{Future, Sink, Stream};
+use futures_core::task::Poll;
 use tokio::io::{AsyncRead, AsyncWrite};
 
 use super::client::Client;
@@ -43,8 +44,8 @@ where
     fn flush(&mut self) {
         trace!("flushing stream");
         match self.stream.poll_complete() {
-            Ok(Async::Ready(())) => self.client.acknowledge_notifications(),
-            Ok(Async::NotReady) => (),
+            Ok(Poll::Ready(())) => self.client.acknowledge_notifications(),
+            Ok(Poll::NotReady) => (),
             Err(e) => panic!("Failed to flush the sink: {:?}", e),
         }
     }
@@ -54,19 +55,18 @@ impl<S, T: AsyncRead + AsyncWrite> Future for Endpoint<S, T>
 where
     S: Service,
 {
-    type Item = ();
-    type Error = io::Error;
+    type Output = Result<(), io::Error>;
 
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+    fn poll(&mut self) -> Poll<Self::Output> {
         trace!("polling stream");
         loop {
             match self.stream.poll()? {
-                Async::Ready(Some(msg)) => self.handle_message(msg),
-                Async::Ready(None) => {
+                Poll::Ready(Some(msg)) => self.handle_message(msg),
+                Poll::Ready(None) => {
                     warn!("stream closed by remote peer.");
-                    return Ok(Async::Ready(()));
+                    return Ok(Poll::Ready(()));
                 }
-                Async::NotReady => {
+                Poll::NotReady => {
                     trace!("no new message in the stream");
                     break;
                 }
@@ -81,8 +81,8 @@ where
         // Note that errors from poll_complete() are usually fatal,
         // hence the early return. See:
         // https://docs.rs/tokio/0.1.21/tokio/prelude/trait.Sink.html#errors-1
-        if let Async::NotReady = self.server.send_responses(&mut self.stream)? {
-            return Ok(Async::NotReady);
+        if let Poll::NotReady = self.server.send_responses(&mut self.stream)? {
+            return Ok(Poll::NotReady);
         }
 
         let mut client_shutdown = false;
@@ -96,9 +96,9 @@ where
 
         self.flush();
         if client_shutdown {
-            Ok(Async::Ready(()))
+            Ok(Poll::Ready(()))
         } else {
-            Ok(Async::NotReady)
+            Ok(Poll::NotReady)
         }
     }
 }
diff --git a/src/protocol/server.rs b/src/protocol/server.rs
index a25a12f..9fc5de4 100644
--- a/src/protocol/server.rs
+++ b/src/protocol/server.rs
@@ -1,7 +1,9 @@
 use std::io;
 
-use futures::sync::mpsc;
-use futures::{Async, Future, IntoFuture, Poll, Sink, Stream};
+use futures::{Future, Sink, Stream};
+use futures_channel::mpsc;
+use futures_core::task::Poll;
+use futures_util::future::IntoFuture;
 use serde_json::Value;
 use tokio::io::{AsyncRead, AsyncWrite};
 
@@ -13,8 +15,8 @@ use super::transport::Transport;
 pub trait Service: Send {
     type T: Into<Value> + Send + 'static;
     type E: Into<Value> + Send + 'static;
-    type RequestFuture: IntoStaticFuture<Item = Self::T, Error = Self::E>;
-    type NotificationFuture: IntoStaticFuture<Item = (), Error = ()>;
+    type RequestFuture: IntoStaticFuture;
+    type NotificationFuture: IntoStaticFuture;
 
     fn handle_request(&mut self, method: &str, params: Value) -> Self::RequestFuture;
 
@@ -45,10 +47,10 @@ impl<S: Service> Server<S> {
     pub fn send_responses<T: AsyncRead + AsyncWrite>(
         &mut self,
         sink: &mut Transport<T>,
-    ) -> Poll<(), io::Error> {
+    ) -> Poll<()> {
         trace!("Server: flushing responses");
         while let Ok(poll) = self.pending_responses.poll() {
-            if let Async::Ready(Some((id, result))) = poll {
+            if let Poll::Ready(Some((id, result))) = poll {
                 let msg = Message::Response(ResponseMessage {
                     id,
                     result: result.map(Into::into).map_err(Into::into),
@@ -57,7 +59,7 @@ impl<S: Service> Server<S> {
                 // don't panic here.
                 sink.start_send(msg).unwrap();
             } else {
-                if let Async::Ready(None) = poll {
+                if let Poll::Ready(None) = poll {
                     panic!("we store the sender, it can't be dropped");
                 }
 
@@ -114,25 +116,21 @@ pub trait ServiceBuilder {
 /// in that the returned future has the `'static` lifetime.
 pub trait IntoStaticFuture {
     /// The future that this type can be converted into.
-    type Future: Future<Item = Self::Item, Error = Self::Error> + 'static + Send;
+    type Future: Future<Output = Self::Output> + 'static + Send;
     /// The item that the future may resolve with.
-    type Item;
-    /// The error that the future may resolve with.
-    type Error;
+    type Output;
 
     /// Consumes this object and produces a future.
     fn into_static_future(self) -> Self::Future;
 }
 
-impl<F: IntoFuture> IntoStaticFuture for F
+impl<F: Future> IntoStaticFuture for F
 where
-    <F as IntoFuture>::Future: 'static + Send,
+    F: 'static + Send,
 {
-    type Future = <F as IntoFuture>::Future;
-    type Item = <F as IntoFuture>::Item;
-    type Error = <F as IntoFuture>::Error;
+    type Output = F::Output;
 
     fn into_static_future(self) -> Self::Future {
-        self.into_future()
+        IntoFuture::new(self)
     }
 }
diff --git a/src/protocol/transport.rs b/src/protocol/transport.rs
index b64dc75..fb8f6d1 100644
--- a/src/protocol/transport.rs
+++ b/src/protocol/transport.rs
@@ -1,6 +1,8 @@
 use std::io;
 
-use futures::{AsyncSink, Poll, Sink, StartSend, Stream};
+use futures::task::Context;
+use futures::{Sink, Stream};
+use futures_core::task::Poll;
 use tokio::io::{AsyncRead, AsyncWrite};
 use tokio_codec::{Decoder, Framed};
 
@@ -20,9 +22,9 @@ where
     pub fn send(&mut self, message: Message) {
         debug!("sending message to remote peer: {:?}", message);
         match self.start_send(message) {
-            Ok(AsyncSink::Ready) => (),
+            Ok(Poll::Ready(Ok())) => (),
             // FIXME: there should probably be a retry mechanism.
-            Ok(AsyncSink::NotReady(_message)) => panic!("The sink is full."),
+            Ok(Poll::NotReady(_message)) => panic!("The sink is full."),
             Err(e) => panic!("An error occured while trying to send message: {:?}", e),
         }
     }
@@ -33,25 +35,23 @@ where
     T: AsyncRead + AsyncWrite,
 {
     type Item = Message;
-    type Error = io::Error;
 
-    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+    fn poll_next(&mut self, cx: &mut Context) -> Poll<Option<Self::Item>> {
         self.0.poll()
     }
 }
 
-impl<T> Sink for Transport<T>
+impl<T> Sink<Message> for Transport<T>
 where
     T: AsyncRead + AsyncWrite,
 {
-    type SinkItem = Message;
-    type SinkError = io::Error;
+    type Error = io::Error;
 
-    fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
+    fn start_send(&mut self, item: Message) -> Result<Message, Self::Error> {
         self.0.start_send(item)
     }
 
-    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
-        self.0.poll_complete()
+    fn poll_close(&mut self) -> Poll<Result<(), Self::Error>> {
+        self.0.poll_close()
     }
 }