connect to the teleterm server on watch requests

6 files changed, 119 insertions, 36 deletions

diff --git a/Cargo.lock b/Cargo.lock
index 0341cbb..a9a0a77 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -818,6 +818,16 @@ dependencies = [
 ]
 
 [[package]]
+name = "gotham_derive"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b207d2aa120b2d7be4bb7edc48b199297ebce4227979127b0879890a84163480"
+dependencies = [
+ "quote 0.6.13",
+ "syn 0.15.44",
+]
+
+[[package]]
 name = "h2"
 version = "0.1.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -2095,6 +2105,7 @@ dependencies = [
  "env_logger",
  "futures",
  "gotham",
+ "gotham_derive",
  "hyper",
  "lazy-static-include",
  "lazy_static",
diff --git a/teleterm-web/src/model.rs b/teleterm-web/src/model.rs
index 9f7ab05..0e6f82d 100644
--- a/teleterm-web/src/model.rs
+++ b/teleterm-web/src/model.rs
@@ -33,7 +33,11 @@ impl Model {
         id: &str,
         orders: &mut impl Orders<crate::Msg>,
     ) {
-        let ws = crate::ws::connect(WATCH_URL, crate::Msg::Watch, orders);
+        let ws = crate::ws::connect(
+            &format!("{}?id={}", WATCH_URL, id),
+            crate::Msg::Watch,
+            orders,
+        );
         self.watch_conn = Some(WatchConn {
             id: id.to_string(),
             ws,
diff --git a/teleterm/Cargo.toml b/teleterm/Cargo.toml
index d9be990..40949ae 100644
--- a/teleterm/Cargo.toml
+++ b/teleterm/Cargo.toml
@@ -23,6 +23,7 @@ env_logger = "0.7"
 futures = "0.1.29"
 # for websocket support - should be able to go back to released version in 0.5
 gotham = { git = "https://github.com/gotham-rs/gotham", rev = "d2395926b93710832f8d72b49c9bd3e77516e386" }
+gotham_derive = "0.4"
 hyper = "0.12"
 lazy_static = "1"
 lazy-static-include = "2"
diff --git a/teleterm/src/web.rs b/teleterm/src/web.rs
index 202bc7e..af0e458 100644
--- a/teleterm/src/web.rs
+++ b/teleterm/src/web.rs
@@ -20,6 +20,15 @@ lazy_static_include::lazy_static_include_bytes!(
     "static/teleterm_web_bg.wasm"
 );
 
+#[derive(
+    serde::Deserialize,
+    gotham_derive::StateData,
+    gotham_derive::StaticResponseExtender,
+)]
+struct WatchQueryParams {
+    id: String,
+}
+
 pub struct Server {
     server: Box<dyn futures::Future<Item = (), Error = ()> + Send>,
 }
@@ -70,7 +79,10 @@ pub fn router() -> impl gotham::handler::NewHandler {
             .get("/teleterm_web_bg.wasm")
             .to(serve_static("application/wasm", &TELETERM_WEB_WASM));
         route.get("/list").to(handle_list);
-        route.get("/watch").to(handle_watch);
+        route
+            .get("/watch")
+            .with_query_string_extractor::<WatchQueryParams>()
+            .to(handle_watch);
     })
 }
 
