From ac3a7442dab14fc6059f30f54babe9c4b561f81a Mon Sep 17 00:00:00 2001
From: Zach Johnson <zachoverflow@google.com>
Date: Tue, 22 Dec 2020 15:24:40 -0800
Subject: [PATCH] rusty-gd: switch hci facade to grpc context

this is backed by the tokio runtime, but
we don't need to pass it around

Bug: 171749953
Tag: #gd-refactor
Test: gd/cert/run --rhost SimpleHalTest
Change-Id: I606f4aff6c0ad28b9a453a68d41f8b1815196bd2
---
 gd/rust/hci/src/facade.rs | 59 +++++++++++++++++++++++++----------------------
 1 file changed, 31 insertions(+), 28 deletions(-)

diff --git a/gd/rust/hci/src/facade.rs b/gd/rust/hci/src/facade.rs
index 31e3fa87d..c07865731 100644
--- a/gd/rust/hci/src/facade.rs
+++ b/gd/rust/hci/src/facade.rs
@@ -15,7 +15,6 @@ use gddi::{module, provides, Stoppable};
 use grpcio::*;
 use num_traits::FromPrimitive;
 use std::sync::Arc;
-use tokio::runtime::Runtime;
 use tokio::sync::mpsc::{channel, Receiver, Sender};
 use tokio::sync::Mutex;
 
@@ -31,7 +30,6 @@ async fn provide_facade(
     commands: RawCommandSender,
     events: EventRegistry,
     acl: AclHal,
-    rt: Arc<Runtime>,
 ) -> HciFacadeService {
     let (evt_tx, evt_rx) = channel::<EventPacket>(10);
     let (le_evt_tx, le_evt_rx) = channel::<LeMetaEventPacket>(10);
@@ -39,7 +37,6 @@ async fn provide_facade(
         commands,
         events,
         acl,
-        rt,
         evt_tx,
         evt_rx: Arc::new(Mutex::new(evt_rx)),
         le_evt_tx,
@@ -54,7 +51,6 @@ pub struct HciFacadeService {
     pub commands: RawCommandSender,
     events: EventRegistry,
     pub acl: AclHal,
-    rt: Arc<Runtime>,
     evt_tx: Sender<EventPacket>,
     pub evt_rx: Arc<Mutex<Receiver<EventPacket>>>,
     le_evt_tx: Sender<LeMetaEventPacket>,
@@ -63,14 +59,13 @@ pub struct HciFacadeService {
 
 impl HciFacadeService {
     /// Register for the event & plug in the channel to get them back on
-    pub async fn register_event(&self, code: u32) {
-        self.events.clone().register(EventCode::from_u32(code).unwrap(), self.evt_tx.clone()).await;
+    pub async fn register_event(&mut self, code: u32) {
+        self.events.register(EventCode::from_u32(code).unwrap(), self.evt_tx.clone()).await;
     }
 
     /// Register for the le event & plug in the channel to get them back on
-    pub async fn register_le_event(&self, code: u32) {
+    pub async fn register_le_event(&mut self, code: u32) {
         self.events
-            .clone()
             .register_le(SubeventCode::from_u32(code).unwrap(), self.le_evt_tx.clone())
             .await;
     }
@@ -83,45 +78,53 @@ impl GrpcFacade for HciFacadeService {
 }
 
 impl HciFacade for HciFacadeService {
-    fn send_command(&mut self, _ctx: RpcContext<'_>, mut data: Data, sink: UnarySink<Empty>) {
-        self.rt
-            .block_on(self.commands.send(CommandPacket::parse(&data.take_payload()).unwrap()))
-            .unwrap();
-        sink.success(Empty::default());
+    fn send_command(&mut self, ctx: RpcContext<'_>, mut data: Data, sink: UnarySink<Empty>) {
+        let packet = CommandPacket::parse(&data.take_payload()).unwrap();
+        let mut commands = self.commands.clone();
+        ctx.spawn(async move {
+            commands.send(packet).await.unwrap();
+            sink.success(Empty::default()).await.unwrap();
+        });
     }
 
-    fn request_event(&mut self, _ctx: RpcContext<'_>, req: EventRequest, sink: UnarySink<Empty>) {
-        self.rt.block_on(self.register_event(req.get_code()));
-        sink.success(Empty::default());
+    fn request_event(&mut self, ctx: RpcContext<'_>, req: EventRequest, sink: UnarySink<Empty>) {
+        let mut clone = self.clone();
+        ctx.spawn(async move {
+            clone.register_event(req.get_code()).await;
+            sink.success(Empty::default()).await.unwrap();
+        });
     }
 
     fn request_le_subevent(
         &mut self,
-        _ctx: RpcContext<'_>,
+        ctx: RpcContext<'_>,
         req: EventRequest,
         sink: UnarySink<Empty>,
     ) {
-        self.rt.block_on(self.register_le_event(req.get_code()));
-        sink.success(Empty::default());
+        let mut clone = self.clone();
+        ctx.spawn(async move {
+            clone.register_le_event(req.get_code()).await;
+            sink.success(Empty::default()).await.unwrap();
+        });
     }
 
-    fn send_acl(&mut self, _ctx: RpcContext<'_>, mut packet: Data, sink: UnarySink<Empty>) {
+    fn send_acl(&mut self, ctx: RpcContext<'_>, mut packet: Data, sink: UnarySink<Empty>) {
         let acl_tx = self.acl.tx.clone();
-        self.rt.block_on(async move {
+        ctx.spawn(async move {
             acl_tx.send(AclPacket::parse(&packet.take_payload()).unwrap()).await.unwrap();
+            sink.success(Empty::default()).await.unwrap();
         });
-        sink.success(Empty::default());
     }
 
     fn stream_events(
         &mut self,
-        _ctx: RpcContext<'_>,
+        ctx: RpcContext<'_>,
         _req: Empty,
         mut resp: ServerStreamingSink<Data>,
     ) {
         let evt_rx = self.evt_rx.clone();
 
-        self.rt.spawn(async move {
+        ctx.spawn(async move {
             while let Some(event) = evt_rx.lock().await.recv().await {
                 let mut evt = Data::default();
                 evt.set_payload(event.to_vec());
@@ -132,13 +135,13 @@ impl HciFacade for HciFacadeService {
 
     fn stream_le_subevents(
         &mut self,
-        _ctx: RpcContext<'_>,
+        ctx: RpcContext<'_>,
         _req: Empty,
         mut resp: ServerStreamingSink<Data>,
     ) {
         let evt_rx = self.le_evt_rx.clone();
 
-        self.rt.spawn(async move {
+        ctx.spawn(async move {
             while let Some(event) = evt_rx.lock().await.recv().await {
                 let mut evt = Data::default();
                 evt.set_payload(event.to_vec());
@@ -149,13 +152,13 @@ impl HciFacade for HciFacadeService {
 
     fn stream_acl(
         &mut self,
-        _ctx: RpcContext<'_>,
+        ctx: RpcContext<'_>,
         _req: Empty,
         mut resp: ServerStreamingSink<Data>,
     ) {
         let acl_rx = self.acl.rx.clone();
 
-        self.rt.spawn(async move {
+        ctx.spawn(async move {
             while let Some(data) = acl_rx.lock().await.recv().await {
                 let mut packet = Data::default();
                 packet.set_payload(data.to_vec());
-- 
2.11.0