Something went wrong. Try again.
Buttplug sex toy control library
Something went wrong. Try again.
13 kB · 350 lines
Rust
at dev
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351// Buttplug Rust Source Code File - See https://buttplug.io for more info.//// Copyright 2016-2024 Nonpolynomial Labs LLC. All rights reserved.//// Licensed under the BSD 3-Clause license. See LICENSE file in the project root// for full license information.
//! Implementation of internal Buttplug Client event loop.
use super::{ ButtplugClientEvent, ButtplugClientMessageFuturePair, ButtplugClientMessageSender, client_message_sorter::ClientMessageSorter, device::{ButtplugClientDevice, ButtplugClientDeviceEvent},};use buttplug_core::{ connector::{ButtplugConnector, ButtplugConnectorStateShared}, errors::ButtplugError, message::{ ButtplugClientMessageV4, ButtplugDeviceMessage, ButtplugMessageValidator, ButtplugServerMessageV4, DeviceListV4, DeviceMessageInfoV4, },};use dashmap::DashMap;use log::*;use std::sync::{ Arc, atomic::{AtomicBool, Ordering},};use tokio::{ select, sync::{broadcast, mpsc},};
/// Enum used for communication from the client to the event loop.#[derive(Clone)]pub enum ButtplugClientRequest { /// Client request to disconnect, via already sent connector instance. Disconnect(ButtplugConnectorStateShared), /// Given a DeviceList message, update the inner loop values and create /// events for additions. HandleDeviceList(DeviceListV4), /// Client request to send a message via the connector. /// /// Bundled future should have reply set and waker called when this is /// finished. Message(ButtplugClientMessageFuturePair),}
/// Event loop for running [ButtplugClient] connections.////// Acts as a hub for communication between the connector and [ButtplugClient]/// instances.////// Created whenever a new [super::ButtplugClient] is created, the internal loop/// handles connection and communication with the server through the connector,/// and creation of events received from the server.////// The event_loop does a few different things during its lifetime:////// - It will listen for events from the connector, or messages from the client,/// routing them to their proper receivers until either server/client/// disconnects.////// - On disconnect, it will tear down, and cannot be used again. All clients/// and devices associated with the loop will be invalidated, and connect must/// be called on the client again (or a new client should be created).////// # Why an event loop?////// Due to the async nature of Buttplug, we many channels routed to many/// different tasks. However, all of those tasks will refer to the same event/// loop. This allows us to coordinate and centralize our information while/// keeping the API async.////// Note that no async call here should block. Any .await should only be on/// async channels, and those channels should never have backpressure. We hope.pub(super) struct ButtplugClientEventLoop<ConnectorType>where ConnectorType: ButtplugConnector<ButtplugClientMessageV4, ButtplugServerMessageV4> + 'static,{ /// Connected status from client, managed by the event loop in case of disconnect. connected_status: Arc<AtomicBool>, /// Connector the event loop will use to communicate with the [ButtplugServer] connector: ConnectorType, /// Receiver for messages send from the [ButtplugServer] via the connector. from_connector_receiver: mpsc::Receiver<ButtplugServerMessageV4>, /// Map of devices shared between the client and the event loop device_map: Arc<DashMap<u32, ButtplugClientDevice>>, /// Sends events to the [ButtplugClient] instance. to_client_sender: broadcast::Sender<ButtplugClientEvent>, /// Sends events to the client receiver. Stored here so it can be handed to /// new ButtplugClientDevice instances. from_client_sender: ButtplugClientMessageSender, /// Receives incoming messages from client instances. from_client_receiver: broadcast::Receiver<ButtplugClientRequest>, sorter: ClientMessageSorter,}
impl<ConnectorType> ButtplugClientEventLoop<ConnectorType>where ConnectorType: ButtplugConnector<ButtplugClientMessageV4, ButtplugServerMessageV4> + 'static,{ /// Creates a new [ButtplugClientEventLoop]. /// /// Given the [ButtplugClientConnector] object, as well as the channels used /// for communicating with the client, creates an event loop structure and /// returns it. pub fn new( connected_status: Arc<AtomicBool>, connector: ConnectorType, from_connector_receiver: mpsc::Receiver<ButtplugServerMessageV4>, to_client_sender: broadcast::Sender<ButtplugClientEvent>, from_client_sender: ButtplugClientMessageSender, device_map: Arc<DashMap<u32, ButtplugClientDevice>>, ) -> Self { trace!("Creating ButtplugClientEventLoop instance."); Self { connected_status, device_map, from_client_receiver: from_client_sender.subscribe(), from_client_sender, to_client_sender, from_connector_receiver, connector, sorter: ClientMessageSorter::default(), } }
/// Creates a [ButtplugClientDevice] from [DeviceMessageInfo]. /// /// Given a [DeviceMessageInfo] from a [DeviceAdded] or [DeviceList] message, /// creates a ButtplugClientDevice and adds it the internal device map, then /// returns the instance. fn create_client_device(&mut self, info: &DeviceMessageInfoV4) -> ButtplugClientDevice { debug!( "Trying to create a client device from DeviceMessageInfo: {:?}", info ); match self.device_map.get(&info.device_index()) { // If the device already exists in our map, clone it. Some(dev) => { debug!("Device already exists, creating clone."); dev.clone() } // If it doesn't, insert it. None => { debug!("Device does not exist, creating new entry."); let device = ButtplugClientDevice::new_from_device_info(info, &self.from_client_sender); self.device_map.insert(info.device_index(), device.clone()); device } } }
fn send_client_event(&mut self, event: ButtplugClientEvent) { trace!("Forwarding event {:?} to client", event);
if self.to_client_sender.receiver_count() == 0 { error!( "Client event {:?} dropped, no client event listener available.", event ); return; }
self .to_client_sender .send(event) .expect("Already checked for receivers."); }
fn disconnect_device(&mut self, device_index: u32) { if !self.device_map.contains_key(&device_index) { return; }
let device = (*self .device_map .get(&device_index) .expect("Checked for device index already.")) .clone(); device.set_device_connected(false); device.queue_event(ButtplugClientDeviceEvent::DeviceRemoved); // Then remove it from our storage map self.device_map.remove(&device_index); self.send_client_event(ButtplugClientEvent::DeviceRemoved(device)); }
/// Parse device messages from the connector. /// /// Since the event loop maintains the state of all devices reported from the /// server, it will catch [DeviceAdded]/[DeviceList]/[DeviceRemoved] messages /// and update its map accordingly. After that, it will pass the information /// on as a [ButtplugClientEvent] to the [ButtplugClient]. async fn parse_connector_message(&mut self, msg: ButtplugServerMessageV4) { if self.sorter.maybe_resolve_result(&msg) { trace!("Message future found, returning"); return; } if let Err(e) = msg.is_valid() { error!("Message not valid: {:?} - Error: {}", msg, e); self.send_client_event(ButtplugClientEvent::Error(ButtplugError::from(e))); return; } trace!("Message future not found, assuming server event."); info!("{:?}", msg); match msg { ButtplugServerMessageV4::DeviceList(list) => { trace!("Got device list, devices either added or removed"); for dev in list.devices() { if self.device_map.contains_key(&dev.1.device_index()) { continue; } trace!("Device added, updating map and sending to client"); let info = dev.1.clone(); let device = self.create_client_device(&info); self.send_client_event(ButtplugClientEvent::DeviceAdded(device)); } let new_indexes: Vec<u32> = list.devices().iter().map(|x| x.1.device_index()).collect(); let disconnected_indexes: Vec<u32> = self .device_map .iter() .filter(|x| !new_indexes.contains(x.key())) .map(|x| *x.key()) .collect(); for index in disconnected_indexes { trace!("Device removed, updating map and sending to client"); self.disconnect_device(index); } } ButtplugServerMessageV4::ScanningFinished(_) => { trace!("Scanning finished event received, forwarding to client."); self.send_client_event(ButtplugClientEvent::ScanningFinished); } ButtplugServerMessageV4::InputReading(msg) => { let device_idx = msg.device_index(); if let Some(device) = self.device_map.get(&device_idx) { device .value() .queue_event(ButtplugClientDeviceEvent::Message( ButtplugServerMessageV4::from(msg), )); } } ButtplugServerMessageV4::Error(e) => { self.send_client_event(ButtplugClientEvent::Error(e.into())); } _ => error!("Cannot process message, dropping: {:?}", msg), } }
/// Send a message from the [ButtplugClient] to the [ButtplugClientConnector]. async fn send_message(&mut self, mut msg_fut: ButtplugClientMessageFuturePair) { if let Err(e) = &msg_fut.msg.is_valid() { error!("Message not valid: {:?} - Error: {}", msg_fut.msg, e); msg_fut .waker .set_reply(Err(ButtplugError::from(e.clone()).into())); return; }
trace!("Sending message to connector: {:?}", msg_fut.msg); self.sorter.register_future(&mut msg_fut); if self.connector.send(msg_fut.msg).await.is_err() { error!("Sending message failed, connector most likely no longer connected."); } }
/// Parses message types from the client, returning false when disconnect /// happens. /// /// Takes different messages from the client and handles them: /// /// - For outbound messages to the server, sends them to the connector/server. /// - For disconnections, requests connector disconnect /// - For RequestDeviceList, builds a reply out of its own async fn parse_client_request(&mut self, msg: ButtplugClientRequest) -> bool { match msg { ButtplugClientRequest::Message(msg_fut) => { trace!("Sending message through connector: {:?}", msg_fut.msg); self.send_message(msg_fut).await; true } ButtplugClientRequest::Disconnect(state) => { trace!("Client requested disconnect"); state.set_reply(self.connector.disconnect().await); false } ButtplugClientRequest::HandleDeviceList(device_list) => { trace!("Device list received, updating map."); for (i, device) in device_list.devices() { if self.device_map.contains_key(i) { continue; } let device = self.create_client_device(device); self.send_client_event(ButtplugClientEvent::DeviceAdded(device)); } true } } }
/// Runs the event loop, returning once either the client or connector drops. pub async fn run(&mut self) { debug!("Running client event loop."); loop { select! { event = self.from_connector_receiver.recv() => match event { None => { info!("Connector disconnected, exiting loop."); break; } Some(msg) => { self.parse_connector_message(msg).await; } }, client = self.from_client_receiver.recv() => match client { Err(_) => { info!("Client disconnected, exiting loop."); break; } Ok(msg) => { if !self.parse_client_request(msg).await { break; } } }, }; } self .device_map .iter() .for_each(|val| val.value().set_client_connected(false));
let device_indexes: Vec<u32> = self.device_map.iter().map(|k| *k.key()).collect(); device_indexes .iter() .for_each(|k| self.disconnect_device(*k)); self.connected_status.store(false, Ordering::Relaxed); self.send_client_event(ButtplugClientEvent::ServerDisconnect);
debug!("Exiting client event loop."); }}