Retreive an async stream of ButtplugServerMessages. This is how the server sends out non-query-related updates to the system, including information on devices being added/removed, client disconnection, etc...
(&self)
| 146 | /// non-query-related updates to the system, including information on devices being added/removed, |
| 147 | /// client disconnection, etc... |
| 148 | pub fn event_stream(&self) -> impl Stream<Item = ButtplugServerMessageVariant> + use<> { |
| 149 | let state = self.state.clone(); |
| 150 | let converter = ButtplugServerMessageConverter::new(None); |
| 151 | let device_indexes: Vec<u32> = self |
| 152 | .device_manager |
| 153 | .devices() |
| 154 | .iter() |
| 155 | .map(|x| *x.key()) |
| 156 | .collect(); |
| 157 | let device_event_converter = ButtplugServerDeviceEventMessageConverter::new(device_indexes); |
| 158 | self.server_version_event_stream().filter_map(move |m| { |
| 159 | // Get spec_version from Connected state, default to Version4 if not connected |
| 160 | let spec_version = { |
| 161 | let state_guard = state.read().expect("State lock poisoned"); |
| 162 | match &*state_guard { |
| 163 | ConnectionState::Connected { spec_version, .. } => *spec_version, |
| 164 | _ => ButtplugMessageSpecVersion::Version4, |
| 165 | } |
| 166 | }; |
| 167 | if let ButtplugServerMessageV4::DeviceList(list) = m { |
| 168 | device_event_converter.convert_device_list(&spec_version, &list) |
| 169 | } else { |
| 170 | // If we get an event and don't have a spec version yet, just throw out the latest. |
| 171 | Some(converter.convert_outgoing(&m, &spec_version).unwrap()) |
| 172 | } |
| 173 | }) |
| 174 | } |
| 175 | |
| 176 | /// Retreive an async stream of ButtplugServerMessages, always at the latest available message |
| 177 | /// spec. This is how the server sends out non-query-related updates to the system, including |
no test coverage detected