src/rust/nci/nci.rs - platform/system/nfc - Git at Google

 // Copyright 2021, The Android Open Source Project
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
 //     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.

 //! NCI Protocol Abstraction Layer
 //! Supports sending NCI commands to the HAL and receiving
 //! NCI messages back

 use bytes::{BufMut, BytesMut};
 use log::{debug, error};
 use nfc_hal::{Hal, HalEventRegistry};
 use nfc_packets::nci::DataPacketChild::Payload;
 use nfc_packets::nci::NciPacketChild;
 use nfc_packets::nci::NotificationChild::ConnCreditsNotification;
 use nfc_packets::nci::{Command, DataPacket, DataPacketBuilder, Notification};
 use nfc_packets::nci::{Opcode, PacketBoundaryFlag, Response};
 use pdl_runtime::Packet;
 use std::collections::HashMap;
 use std::collections::VecDeque;
 use std::sync::{Arc, Mutex};
 use tokio::select;
 use tokio::sync::mpsc::{channel, Receiver, Sender, UnboundedSender};
 use tokio::sync::{oneshot, RwLock};
 use tokio::time::{sleep, Duration, Instant};

 pub mod api;

 /// Result type
 type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;

 /// Initialize the module and connect the channels
 pub async fn init() -> Nci {
     let hc = nfc_hal::init().await;
     // Channel to handle data upstream messages
     //    let (in_data_int, in_data_ext) = channel::<DataPacket>(10);
     // Internal data channels
     //    let ic = InternalChannels { in_data_int };

     let (cmd_tx, cmd_rx) = channel::<QueuedCommand>(10);
     let commands = CommandSender { cmd_tx };
     let hal_events = hc.hal_events.clone();

     let notifications = EventRegistry { handlers: Arc::new(Mutex::new(HashMap::new())) };
     let connections = LogicalConnectionsRegistry {
         conns: Arc::new(RwLock::new(HashMap::new())),
         sender: hc.out_data_tx.clone(),
     };

     tokio::spawn(dispatch(notifications, connections.clone(), hc, cmd_rx));
     Nci { hal_events, commands, connections }
 }

 /// NCI module external interface
 pub struct Nci {
     /// HAL events
     pub hal_events: HalEventRegistry,
     /// NCI command communication interface
     pub commands: CommandSender,
     /// NCI logical connections
     pub connections: LogicalConnectionsRegistry,
 }

 #[derive(Debug)]
 struct PendingCommand {
     cmd: Command,
     response: oneshot::Sender<Response>,
 }

 #[derive(Debug)]
 struct QueuedCommand {
     pending: PendingCommand,
     notification: Option<oneshot::Sender<Notification>>,
 }

 /// Sends raw commands. Only useful for facades & shims, or wrapped as a CommandSender.
 pub struct CommandSender {
     cmd_tx: Sender<QueuedCommand>,
 }

 /// The data returned by send_notify() method.
 pub struct ResponsePendingNotification {
     /// Command response
     pub response: Response,
     /// Pending notification receiver
     pub notification: oneshot::Receiver<Notification>,
 }

 impl CommandSender {
     /// Send a command, but do not expect notification to be returned
     pub async fn send(&mut self, cmd: Command) -> Result<Response> {
         let (tx, rx) = oneshot::channel::<Response>();
         self.cmd_tx
             .send(QueuedCommand {
                 pending: PendingCommand { cmd, response: tx },
                 notification: None,
             })
             .await?;
         let event = rx.await?;
         Ok(event)
     }
     /// Send a command which expects notification as a result
     pub async fn send_and_notify(&mut self, cmd: Command) -> Result<ResponsePendingNotification> {
         let (tx, rx) = oneshot::channel::<Response>();
         let (ntx, nrx) = oneshot::channel::<Notification>();
         self.cmd_tx
             .send(QueuedCommand {
                 pending: PendingCommand { cmd, response: tx },
                 notification: Some(ntx),
             })
             .await?;
         let event = rx.await?;
         Ok(ResponsePendingNotification { response: event, notification: nrx })
     }
 }

 impl Drop for CommandSender {
     fn drop(&mut self) {
         debug!("CommandSender is dropped");
     }
 }

 /// Parameters of a logical connection
 struct ConnectionParameters {
     callback: Option<fn(u8, u16, &[u8])>,
     max_payload_size: u8,
     nfcc_credits_avail: u8,
     sendq: VecDeque<DataPacket>,
     recvq: VecDeque<DataPacket>,
 }

 impl ConnectionParameters {
     /// Flush TX queue
     fn flush_tx(&mut self) {
         self.sendq.clear();
     }
 }

 /// To keep track of currentry open logical connections
 #[derive(Clone)]
 pub struct LogicalConnectionsRegistry {
     conns: Arc<RwLock<HashMap<u8, Mutex<ConnectionParameters>>>>,
     sender: UnboundedSender<DataPacket>,
 }

 impl LogicalConnectionsRegistry {
     /// Create a logical connection
     pub async fn open(
         &mut self,
         conn_id: u8,
         cb: Option<fn(u8, u16, &[u8])>,
         max_payload_size: u8,
         nfcc_credits_avail: u8,
     ) {
         let conn_params = ConnectionParameters {
             callback: cb,
             max_payload_size,
             nfcc_credits_avail,
             sendq: VecDeque::<DataPacket>::new(),
             recvq: VecDeque::<DataPacket>::new(),
         };
         assert!(
             self.conns.write().await.insert(conn_id, Mutex::new(conn_params)).is_none(),
             "A logical connection with id {:?} already exists",
             conn_id
         );
     }
     /// Set static callback
     pub async fn set_static_callback(&mut self, conn_id: u8, cb: Option<fn(u8, u16, &[u8])>) {
         if conn_id < 2 && cb.is_some() {
             // Static connections
             if let Some(conn_params) = self.conns.read().await.get(&conn_id) {
                 let mut conn_params = conn_params.lock().unwrap();
                 conn_params.callback = cb;
             }
         }
     }
     /// Close a logical connection
     pub async fn close(&mut self, conn_id: u8) -> Option<fn(u8, u16, &[u8])> {
         if let Some(conn_params) = self.conns.write().await.remove(&conn_id) {
             conn_params.lock().unwrap().callback
         } else {
             None
         }
     }
     /// Add credits to a logical connection
     pub async fn add_credits(&self, conn_id: u8, ncreds: u8) {
         if let Some(conn_params) = self.conns.read().await.get(&conn_id) {
             let mut conn_params = conn_params.lock().unwrap();
             conn_params.nfcc_credits_avail += ncreds;
             while !conn_params.sendq.is_empty() && conn_params.nfcc_credits_avail > 0 {
                 self.sender.send(conn_params.sendq.pop_front().unwrap()).unwrap();
                 conn_params.nfcc_credits_avail -= 1;
             }
         }
     }

     /// Send a packet to a logical channel, splitting it if needed
     pub async fn send_packet(&mut self, conn_id: u8, pkt: DataPacket) {
         if let Some(conn_params) = self.conns.read().await.get(&conn_id) {
             let mut conn_params = conn_params.lock().unwrap();
             if let Payload(mut p) = pkt.specialize() {
                 if p.len() > conn_params.max_payload_size.into() {
                     let conn_id = pkt.get_conn_id();
                     while p.len() > conn_params.max_payload_size.into() {
                         let part = DataPacketBuilder {
                             conn_id,
                             pbf: PacketBoundaryFlag::Incomplete,
                             cr: 0,
                             payload: Some(p.split_to(conn_params.max_payload_size.into())),
                         }
                         .build();
                         conn_params.sendq.push_back(part);
                     }
                     if !p.is_empty() {
                         let end = DataPacketBuilder {
                             conn_id,
                             pbf: PacketBoundaryFlag::CompleteOrFinal,
                             cr: 0,
                             payload: Some(p),
                         }
                         .build();
                         conn_params.sendq.push_back(end);
                     }
                 } else {
                     conn_params.sendq.push_back(pkt);
                 }
             }
             while conn_params.nfcc_credits_avail > 0 && !conn_params.sendq.is_empty() {
                 self.sender.send(conn_params.sendq.pop_front().unwrap()).unwrap();
                 conn_params.nfcc_credits_avail -= 1;
             }
         }
     }

     /// Send data packet callback to the upper layers
     pub async fn send_callback(&self, pkt: DataPacket) {
         let conn_id = pkt.get_conn_id();
         let ncreds = pkt.get_cr();
         if ncreds > 0 {
             self.add_credits(conn_id, ncreds).await;
         }
         let done = pkt.get_pbf() == PacketBoundaryFlag::CompleteOrFinal;
         if let Some(conn_params) = self.conns.read().await.get(&conn_id) {
             let mut conn_params = conn_params.lock().unwrap();
             if !done && conn_params.recvq.is_empty() {
                 const NFC_DATA_START_CEVT: u16 = 5;
                 let cb = conn_params.callback.unwrap();
                 cb(conn_id, NFC_DATA_START_CEVT, &[]);
             }
             conn_params.recvq.push_back(pkt);
             if done {
                 const NFC_DATA_CEVT_SIZE: usize = 4; // 3 for header and 1 for status
                 let cap = conn_params.recvq.len() * conn_params.max_payload_size as usize
                     + NFC_DATA_CEVT_SIZE;
                 let mut buffer = BytesMut::with_capacity(cap);
                 buffer.put_u8(0u8); // status
                 let pkt = conn_params.recvq.pop_front().unwrap();
                 buffer.put(pkt.encode_to_bytes().unwrap());
                 while !conn_params.recvq.is_empty() {
                     let pkt = conn_params.recvq.pop_front().unwrap();
                     if let Payload(p) = pkt.specialize() {
                         buffer.put(p);
                     }
                 }
                 let data_cevt = buffer.freeze();
                 let cb = conn_params.callback.unwrap();
                 const NFC_DATA_CEVT: u16 = 3;
                 cb(conn_id, NFC_DATA_CEVT, data_cevt.as_ref());
             }
         }
     }

     /// Flush outgoing data queue
     pub async fn flush_data(&mut self, conn_id: u8) -> bool {
         if let Some(conn_params) = self.conns.read().await.get(&conn_id) {
             conn_params.lock().unwrap().flush_tx();
             true
         } else {
             false
         }
     }
 }

 /// Provides ability to register and unregister for NCI notifications
 #[derive(Clone)]
 pub struct EventRegistry {
     handlers: Arc<Mutex<HashMap<Opcode, oneshot::Sender<Notification>>>>,
 }

 impl EventRegistry {
     /// Indicate interest in specific NCI notification
     pub async fn register(&mut self, code: Opcode, sender: oneshot::Sender<Notification>) {
         assert!(
             self.handlers.lock().unwrap().insert(code, sender).is_none(),
             "A handler for {:?} is already registered",
             code
         );
     }

     /// Remove interest in specific NCI notification
     pub async fn unregister(&mut self, code: Opcode) -> Option<oneshot::Sender<Notification>> {
         self.handlers.lock().unwrap().remove(&code)
     }
 }

 async fn dispatch(
     mut ntfs: EventRegistry,
     lcons: LogicalConnectionsRegistry,
     mut hc: Hal,
     //    ic: InternalChannels,
     mut cmd_rx: Receiver<QueuedCommand>,
 ) -> Result<()> {
     let mut pending: Option<PendingCommand> = None;
     let timeout = sleep(Duration::MAX);
     // The max_deadline is used to set  the sleep() deadline to a very distant moment in
     // the future, when the notification from the timer is not required.
     let max_deadline = timeout.deadline();
     tokio::pin!(timeout);
     loop {
         select! {
             Some(cmd) = hc.in_cmd_rx.recv() => {
                 match cmd.specialize() {
                     NciPacketChild::Response(rsp) => {
                         timeout.as_mut().reset(max_deadline);
                         let this_opcode = rsp.get_cmd_op();
                         match pending.take() {
                             Some(PendingCommand{cmd, response}) if cmd.get_op() == this_opcode => {
                                 if let Err(e) = response.send(rsp) {
                                     error!("failure dispatching command status {:?}", e);
                                 }
                             },
                             Some(PendingCommand{cmd, ..}) => panic!("Waiting for {:?}, got {:?}", cmd.get_op(), this_opcode),
                             None => panic!("Unexpected status event with opcode {:?}", this_opcode),
                         }
                     },
                     NciPacketChild::Notification(ntfy) => {
                         match ntfy.specialize() {
                             ConnCreditsNotification(ccnp) => {
                                 let conns = ccnp.get_conns();
                                 for conn in conns {
                                     lcons.add_credits(conn.conn_id, conn.ncredits).await;
                                 }
                             },
                             _ => {
                                 let code = ntfy.get_cmd_op();
                                 match ntfs.unregister(code).await {
                                     Some(sender) => {
                                         if let Err(e) = sender.send(ntfy) {
                                             error!("notification channel closed {:?}", e);
                                         }
                                     },
                                     None => panic!("Unhandled notification {:?}", code),
                                 }
                             },
                         }
                     },
                     _ => error!("Unexpected NCI data received {:?}", cmd),
                 }
             },
             qc = cmd_rx.recv(), if pending.is_none() => if let Some(queued) = qc {
                 debug!("cmd_rx got a q");
                 if let Some(nsender) = queued.notification {
                     ntfs.register(queued.pending.cmd.get_op(), nsender).await;
                 }
                 if let Err(e) = hc.out_cmd_tx.send(queued.pending.cmd.clone().into()) {
                     error!("command queue closed: {:?}", e);
                 }
                 timeout.as_mut().reset(Instant::now() + Duration::from_millis(20));
                 pending = Some(queued.pending);
             } else {
                 break;
             },
             () = &mut timeout => {
                 error!("Command processing timeout");
                 timeout.as_mut().reset(max_deadline);
                 pending = None;
             },
             Some(data) = hc.in_data_rx.recv() => lcons.send_callback(data).await,
             else => {
                 debug!("Select is done");
                 break;
             },
         }
     }
     debug!("NCI dispatch is terminated.");
     Ok(())
 }
	// Copyright 2021, The Android Open Source Project
	//
	// Licensed under the Apache License, Version 2.0 (the "License");
	// you may not use this file except in compliance with the License.
	// You may obtain a copy of the License at
	//
	// http://www.apache.org/licenses/LICENSE-2.0
	//
	// Unless required by applicable law or agreed to in writing, software
	// distributed under the License is distributed on an "AS IS" BASIS,
	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	// See the License for the specific language governing permissions and
	// limitations under the License.

	//! NCI Protocol Abstraction Layer
	//! Supports sending NCI commands to the HAL and receiving
	//! NCI messages back

	use bytes::{BufMut, BytesMut};
	use log::{debug, error};
	use nfc_hal::{Hal, HalEventRegistry};
	use nfc_packets::nci::DataPacketChild::Payload;
	use nfc_packets::nci::NciPacketChild;
	use nfc_packets::nci::NotificationChild::ConnCreditsNotification;
	use nfc_packets::nci::{Command, DataPacket, DataPacketBuilder, Notification};
	use nfc_packets::nci::{Opcode, PacketBoundaryFlag, Response};
	use pdl_runtime::Packet;
	use std::collections::HashMap;
	use std::collections::VecDeque;
	use std::sync::{Arc, Mutex};
	use tokio::select;
	use tokio::sync::mpsc::{channel, Receiver, Sender, UnboundedSender};
	use tokio::sync::{oneshot, RwLock};
	use tokio::time::{sleep, Duration, Instant};

	pub mod api;

	/// Result type
	type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;

	/// Initialize the module and connect the channels
	pub async fn init() -> Nci {
	let hc = nfc_hal::init().await;
	// Channel to handle data upstream messages
	// let (in_data_int, in_data_ext) = channel::<DataPacket>(10);
	// Internal data channels
	// let ic = InternalChannels { in_data_int };

	let (cmd_tx, cmd_rx) = channel::<QueuedCommand>(10);
	let commands = CommandSender { cmd_tx };
	let hal_events = hc.hal_events.clone();

	let notifications = EventRegistry { handlers: Arc::new(Mutex::new(HashMap::new())) };
	let connections = LogicalConnectionsRegistry {
	conns: Arc::new(RwLock::new(HashMap::new())),
	sender: hc.out_data_tx.clone(),
	};

	tokio::spawn(dispatch(notifications, connections.clone(), hc, cmd_rx));
	Nci { hal_events, commands, connections }
	}

	/// NCI module external interface
	pub struct Nci {
	/// HAL events
	pub hal_events: HalEventRegistry,
	/// NCI command communication interface
	pub commands: CommandSender,
	/// NCI logical connections
	pub connections: LogicalConnectionsRegistry,
	}

	#[derive(Debug)]
	struct PendingCommand {
	cmd: Command,
	response: oneshot::Sender<Response>,
	}

	#[derive(Debug)]
	struct QueuedCommand {
	pending: PendingCommand,
	notification: Option<oneshot::Sender<Notification>>,
	}

	/// Sends raw commands. Only useful for facades & shims, or wrapped as a CommandSender.
	pub struct CommandSender {
	cmd_tx: Sender<QueuedCommand>,
	}

	/// The data returned by send_notify() method.
	pub struct ResponsePendingNotification {
	/// Command response
	pub response: Response,
	/// Pending notification receiver
	pub notification: oneshot::Receiver<Notification>,
	}

	impl CommandSender {
	/// Send a command, but do not expect notification to be returned
	pub async fn send(&mut self, cmd: Command) -> Result<Response> {
	let (tx, rx) = oneshot::channel::<Response>();
	self.cmd_tx
	.send(QueuedCommand {
	pending: PendingCommand { cmd, response: tx },
	notification: None,
	})
	.await?;
	let event = rx.await?;
	Ok(event)
	}
	/// Send a command which expects notification as a result
	pub async fn send_and_notify(&mut self, cmd: Command) -> Result<ResponsePendingNotification> {
	let (tx, rx) = oneshot::channel::<Response>();
	let (ntx, nrx) = oneshot::channel::<Notification>();
	self.cmd_tx
	.send(QueuedCommand {
	pending: PendingCommand { cmd, response: tx },
	notification: Some(ntx),
	})
	.await?;
	let event = rx.await?;
	Ok(ResponsePendingNotification { response: event, notification: nrx })
	}
	}

	impl Drop for CommandSender {
	fn drop(&mut self) {
	debug!("CommandSender is dropped");
	}
	}

	/// Parameters of a logical connection
	struct ConnectionParameters {
	callback: Option<fn(u8, u16, &[u8])>,
	max_payload_size: u8,
	nfcc_credits_avail: u8,
	sendq: VecDeque<DataPacket>,
	recvq: VecDeque<DataPacket>,
	}

	impl ConnectionParameters {
	/// Flush TX queue
	fn flush_tx(&mut self) {
	self.sendq.clear();
	}
	}

	/// To keep track of currentry open logical connections
	#[derive(Clone)]
	pub struct LogicalConnectionsRegistry {
	conns: Arc<RwLock<HashMap<u8, Mutex<ConnectionParameters>>>>,
	sender: UnboundedSender<DataPacket>,
	}

	impl LogicalConnectionsRegistry {
	/// Create a logical connection
	pub async fn open(
	&mut self,
	conn_id: u8,
	cb: Option<fn(u8, u16, &[u8])>,
	max_payload_size: u8,
	nfcc_credits_avail: u8,
	) {
	let conn_params = ConnectionParameters {
	callback: cb,
	max_payload_size,
	nfcc_credits_avail,
	sendq: VecDeque::<DataPacket>::new(),
	recvq: VecDeque::<DataPacket>::new(),
	};
	assert!(
	self.conns.write().await.insert(conn_id, Mutex::new(conn_params)).is_none(),
	"A logical connection with id {:?} already exists",
	conn_id
	);
	}
	/// Set static callback
	pub async fn set_static_callback(&mut self, conn_id: u8, cb: Option<fn(u8, u16, &[u8])>) {
	if conn_id < 2 && cb.is_some() {
	// Static connections
	if let Some(conn_params) = self.conns.read().await.get(&conn_id) {
	let mut conn_params = conn_params.lock().unwrap();
	conn_params.callback = cb;
	}
	}
	}
	/// Close a logical connection
	pub async fn close(&mut self, conn_id: u8) -> Option<fn(u8, u16, &[u8])> {
	if let Some(conn_params) = self.conns.write().await.remove(&conn_id) {
	conn_params.lock().unwrap().callback
	} else {
	None
	}
	}
	/// Add credits to a logical connection
	pub async fn add_credits(&self, conn_id: u8, ncreds: u8) {
	if let Some(conn_params) = self.conns.read().await.get(&conn_id) {
	let mut conn_params = conn_params.lock().unwrap();
	conn_params.nfcc_credits_avail += ncreds;
	while !conn_params.sendq.is_empty() && conn_params.nfcc_credits_avail > 0 {
	self.sender.send(conn_params.sendq.pop_front().unwrap()).unwrap();
	conn_params.nfcc_credits_avail -= 1;
	}
	}
	}

	/// Send a packet to a logical channel, splitting it if needed
	pub async fn send_packet(&mut self, conn_id: u8, pkt: DataPacket) {
	if let Some(conn_params) = self.conns.read().await.get(&conn_id) {
	let mut conn_params = conn_params.lock().unwrap();
	if let Payload(mut p) = pkt.specialize() {
	if p.len() > conn_params.max_payload_size.into() {
	let conn_id = pkt.get_conn_id();
	while p.len() > conn_params.max_payload_size.into() {
	let part = DataPacketBuilder {
	conn_id,
	pbf: PacketBoundaryFlag::Incomplete,
	cr: 0,
	payload: Some(p.split_to(conn_params.max_payload_size.into())),
	}
	.build();
	conn_params.sendq.push_back(part);
	}
	if !p.is_empty() {
	let end = DataPacketBuilder {
	conn_id,
	pbf: PacketBoundaryFlag::CompleteOrFinal,
	cr: 0,
	payload: Some(p),
	}
	.build();
	conn_params.sendq.push_back(end);
	}
	} else {
	conn_params.sendq.push_back(pkt);
	}
	}
	while conn_params.nfcc_credits_avail > 0 && !conn_params.sendq.is_empty() {
	self.sender.send(conn_params.sendq.pop_front().unwrap()).unwrap();
	conn_params.nfcc_credits_avail -= 1;
	}
	}
	}

	/// Send data packet callback to the upper layers
	pub async fn send_callback(&self, pkt: DataPacket) {
	let conn_id = pkt.get_conn_id();
	let ncreds = pkt.get_cr();
	if ncreds > 0 {
	self.add_credits(conn_id, ncreds).await;
	}
	let done = pkt.get_pbf() == PacketBoundaryFlag::CompleteOrFinal;
	if let Some(conn_params) = self.conns.read().await.get(&conn_id) {
	let mut conn_params = conn_params.lock().unwrap();
	if !done && conn_params.recvq.is_empty() {
	const NFC_DATA_START_CEVT: u16 = 5;
	let cb = conn_params.callback.unwrap();
	cb(conn_id, NFC_DATA_START_CEVT, &[]);
	}
	conn_params.recvq.push_back(pkt);
	if done {
	const NFC_DATA_CEVT_SIZE: usize = 4; // 3 for header and 1 for status
	let cap = conn_params.recvq.len() * conn_params.max_payload_size as usize
	+ NFC_DATA_CEVT_SIZE;
	let mut buffer = BytesMut::with_capacity(cap);
	buffer.put_u8(0u8); // status
	let pkt = conn_params.recvq.pop_front().unwrap();
	buffer.put(pkt.encode_to_bytes().unwrap());
	while !conn_params.recvq.is_empty() {
	let pkt = conn_params.recvq.pop_front().unwrap();
	if let Payload(p) = pkt.specialize() {
	buffer.put(p);
	}
	}
	let data_cevt = buffer.freeze();
	let cb = conn_params.callback.unwrap();
	const NFC_DATA_CEVT: u16 = 3;
	cb(conn_id, NFC_DATA_CEVT, data_cevt.as_ref());
	}
	}
	}

	/// Flush outgoing data queue
	pub async fn flush_data(&mut self, conn_id: u8) -> bool {
	if let Some(conn_params) = self.conns.read().await.get(&conn_id) {
	conn_params.lock().unwrap().flush_tx();
	true
	} else {
	false
	}
	}
	}

	/// Provides ability to register and unregister for NCI notifications
	#[derive(Clone)]
	pub struct EventRegistry {
	handlers: Arc<Mutex<HashMap<Opcode, oneshot::Sender<Notification>>>>,
	}

	impl EventRegistry {
	/// Indicate interest in specific NCI notification
	pub async fn register(&mut self, code: Opcode, sender: oneshot::Sender<Notification>) {
	assert!(
	self.handlers.lock().unwrap().insert(code, sender).is_none(),
	"A handler for {:?} is already registered",
	code
	);
	}

	/// Remove interest in specific NCI notification
	pub async fn unregister(&mut self, code: Opcode) -> Option<oneshot::Sender<Notification>> {
	self.handlers.lock().unwrap().remove(&code)
	}
	}

	async fn dispatch(
	mut ntfs: EventRegistry,
	lcons: LogicalConnectionsRegistry,
	mut hc: Hal,
	// ic: InternalChannels,
	mut cmd_rx: Receiver<QueuedCommand>,
	) -> Result<()> {
	let mut pending: Option<PendingCommand> = None;
	let timeout = sleep(Duration::MAX);
	// The max_deadline is used to set the sleep() deadline to a very distant moment in
	// the future, when the notification from the timer is not required.
	let max_deadline = timeout.deadline();
	tokio::pin!(timeout);
	loop {
	select! {
	Some(cmd) = hc.in_cmd_rx.recv() => {
	match cmd.specialize() {
	NciPacketChild::Response(rsp) => {
	timeout.as_mut().reset(max_deadline);
	let this_opcode = rsp.get_cmd_op();
	match pending.take() {
	Some(PendingCommand{cmd, response}) if cmd.get_op() == this_opcode => {
	if let Err(e) = response.send(rsp) {
	error!("failure dispatching command status {:?}", e);
	}
	},
	Some(PendingCommand{cmd, ..}) => panic!("Waiting for {:?}, got {:?}", cmd.get_op(), this_opcode),
	None => panic!("Unexpected status event with opcode {:?}", this_opcode),
	}
	},
	NciPacketChild::Notification(ntfy) => {
	match ntfy.specialize() {
	ConnCreditsNotification(ccnp) => {
	let conns = ccnp.get_conns();
	for conn in conns {
	lcons.add_credits(conn.conn_id, conn.ncredits).await;
	}
	},
	_ => {
	let code = ntfy.get_cmd_op();
	match ntfs.unregister(code).await {
	Some(sender) => {
	if let Err(e) = sender.send(ntfy) {
	error!("notification channel closed {:?}", e);
	}
	},
	None => panic!("Unhandled notification {:?}", code),
	}
	},
	}
	},
	_ => error!("Unexpected NCI data received {:?}", cmd),
	}
	},
	qc = cmd_rx.recv(), if pending.is_none() => if let Some(queued) = qc {
	debug!("cmd_rx got a q");
	if let Some(nsender) = queued.notification {
	ntfs.register(queued.pending.cmd.get_op(), nsender).await;
	}
	if let Err(e) = hc.out_cmd_tx.send(queued.pending.cmd.clone().into()) {
	error!("command queue closed: {:?}", e);
	}
	timeout.as_mut().reset(Instant::now() + Duration::from_millis(20));
	pending = Some(queued.pending);
	} else {
	break;
	},
	() = &mut timeout => {
	error!("Command processing timeout");
	timeout.as_mut().reset(max_deadline);
	pending = None;
	},
	Some(data) = hc.in_data_rx.recv() => lcons.send_callback(data).await,
	else => {
	debug!("Select is done");
	break;
	},
	}
	}
	debug!("NCI dispatch is terminated.");
	Ok(())
	}