feat: More complete disconnect and notif support.

* Properly propagate disconnect from Rust to TS.
* Add more code to exercise notification readable.

Author: petejohanson

src-tauri/src/transport/commands.rs

@@ -18,7 +18,7 @@ pub struct AvailableDevice {
     pub id: String
 }
 
-#[derive(Default)]
+#[derive(Debug, Default)]
 pub struct ActiveConnection<'a> { pub conn: Mutex<Option<Box<dyn Sink<Vec<u8>, Error = SendError> + Unpin + Send + 'a>>> }
 
 #[command]

src-tauri/src/transport/gatt.rs

@@ -51,11 +51,16 @@ pub async fn gatt_connect(
                     // Need to keep adapter from being dropped while active/connected
                     let a = adapter;
 
-                    while let Some(Ok(vn)) = n.next().await {
-                        use tauri::Manager;
+                    use tauri::Manager;
 
+                    while let Some(Ok(vn)) = n.next().await {
                         app_handle.emit("connection_data", vn.clone());
                     }
+
+                    let state = app_handle.state::<super::commands::ActiveConnection>();
+                    *state.conn.lock().await = None;
+
+                    app_handle.emit("connection_disconnected", ());
                 }
             });
 
@@ -89,10 +94,10 @@ pub async fn gatt_list_devices() -> Result<Vec<super::commands::AvailableDevice>
         .take_until(async_std::task::sleep(Duration::from_secs(2)))
         .filter_map(|d| ready(d.ok()))
         .then(move |device| async move {
-                let label = device.name_async().await.unwrap_or("Unknown".to_string());
-                let id = serde_json::to_string(&device.id()).unwrap();
+            let label = device.name_async().await.unwrap_or("Unknown".to_string());
+            let id = serde_json::to_string(&device.id()).unwrap();
 
-                super::commands::AvailableDevice { label, id }
+            super::commands::AvailableDevice { label, id }
         })
         .collect::<Vec<_>>()
         .await;

src-tauri/src/transport/serial.rs

@@ -1,69 +1,87 @@
-
 use blocking::unblock;
-use futures::stream;
 use futures::channel::mpsc::channel;
-use futures::StreamExt;
 use futures::lock::Mutex;
+use futures::stream;
+use futures::StreamExt;
 
-use std::{
-    time::Duration,
-};
+use std::time::Duration;
 
-use tokio::io::{ AsyncWriteExt, AsyncReadExt };
-use tokio_serial::{available_ports, SerialPort, SerialPortType, SerialPortBuilderExt};
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio_serial::{available_ports, SerialPort, SerialPortBuilderExt, SerialPortType};
 
 use serde::{Deserialize, Serialize};
 use tauri::{
-  AppHandle,
-  command,
-  ipc::{Request, Response},
-  State, Window,
+    command,
+    ipc::{Request, Response},
+    AppHandle, State, Window,
 };
 
 const READ_BUF_SIZE: usize = 1024;
 
 #[command]
