Создает новое соединение с управлением событиями
(
config: Arc<RwLock<ClientConfig>>,
command_tx: mpsc::Sender<Message>,
mut result_rx: mpsc::Receiver<Message>,
timeout: Duration,
check_signal_fn: Option<Check
| 176 | |
| 177 | /// Создает новое соединение с управлением событиями |
| 178 | pub(crate) fn new( |
| 179 | config: Arc<RwLock<ClientConfig>>, |
| 180 | command_tx: mpsc::Sender<Message>, |
| 181 | mut result_rx: mpsc::Receiver<Message>, |
| 182 | timeout: Duration, |
| 183 | check_signal_fn: Option<CheckSignalFn>, |
| 184 | worker_guard: WorkerGuard, |
| 185 | client_handle: Option<tokio::task::JoinHandle<()>>, |
| 186 | ) -> Self { |
| 187 | // Create broadcast channel for event changes with a reasonable buffer |
| 188 | let (event_tx, _) = broadcast::channel(100); |
| 189 | let event_tx_clone = event_tx.clone(); |
| 190 | |
| 191 | // Start the receiver loop in a separate task |
| 192 | let command_tx_clone = command_tx.clone(); |
| 193 | let receiver_handle = tokio::spawn(async move { |
| 194 | while let Some(msg) = result_rx.recv().await { |
| 195 | debug!("Received message: {:?}", msg); |
| 196 | let new_event = match msg { |
| 197 | Message::ConnectState(st) => { |
| 198 | if st == cloudpub_common::protocol::ConnectState::Connected as i32 { |
| 199 | command_tx_clone |
| 200 | .send(Message::EndpointStartAll(EndpointStartAll {})) |
| 201 | .await |
| 202 | .ok(); |
| 203 | ConnectionEvent::Connected |
| 204 | } else if st == cloudpub_common::protocol::ConnectState::Disconnected as i32 |
| 205 | { |
| 206 | ConnectionEvent::Closed |
| 207 | } else { |
| 208 | continue; |
| 209 | } |
| 210 | } |
| 211 | Message::EndpointAck(endpoint) => ConnectionEvent::Endpoint(Box::new(endpoint)), |
| 212 | Message::EndpointListAck(list) => ConnectionEvent::List(list.endpoints), |
| 213 | Message::EndpointStopAck(_) |
| 214 | | Message::EndpointRemoveAck(_) |
| 215 | | Message::EndpointClearAck(_) => ConnectionEvent::Acknowledged, |
| 216 | Message::Error(err) => ConnectionEvent::Error(err.message), |
| 217 | Message::Break(_) => ConnectionEvent::Closed, |
| 218 | _ => continue, // Игнорировать другие сообщения |
| 219 | }; |
| 220 | |
| 221 | // Отправить событие через широковещательный канал |
| 222 | debug!("New event: {:?}", new_event); |
| 223 | // Игнорировать ошибки отправки (нет получателей) |
| 224 | let _ = event_tx_clone.send(new_event); |
| 225 | } |
| 226 | }); |
| 227 | |
| 228 | Connection { |
| 229 | config, |
| 230 | command_tx, |
| 231 | event_tx, |
| 232 | receiver_handle: Some(receiver_handle), |
| 233 | timeout: Arc::new(Mutex::new(timeout)), |
| 234 | check_signal_fn, |
| 235 | _worker_guard: worker_guard, |