@@ -238,8 +250,26 @@ fn handle_watch(
             .context(crate::error::WebSocketAccept)
             .map(|stream| stream.context(crate::error::WebSocket))
             .flatten_stream();
+
+        let query_params = WatchQueryParams::borrow_from(&state);
+        let address = "127.0.0.1:4144";
+        let (_, address) =
+            crate::config::to_connect_address(address).unwrap();
+        let connector: crate::client::Connector<_> = Box::new(move || {
+            Box::new(
+                tokio::net::tcp::TcpStream::connect(&address)
+                    .context(crate::error::Connect { address }),
+            )
+        });
+        let client = crate::client::Client::watch(
+            "teleterm-web",
+            connector,
+            &crate::protocol::Auth::plain("test"),
+            &query_params.id,
+        );
         let conn = Connection::new(
             gotham::state::request_id(&state),
+            client,
             Box::new(stream),
         );
 
@@ -263,19 +293,38 @@ type MessageStream = Box<
         > + Send,
 >;
 
-struct Connection {
+struct Connection<
+    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + 'static,
+> {
     id: String,
+    client: crate::client::Client<S>,
     stream: MessageStream,
 }
 
-impl Connection {
-    fn new(id: &str, stream: MessageStream) -> Self {
+impl<S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + 'static>
+    Connection<S>
+{
+    fn new(
+        id: &str,
+        client: crate::client::Client<S>,
+        stream: MessageStream,
+    ) -> Self {
         Self {
+            client,
             id: id.to_string(),
             stream,
         }
     }
 
+    fn handle_client_message(
+        &mut self,
+        msg: &crate::protocol::Message,
+    ) -> Result<()> {
+        // TODO
+        log::info!("teleterm client message for {}: {:?}", self.id, msg);
+        Ok(())
+    }
+
     fn handle_websocket_message(
         &mut self,
         msg: &tokio_tungstenite::tungstenite::protocol::Message,
@@ -286,7 +335,9 @@ impl Connection {
     }
 }
 
-impl Connection {
+impl<S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + 'static>
+    Connection<S>
+{
     const POLL_FNS:
         &'static [&'static dyn for<'a> Fn(
             &'a mut Self,
@@ -294,7 +345,21 @@ impl Connection {
             -> component_future::Poll<
             (),
             Error,
-        >] = &[&Self::poll_websocket_stream];
+        >] = &[&Self::poll_client, &Self::poll_websocket_stream];
+
+    fn poll_client(&mut self) -> component_future::Poll<(), Error> {
+        match component_future::try_ready!(self.client.poll()).unwrap() {
+            crate::client::Event::Disconnect => {
+                // TODO: better reconnect handling?
+                return Ok(component_future::Async::Ready(()));
+            }
+            crate::client::Event::Connect => {}
+            crate::client::Event::ServerMessage(msg) => {
+                self.handle_client_message(&msg)?;
+            }
+        }
+        Ok(component_future::Async::DidWork)
+    }
 
     fn poll_websocket_stream(&mut self) -> component_future::Poll<(), Error> {
         if let Some(msg) = component_future::try_ready!(self.stream.poll()) {
@@ -307,7 +372,9 @@ impl Connection {
     }
 }
 
-impl futures::Future for Connection {
+impl<S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + 'static>
+    futures::Future for Connection<S>
+{
     type Item = ();
     type Error = Error;
 
diff --git a/teleterm/static/teleterm_web.js b/teleterm/static/teleterm_web.js
index e8e2555..36cfe4b 100644
--- a/teleterm/static/teleterm_web.js
+++ b/teleterm/static/teleterm_web.js
@@ -18,31 +18,31 @@ function addHeapObject(obj) {
     return idx;
 }
 function __wbg_elem_binding0(arg0, arg1, arg2) {
-    wasm.__wbg_function_table.get(49)(arg0, arg1, addHeapObject(arg2));
+    wasm.__wbg_function_table.get(244)(arg0, arg1, addHeapObject(arg2));
 }
-function __wbg_elem_binding1(arg0, arg1, arg2) {
-    wasm.__wbg_function_table.get(49)(arg0, arg1, addHeapObject(arg2));
+function __wbg_elem_binding1(arg0, arg1) {
+    wasm.__wbg_function_table.get(174)(arg0, arg1);
 }
 function __wbg_elem_binding2(arg0, arg1, arg2) {
     wasm.__wbg_function_table.get(49)(arg0, arg1, addHeapObject(arg2));
 }
 function __wbg_elem_binding3(arg0, arg1, arg2) {
-    wasm.__wbg_function_table.get(49)(arg0, arg1, addHeapObject(arg2));
+    wasm.__wbg_function_table.get(54)(arg0, arg1, arg2);
 }
 function __wbg_elem_binding4(arg0, arg1, arg2) {
-    wasm.__wbg_function_table.get(54)(arg0, arg1, arg2);
+    wasm.__wbg_function_table.get(49)(arg0, arg1, addHeapObject(arg2));
 }
 function __wbg_elem_binding5(arg0, arg1, arg2) {
-    wasm.__wbg_function_table.get(242)(arg0, arg1, addHeapObject(arg2));
+    wasm.__wbg_function_table.get(49)(arg0, arg1, addHeapObject(arg2));
 }
-function __wbg_elem_binding6(arg0, arg1) {
-    wasm.__wbg_function_table.get(172)(arg0, arg1);
+function __wbg_elem_binding6(arg0, arg1, arg2) {
+    wasm.__wbg_function_table.get(49)(arg0, arg1, addHeapObject(arg2));
 }
 function __wbg_elem_binding7(arg0, arg1, arg2, arg3, arg4) {
-    wasm.__wbg_function_table.get(269)(arg0, arg1, addHeapObject(arg2), arg3, addHeapObject(arg4));
+    wasm.__wbg_function_table.get(271)(arg0, arg1, addHeapObject(arg2), arg3, addHeapObject(arg4));
 }
 function __wbg_elem_binding8(arg0, arg1, arg2, arg3) {
-    wasm.__wbg_function_table.get(273)(arg0, arg1, addHeapObject(arg2), addHeapObject(arg3));
+    wasm.__wbg_function_table.get(275)(arg0, arg1, addHeapObject(arg2), addHeapObject(arg3));
 }
 /**
 */
@@ -864,7 +864,7 @@ function init(module) {
     imports.wbg.__wbindgen_throw = function(arg0, arg1) {
         throw new Error(getStringFromWasm(arg0, arg1));
     };
-    imports.wbg.__wbindgen_closure_wrapper218 = function(arg0, arg1, arg2) {
+    imports.wbg.__wbindgen_closure_wrapper215 = function(arg0, arg1, arg2) {
         const state = { a: arg0, b: arg1, cnt: 1 };
         const real = (arg0) => {
             state.cnt++;
@@ -882,16 +882,16 @@ function init(module) {
         const ret = real;
         return addHeapObject(ret);
     };
-    imports.wbg.__wbindgen_closure_wrapper501 = function(arg0, arg1, arg2) {
+    imports.wbg.__wbindgen_closure_wrapper213 = function(arg0, arg1, arg2) {
         const state = { a: arg0, b: arg1, cnt: 1 };
-        const real = () => {
+        const real = (arg0) => {
             state.cnt++;
             const a = state.a;
             state.a = 0;
             try {
-                return __wbg_elem_binding6(a, state.b, );
+                return __wbg_elem_binding5(a, state.b, arg0);
             } finally {
-                if (--state.cnt === 0) wasm.__wbg_function_table.get(173)(a, state.b);
+                if (--state.cnt === 0) wasm.__wbg_function_table.get(50)(a, state.b);
                 else state.a = a;
             }
         }
@@ -900,16 +900,16 @@ function init(module) {
         const ret = real;
         return addHeapObject(ret);
     };
-    imports.wbg.__wbindgen_closure_wrapper903 = function(arg0, arg1, arg2) {
+    imports.wbg.__wbindgen_closure_wrapper503 = function(arg0, arg1, arg2) {
         const state = { a: arg0, b: arg1, cnt: 1 };
-        const real = (arg0) => {
+        const real = () => {
             state.cnt++;
             const a = state.a;
             state.a = 0;
             try {
-                return __wbg_elem_binding5(a, state.b, arg0);
+                return __wbg_elem_binding1(a, state.b, );
             } finally {
-                if (--state.cnt === 0) wasm.__wbg_function_table.get(243)(a, state.b);
+                if (--state.cnt === 0) wasm.__wbg_function_table.get(175)(a, state.b);
                 else state.a = a;
             }
         }
@@ -918,14 +918,14 @@ function init(module) {
         const ret = real;
         return addHeapObject(ret);
     };
-    imports.wbg.__wbindgen_closure_wrapper212 = function(arg0, arg1, arg2) {
+    imports.wbg.__wbindgen_closure_wrapper217 = function(arg0, arg1, arg2) {
         const state = { a: arg0, b: arg1, cnt: 1 };
         const real = (arg0) => {
             state.cnt++;
             const a = state.a;
             state.a = 0;
             try {
-                return __wbg_elem_binding0(a, state.b, arg0);
+                return __wbg_elem_binding3(a, state.b, arg0);
             } finally {
                 if (--state.cnt === 0) wasm.__wbg_function_table.get(50)(a, state.b);
                 else state.a = a;
@@ -936,14 +936,14 @@ function init(module) {
         const ret = real;
         return addHeapObject(ret);
     };
-    imports.wbg.__wbindgen_closure_wrapper220 = function(arg0, arg1, arg2) {
+    imports.wbg.__wbindgen_closure_wrapper219 = function(arg0, arg1, arg2) {
         const state = { a: arg0, b: arg1, cnt: 1 };
         const real = (arg0) => {
             state.cnt++;
             const a = state.a;
             state.a = 0;
             try {
-                return __wbg_elem_binding1(a, state.b, arg0);
+                return __wbg_elem_binding4(a, state.b, arg0);
             } finally {
                 if (--state.cnt === 0) wasm.__wbg_function_table.get(50)(a, state.b);
                 else state.a = a;
@@ -954,14 +954,14 @@ function init(module) {
         const ret = real;
         return addHeapObject(ret);
     };
-    imports.wbg.__wbindgen_closure_wrapper214 = function(arg0, arg1, arg2) {
+    imports.wbg.__wbindgen_closure_wrapper221 = function(arg0, arg1, arg2) {
         const state = { a: arg0, b: arg1, cnt: 1 };
         const real = (arg0) => {
             state.cnt++;
             const a = state.a;
             state.a = 0;
             try {
-                return __wbg_elem_binding3(a, state.b, arg0);
+                return __wbg_elem_binding6(a, state.b, arg0);
             } finally {
                 if (--state.cnt === 0) wasm.__wbg_function_table.get(50)(a, state.b);
                 else state.a = a;
@@ -972,16 +972,16 @@ function init(module) {
         const ret = real;
         return addHeapObject(ret);
     };
-    imports.wbg.__wbindgen_closure_wrapper216 = function(arg0, arg1, arg2) {
+    imports.wbg.__wbindgen_closure_wrapper905 = function(arg0, arg1, arg2) {
         const state = { a: arg0, b: arg1, cnt: 1 };
         const real = (arg0) => {
             state.cnt++;
             const a = state.a;
             state.a = 0;
             try {
-                return __wbg_elem_binding4(a, state.b, arg0);
+                return __wbg_elem_binding0(a, state.b, arg0);
             } finally {
-                if (--state.cnt === 0) wasm.__wbg_function_table.get(50)(a, state.b);
+                if (--state.cnt === 0) wasm.__wbg_function_table.get(245)(a, state.b);
                 else state.a = a;
             }
         }
diff --git a/teleterm/static/teleterm_web_bg.wasm b/teleterm/static/teleterm_web_bg.wasm
index ec5717b..2ee77b7 100644
--- a/teleterm/static/teleterm_web_bg.wasm
+++ b/teleterm/static/teleterm_web_bg.wasm