Changeset View
Changeset View
Standalone View
Standalone View
services/tunnelbroker/src/websockets/session.rs
| Show First 20 Lines • Show All 43 Lines • ▼ Show 20 Lines | |||||
| // Parse a session request and retrieve the device information | // Parse a session request and retrieve the device information | ||||
| pub fn handle_first_message_from_device( | pub fn handle_first_message_from_device( | ||||
| message: &str, | message: &str, | ||||
| ) -> Result<DeviceInfo, SessionError> { | ) -> Result<DeviceInfo, SessionError> { | ||||
| let serialized_message = serde_json::from_str::<Messages>(message)?; | let serialized_message = serde_json::from_str::<Messages>(message)?; | ||||
| match serialized_message { | match serialized_message { | ||||
| Messages::SessionRequest(mut session_info) => { | Messages::ConnectionInitializationMessage(mut session_info) => { | ||||
| let device_info = DeviceInfo { | let device_info = DeviceInfo { | ||||
| device_id: session_info.device_id.clone(), | device_id: session_info.device_id.clone(), | ||||
| notify_token: session_info.notify_token.take(), | notify_token: session_info.notify_token.take(), | ||||
| device_type: session_info.device_type, | device_type: session_info.device_type, | ||||
| device_app_version: session_info.device_app_version.take(), | device_app_version: session_info.device_app_version.take(), | ||||
| device_os: session_info.device_os.take(), | device_os: session_info.device_os.take(), | ||||
| }; | }; | ||||
| ▲ Show 20 Lines • Show All 44 Lines • ▼ Show 20 Lines | Ok(WebsocketSession { | ||||
| tx, | tx, | ||||
| db_client, | db_client, | ||||
| device_info, | device_info, | ||||
| amqp_consumer, | amqp_consumer, | ||||
| }) | }) | ||||
| } | } | ||||
| pub async fn handle_websocket_frame_from_device( | pub async fn handle_websocket_frame_from_device( | ||||
| &mut self, | &self, | ||||
| frame: Message, | msg: Message, | ||||
| ) -> Result<(), SessionError> { | ) -> Result<(), SessionError> { | ||||
| match frame { | debug!("Received frame: {:?}", msg); | ||||
| Message::Text(payload) => { | |||||
| debug!("Received message from device: {}", payload); | |||||
| Ok(()) | |||||
| } | |||||
| Message::Close(_) => { | |||||
| self.close().await; | |||||
| Ok(()) | Ok(()) | ||||
| } | } | ||||
| _ => Err(SessionError::InvalidMessage), | |||||
| } | |||||
| } | |||||
| pub async fn next_amqp_message( | pub async fn next_amqp_message( | ||||
| &mut self, | &mut self, | ||||
| ) -> Option<Result<Delivery, lapin::Error>> { | ) -> Option<Result<Delivery, lapin::Error>> { | ||||
| self.amqp_consumer.next().await | self.amqp_consumer.next().await | ||||
| } | } | ||||
| pub async fn deliver_persisted_messages( | pub async fn deliver_persisted_messages( | ||||
| ▲ Show 20 Lines • Show All 45 Lines • Show Last 20 Lines | |||||