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;
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);
commands,
events,
acl,
- rt,
evt_tx,
evt_rx: Arc::new(Mutex::new(evt_rx)),
le_evt_tx,
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>,
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;
}
}
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());
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());
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());