use crate::{AclHal, ControlHal};
use bt_common::GrpcFacade;
+use bt_facade_helpers::RxAdapter;
use bt_facade_proto::common::Data;
use bt_facade_proto::empty::Empty;
use bt_facade_proto::hal_facade_grpc::{create_hci_hal_facade, HciHalFacade};
-use bt_packets::hci::{AclPacket, CommandPacket, Packet};
-use futures::sink::SinkExt;
+use bt_packets::hci::{AclPacket, CommandPacket, EventPacket};
use gddi::{module, provides, Stoppable};
use grpcio::*;
-use std::sync::Arc;
-use tokio::runtime::Runtime;
module! {
hal_facade_module,
}
#[provides]
-async fn provide_facade(control: ControlHal, acl: AclHal, rt: Arc<Runtime>) -> HciHalFacadeService {
- HciHalFacadeService { rt, control, acl }
+async fn provide_facade(control: ControlHal, acl: AclHal) -> HciHalFacadeService {
+ HciHalFacadeService {
+ evt_rx: RxAdapter::from_arc(control.rx.clone()),
+ acl_rx: RxAdapter::from_arc(acl.rx.clone()),
+ control,
+ acl,
+ }
}
/// HCI HAL facade service
#[derive(Clone, Stoppable)]
pub struct HciHalFacadeService {
- rt: Arc<Runtime>,
+ evt_rx: RxAdapter<EventPacket>,
+ acl_rx: RxAdapter<AclPacket>,
control: ControlHal,
acl: AclHal,
}
}
impl HciHalFacade for HciHalFacadeService {
- fn send_command(&mut self, _ctx: RpcContext<'_>, mut data: Data, sink: UnarySink<Empty>) {
+ fn send_command(&mut self, ctx: RpcContext<'_>, mut data: Data, sink: UnarySink<Empty>) {
let cmd_tx = self.control.tx.clone();
- self.rt.block_on(async move {
+ ctx.spawn(async move {
cmd_tx.send(CommandPacket::parse(&data.take_payload()).unwrap()).await.unwrap();
+ sink.success(Empty::default()).await.unwrap();
});
- sink.success(Empty::default());
}
- fn send_acl(&mut self, _ctx: RpcContext<'_>, mut data: Data, sink: UnarySink<Empty>) {
+ fn send_acl(&mut self, ctx: RpcContext<'_>, mut data: 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(&data.take_payload()).unwrap()).await.unwrap();
+ sink.success(Empty::default()).await.unwrap();
});
- sink.success(Empty::default());
}
fn send_sco(&mut self, _ctx: RpcContext<'_>, _sco: Data, _sink: UnarySink<Empty>) {
unimplemented!()
}
- fn stream_events(
- &mut self,
- _ctx: RpcContext<'_>,
- _: Empty,
- mut sink: ServerStreamingSink<Data>,
- ) {
- let evt_rx = self.control.rx.clone();
- self.rt.spawn(async move {
- while let Some(event) = evt_rx.lock().await.recv().await {
- let mut output = Data::default();
- output.set_payload(event.to_vec());
- sink.send((output, WriteFlags::default())).await.unwrap();
- }
- });
+ fn stream_events(&mut self, ctx: RpcContext<'_>, _: Empty, sink: ServerStreamingSink<Data>) {
+ self.evt_rx.stream_grpc(ctx, sink);
}
- fn stream_acl(&mut self, _ctx: RpcContext<'_>, _: Empty, mut sink: ServerStreamingSink<Data>) {
- let acl_rx = self.acl.rx.clone();
- self.rt.spawn(async move {
- while let Some(acl) = acl_rx.lock().await.recv().await {
- let mut output = Data::default();
- output.set_payload(acl.to_vec());
- sink.send((output, WriteFlags::default())).await.unwrap();
- }
- });
+ fn stream_acl(&mut self, ctx: RpcContext<'_>, _: Empty, sink: ServerStreamingSink<Data>) {
+ self.acl_rx.stream_grpc(ctx, sink);
}
fn stream_sco(&mut self, _ctx: RpcContext<'_>, _: Empty, _sink: ServerStreamingSink<Data>) {