-pub async fn serial_connect(id: String, app_handle: AppHandle, state: State<'_, super::commands::ActiveConnection<'_>>) -> Result<bool, ()> {
-  match tokio_serial::new(id, 9600).open_native_async() {
-    Ok(mut port) => {
-      #[cfg(unix)]
-      port.set_exclusive(false)
-          .expect("Unable to set serial port exclusive to false");
-    
-      let (mut reader, mut writer) = tokio::io::split(port);
-    
-      let (send, mut recv) = channel(5);
-      *state.conn.lock().await = Some(Box::new(send));
-      tauri::async_runtime::spawn(async move {
-        while let Some(data) = recv.next().await {
-          writer.write(&data).await;
-        }
-      });
-    
-      tauri::async_runtime::spawn(async move {
-              let mut buffer = vec![0; READ_BUF_SIZE];
-              while let Ok(size) = reader.read(&mut buffer).await {
-                if size > 0 {
-                  use tauri::Manager;
-                  app_handle.emit("connection_data", &buffer[..size]);
+pub async fn serial_connect(
+    id: String,
+    app_handle: AppHandle,
+    state: State<'_, super::commands::ActiveConnection<'_>>,
+) -> Result<bool, ()> {
+    match tokio_serial::new(id, 9600).open_native_async() {
+        Ok(mut port) => {
+            #[cfg(unix)]
+            port.set_exclusive(false)
+                .expect("Unable to set serial port exclusive to false");
+
+            let (mut reader, mut writer) = tokio::io::split(port);
+
+            let (send, mut recv) = channel(5);
+            *state.conn.lock().await = Some(Box::new(send));
+            tauri::async_runtime::spawn(async move {
+                while let Some(data) = recv.next().await {
+                    writer.write(&data).await;
                 }
-              }
-      });
-  
-      Ok(true)
-    },
-    Err(_) => Err(())
-  }
+            });
+
+            tauri::async_runtime::spawn(async move {
+                use tauri::Manager;
+
+                let mut buffer = vec![0; READ_BUF_SIZE];
+                while let Ok(size) = reader.read(&mut buffer).await {
+                    if size > 0 {
+                        app_handle.emit("connection_data", &buffer[..size]);
+                    } else {
+                        break;
+                    }
+                }
+
+                let state = app_handle.state::<super::commands::ActiveConnection>();
+                *state.conn.lock().await = None;
+
+                app_handle.emit("connection_disconnected", ());
+            });
+
+            Ok(true)
+        }
+        Err(_) => Err(()),
+    }
 }
 
 #[command]
-pub async fn serial_list_devices(
-) -> Result<Vec<super::commands::AvailableDevice>, ()> {
-  let ports = unblock(|| available_ports()).await.unwrap();
+pub async fn serial_list_devices() -> Result<Vec<super::commands::AvailableDevice>, ()> {
+    let ports = unblock(|| available_ports()).await.unwrap();
 
-  let candidates = ports.into_iter().filter_map(|pi| {println!("port {:?}", pi); match pi.port_type { SerialPortType::UsbPort(u) => Some(super::commands::AvailableDevice {id: pi.port_name, label: u.product.unwrap_or("TODO".to_string())}), _ => None }}).collect();
+    let candidates = ports
+        .into_iter()
+        .filter_map(|pi| {
+            println!("port {:?}", pi);
+            match pi.port_type {
+                SerialPortType::UsbPort(u) => Some(super::commands::AvailableDevice {
+                    id: pi.port_name,
+                    label: u.product.unwrap_or("TODO".to_string()),
+                }),
+                _ => None,
+            }
+        })
+        .collect();
 
-  println!("Candidates {:?}", candidates);
-  Ok(candidates)
+    println!("Candidates {:?}", candidates);
+    Ok(candidates)
 }
-

src/RpcTest.tsx

@@ -4,6 +4,7 @@ import { useState, Dispatch } from 'react';
 import { create_rpc_connection, call_rpc, Request } from "ts-zmk-rpc-core";
 
 import type { RpcTransport } from "ts-zmk-rpc-core/transport/index";
+import type { Notification } from "ts-zmk-rpc-core/transport/core";
 import type { PhysicalLayout, Keymap } from "ts-zmk-rpc-core/keymap";
 import type { GetBehaviorDetailsResponse } from "ts-zmk-rpc-core/behaviors";
 import { connect as gatt_connect } from "ts-zmk-rpc-core/transport/gatt";
@@ -30,6 +31,26 @@ const TRANSPORTS: TransportFactory[] = [
   ... window.__TAURI_INTERNALS__ ? [{ label: "Serial", pick_and_connect: { connect: tauri_serial_connect, list: serial_list_devices }}] : [],
 ];
 
+async function listen_for_notifications(notification_stream: ReadableStream<Notification>): Promise<void> {
+  let reader = notification_stream.getReader();
+  do {
+    try {
+      let { done, value } = await reader.read();
+      if (done) {
+        break;
+      }
+
+      // TODO: Do something with the notifications 
+      console.log("Notification", value);
+    } catch (e) {
+      reader.releaseLock();
+      throw e;
+    }
+  } while (true)
+
+  reader.releaseLock();
+}
+
 async function test(factory: TransportFactory, setPhysicalLayout: Dispatch<PhysicalLayout | undefined>, setKeymap: Dispatch<Keymap | undefined>, setBehaviors: Dispatch<BehaviorMap>) {
   let transport = null;
   if (factory.connect) {
@@ -46,6 +67,11 @@ async function test(factory: TransportFactory, setPhysicalLayout: Dispatch<Physi
 
   let rpc_conn = await create_rpc_connection(transport);
 
+  listen_for_notifications(rpc_conn.notification_readable).then(() => {
+    console.log("NO MORE NOTPIFICATIONS");
+    // TODO: Process disconnect here!
+  });
+
   // let req: Request =  { core: { getLockState: true }, requestId: 0 };
   // let resp = await call_rpc(rpc_conn, req);
 

src/tauri/ble.ts

@@ -21,11 +21,17 @@ export async function connect(dev: AvailableDevice): Promise<RpcTransport> {
 
   let { writable: response_writable, readable } = new TransformStream();
 
-  /* const unlisten = */ await listen('connection_data', async (event: { payload: Array<number> }) => {
+  const unlisten_data = await listen('connection_data', async (event: { payload: Array<number> }) => {
     let writer = response_writable.getWriter();
     await writer.write(new Uint8Array(event.payload));
     writer.releaseLock();
-  })
+  });
+
+  const unlisten_disconnected = await listen('connection_disconnected', async (event: any) => {
+    unlisten_data();
+    unlisten_disconnected();
+    readable.cancel();
+  });
 
   return { readable, writable };
 }

src/tauri/serial.ts

@@ -21,12 +21,20 @@ export async function connect(dev: AvailableDevice): Promise<RpcTransport> {
 
   let { writable: response_writable, readable } = new TransformStream();
 
-  /* const unlisten = */ await listen('connection_data', async (event: { payload: Array<number> }) => {
-    console.log(event);
+  console.log(response_writable);
+  console.log(readable);
+
+  const unlisten_data = await listen('connection_data', async (event: { payload: Array<number> }) => {
     let writer = response_writable.getWriter();
     await writer.write(new Uint8Array(event.payload));
     writer.releaseLock();
-  })
+  });
+
+  const unlisten_disconnected = await listen('connection_disconnected', async (event: any) => {
+    unlisten_data();
+    unlisten_disconnected();
+    response_writable.close();
+  });
 
   return { readable, writable };
 }