feat: rpc with error

Cargo.toml

@@ -37,3 +37,4 @@ indexmap = "2.2"
 spin = "0.9"
 httpmock = "0.7"
 test-log = "0.2"
+anyhow = "1.0"

bin/Cargo.toml

@@ -40,6 +40,7 @@ hex = { version = "0.4", optional = true }
 mime_guess = { version = "2.0", optional = true }
 reqwest = { version = "0.12", features = ["json"]}
 sentry = "0.34"
+anyhow = { workspace = true }
 
 [features]
 default = ["console", "gateway", "media", "connector", "standalone", "cert_utils"]

bin/src/http/api_console/connector.rs

@@ -77,7 +77,7 @@
     #[oai(path = "/:node/log/rooms", method = "get")]
     async fn rooms(&self, _auth: ConsoleAuthorization, Data(ctx): Data<&ConsoleApisCtx>, Path(node): Path<u32>, Query(page): Query<u32>, Query(limit): Query<u32>) -> Json<Response<Vec<RoomInfo>>> {
         match ctx.connector.rooms(node_vnet_addr(node, CONNECTOR_RPC_PORT), GetParams { page, limit }).await {
-            Some(res) => Json(Response {
+            Ok(res) => Json(Response {
                 status: true,
                 data: Some(
                     res.rooms
@@ -98,9 +98,9 @@
                 }),
                 ..Default::default()
             }),
-            None => Json(Response {
+            Err(e) => Json(Response {
                 status: false,
-                error: Some("CLUSTER_ERROR".to_string()),
+                error: Some(e.to_string()),
                 ..Default::default()
             }),
         }
@@ -118,7 +118,7 @@
         Query(limit): Query<u32>,
     ) -> Json<Response<Vec<PeerInfo>>> {
         match ctx.connector.peers(node_vnet_addr(node, CONNECTOR_RPC_PORT), GetPeerParams { room, page, limit }).await {
-            Some(res) => Json(Response {
+            Ok(res) => Json(Response {
                 status: true,
                 data: Some(
                     res.peers
@@ -151,9 +151,9 @@
                 }),
                 ..Default::default()
             }),
-            None => Json(Response {
+            Err(e) => Json(Response {
                 status: false,
-                error: Some("CLUSTER_ERROR".to_string()),
+                error: Some(e.to_string()),
                 ..Default::default()
             }),
         }
@@ -170,7 +170,7 @@
         Query(limit): Query<u32>,
     ) -> Json<Response<Vec<SessionInfo>>> {
         match ctx.connector.sessions(node_vnet_addr(node, CONNECTOR_RPC_PORT), GetParams { page, limit }).await {
-            Some(res) => Json(Response {
+            Ok(res) => Json(Response {
                 status: true,
                 data: Some(
                     res.sessions
@@ -204,9 +204,9 @@
                 }),
                 ..Default::default()
             }),
-            None => Json(Response {
+            Err(e) => Json(Response {
                 status: false,
-                error: Some("CLUSTER_ERROR".to_string()),
+                error: Some(e.to_string()),
                 ..Default::default()
             }),
         }
@@ -240,7 +240,7 @@
             )
             .await
         {
-            Some(res) => Json(Response {
+            Ok(res) => Json(Response {
                 status: true,
                 data: Some(
                     res.events
@@ -262,9 +262,9 @@
                 }),
                 ..Default::default()
             }),
-            None => Json(Response {
+            Err(e) => Json(Response {
                 status: false,
-                error: Some("CLUSTER_ERROR".to_string()),
+                error: Some(e.to_string()),
                 ..Default::default()
             }),
         }

bin/src/server/connector/remote_rpc_handler.rs

@@ -1,5 +1,6 @@
 use std::sync::Arc;
 
+use anyhow::{anyhow, Result};
 use media_server_connector::Querier;
 use media_server_protocol::protobuf::cluster_connector::{
     get_events::EventInfo, get_peers::PeerInfo, get_rooms::RoomInfo, get_sessions::SessionInfo, GetEventParams, GetEvents, GetParams, GetPeerParams, GetPeers, GetRooms, GetSessions,
@@ -16,13 +17,13 @@
 pub struct ConnectorRemoteRpcHandlerImpl {}
 
 impl MediaConnectorServiceHandler<Ctx> for ConnectorRemoteRpcHandlerImpl {
-    async fn rooms(&self, ctx: &Ctx, req: GetParams) -> Option<GetRooms> {
+    async fn rooms(&self, ctx: &Ctx, req: GetParams) -> Result<GetRooms> {
         log::info!("[ConnectorRemoteRpcHandler] on get rooms {req:?}");
         let response = match ctx.storage.rooms(req.page as usize, req.limit as usize).await {
             Ok(res) => res,
             Err(err) => {
                 log::error!("[ConnectorRemoteRpcHandler] on get rooms error {err}");
-                return None;
+                return Err(anyhow!("{err}"));
             }
         };
         log::info!("[ConnectorRemoteRpcHandler] on got {} rooms", response.data.len());
@@ -40,7 +41,7 @@
             })
             .collect::<Vec<_>>();
 
-        Some(GetRooms {
+        Ok(GetRooms {
             rooms,
             pagination: Some(Pagination {
                 total: response.total as u32,
@@ -49,13 +50,13 @@
         })
     }
 
-    async fn peers(&self, ctx: &Ctx, req: GetPeerParams) -> Option<GetPeers> {
+    async fn peers(&self, ctx: &Ctx, req: GetPeerParams) -> Result<GetPeers> {
         log::info!("[ConnectorRemoteRpcHandler] on get peers page {req:?}");
         let response = match ctx.storage.peers(req.room, req.page as usize, req.limit as usize).await {
             Ok(res) => res,
             Err(err) => {
                 log::error!("[ConnectorRemoteRpcHandler] on get peers error {err}");
-                return None;
+                return Err(anyhow!("{err}"));
             }
         };
         log::info!("[ConnectorRemoteRpcHandler] on got {} peers", response.data.len());
@@ -85,7 +86,7 @@
             })
             .collect::<Vec<_>>();
 
-        Some(GetPeers {
+        Ok(GetPeers {
             peers,
             pagination: Some(Pagination {
                 total: response.total as u32,
@@ -94,13 +95,13 @@
         })
     }
 
-    async fn sessions(&self, ctx: &Ctx, req: GetParams) -> Option<GetSessions> {
+    async fn sessions(&self, ctx: &Ctx, req: GetParams) -> Result<GetSessions> {
         log::info!("[ConnectorRemoteRpcHandler] on get sessions page {req:?}");
         let response = match ctx.storage.sessions(req.page as usize, req.limit as usize).await {
             Ok(res) => res,
             Err(err) => {
                 log::error!("[ConnectorRemoteRpcHandler] on get sessions error {err}");
-                return None;
+                return Err(anyhow!("{err}"));
             }
         };
         log::info!("[ConnectorRemoteRpcHandler] on got {} sessions", response.data.len());
@@ -130,7 +131,7 @@
                     .collect::<Vec<_>>(),
             })
             .collect::<Vec<_>>();
-        Some(GetSessions {
+        Ok(GetSessions {
             sessions,
             pagination: Some(Pagination {
                 total: response.total as u32,
@@ -139,13 +140,13 @@
         })
     }
 
-    async fn events(&self, ctx: &Ctx, req: GetEventParams) -> Option<GetEvents> {
+    async fn events(&self, ctx: &Ctx, req: GetEventParams) -> Result<GetEvents> {
         log::info!("[ConnectorRemoteRpcHandler] on get events page {req:?}");
         let response = match ctx.storage.events(req.session, req.start_ts, req.end_ts, req.page as usize, req.limit as usize).await {
             Ok(res) => res,
             Err(err) => {
                 log::error!("[ConnectorRemoteRpcHandler] on get events error {err}");
-                return None;
+                return Err(anyhow!("{err}"));
             }
         };
         log::info!("[ConnectorRemoteRpcHandler] on got {} events", response.data.len());
@@ -163,7 +164,7 @@
                 meta: e.meta.map(|m| m.to_string()),
             })
             .collect::<Vec<_>>();
-        Some(GetEvents {
+        Ok(GetEvents {
             events,
             pagination: Some(Pagination {
                 total: response.total as u32,