diff --git a/Cargo.toml b/Cargo.toml index 4daee98..e4a85e0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,7 +19,7 @@ prost = { version = "0" } nats = { version = "0" } base64 = { version = "0" } -beteran-protobuf-rust = { git = "https://gitlab.loafle.net/bet/beteran-protobuf-rust.git", tag = "v0.1.40-snapshot" } -beteran-common-rust = { git = "https://gitlab.loafle.net/bet/beteran-common-rust.git", tag = "v0.1.23-snapshot" } +beteran-protobuf-rust = { git = "https://gitlab.loafle.net/bet/beteran-protobuf-rust.git", tag = "v0.1.42-snapshot" } +beteran-common-rust = { git = "https://gitlab.loafle.net/bet/beteran-common-rust.git", tag = "v0.1.25-snapshot" } [build-dependencies] diff --git a/src/main.rs b/src/main.rs index 8950b2b..afac4f6 100644 --- a/src/main.rs +++ b/src/main.rs @@ -2,6 +2,8 @@ use std::env; mod identity; mod member; +mod member_class; +mod member_level; mod site; #[tokio::main] @@ -21,6 +23,10 @@ async fn main() -> Result<(), Box> { let site_service = site::service::Service::new(connection_broker.clone(), queue_broker.clone()); let identity_service = identity::service::Service::new(connection_broker.clone(), queue_broker.clone()); + let member_class_service = + member_class::service::Service::new(connection_broker.clone(), queue_broker.clone()); + let member_level_service = + member_level::service::Service::new(connection_broker.clone(), queue_broker.clone()); let member_service = member::service::Service::new(connection_broker.clone(), queue_broker.clone()); @@ -29,7 +35,9 @@ async fn main() -> Result<(), Box> { futures::try_join!( site_service.subscribe(), identity_service.subscribe(), - member_service.subscribe() + member_class_service.subscribe(), + member_level_service.subscribe(), + member_service.subscribe(), )?; Ok(()) diff --git a/src/member_class/mod.rs b/src/member_class/mod.rs new file mode 100644 index 0000000..1f278a4 --- /dev/null +++ b/src/member_class/mod.rs @@ -0,0 +1 @@ +pub mod service; diff --git a/src/member_class/service.rs b/src/member_class/service.rs new file mode 100644 index 0000000..40deff8 --- /dev/null +++ b/src/member_class/service.rs @@ -0,0 +1,490 @@ +use beteran_common_rust as bcr; +use beteran_protobuf_rust as bpr; +use prost::Message; + +/// +pub struct Service { + connection_broker: nats::asynk::Connection, + queue_broker: String, +} + +impl std::fmt::Debug for Service { + fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { + f.debug_struct("beteran-backend-server-edge::member_class::service::Service") + .finish() + } +} + +impl Service { + /// + pub fn new(connection_broker: nats::asynk::Connection, queue_broker: String) -> Service { + Service { + connection_broker, + queue_broker, + } + } + + pub async fn subscribe(&self) -> std::result::Result<(), std::boxed::Box> { + futures::try_join!( + self.list_member_classes(), + self.create_member_class(), + self.update_member_class(), + self.delete_member_class(), + ) + .map(|_| ()) + } + + fn get_client_in_header( + &self, + message: &nats::asynk::Message, + ) -> Result { + match &message.headers { + Some(headers) => { + let client = match headers.get(bpr::c2se::core::network::HEADER_CLIENT) { + Some(c) => { + let msg = base64::decode(c).map_err(|e| { + bcr::error::rpc::Error::Parse(bcr::error::rpc::Parse { + message: format!("invalid header: {}", e), + }) + })?; + bpr::models::core::network::Client::decode(msg.as_slice()).map_err(|e| { + bcr::error::rpc::Error::Parse(bcr::error::rpc::Parse { + message: format!("invalid header: {}", e), + }) + })? + } + None => { + return Err(bcr::error::rpc::Error::Parse(bcr::error::rpc::Parse { + message: "invalid client information".to_string(), + })); + } + }; + + Ok(client) + } + None => Err(bcr::error::rpc::Error::Parse(bcr::error::rpc::Parse { + message: "invalid header".to_string(), + })), + } + } + + async fn create_member_class(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::backend::member_class::SUBJECT_CREATE_MEMBER_CLASS, + self.queue_broker.as_str(), + ) + .await?; + + while let Some(message) = s.next().await { + if let Err(e) = async { + let client = self.get_client_in_header(&message)?; + + let req = bpr::c2se::backend::member_class::CreateMemberClassRequest::decode( + message.data.as_slice(), + ) + .map_err(|e| { + bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest { + message: format!("invalid request: {}", e), + }) + })?; + + let ss_create_member_class_req = bpr::ss::member::member_class::CreateMemberClassRequest { + client: Some(client), + parent_id: req.parent_id, + name: req.name, + }; + + let ss_create_member_class_res_msg = self + .connection_broker + .request( + bpr::ss::member::member_class::SUBJECT_CREATE_MEMBER_CLASS, + ss_create_member_class_req.encode_to_vec(), + ) + .await + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + let ss_create_member_class_res = + bpr::ss::member::member_class::CreateMemberClassResponse::decode( + ss_create_member_class_res_msg.data.as_slice(), + ) + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + if let Some(e) = ss_create_member_class_res.error { + return Err(bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + })); + } + + if let Some(r) = ss_create_member_class_res.result { + message + .respond( + bpr::c2se::backend::member_class::CreateMemberClassResponse { + error: None, + result: Some( + bpr::c2se::backend::member_class::create_member_class_response::Result { + member_class: r.member_class, + }, + ), + } + .encode_to_vec(), + ) + .await + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + } + + Ok::<(), bcr::error::rpc::Error>(()) + } + .await + { + message + .respond( + bpr::c2se::backend::member_class::CreateMemberClassResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } + + async fn list_member_classes(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::backend::member_class::SUBJECT_LIST_MEMBER_CLASSES, + self.queue_broker.as_str(), + ) + .await?; + + while let Some(message) = s.next().await { + if let Err(e) = async { + let client = self.get_client_in_header(&message)?; + + let req = bpr::c2se::backend::member_class::ListMemberClassesRequest::decode( + message.data.as_slice(), + ) + .map_err(|e| { + bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest { + message: format!("invalid request: {}", e), + }) + })?; + + let ss_list_member_classes_req = bpr::ss::member::member_class::ListMemberClassesRequest { + client: Some(client), + pagination: req.pagination, + searches: req.searches, + sorts: req.sorts, + name_like: req.name_like, + }; + + let ss_list_member_classes_res_msg = self + .connection_broker + .request( + bpr::ss::member::member_class::SUBJECT_LIST_MEMBER_CLASSES, + ss_list_member_classes_req.encode_to_vec(), + ) + .await + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + let ss_list_member_classes_res = + bpr::ss::member::member_class::ListMemberClassesResponse::decode( + ss_list_member_classes_res_msg.data.as_slice(), + ) + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + if let Some(e) = ss_list_member_classes_res.error { + return Err(bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + })); + } + + if let Some(r) = ss_list_member_classes_res.result { + message + .respond( + bpr::c2se::backend::member_class::ListMemberClassesResponse { + error: None, + result: Some( + bpr::c2se::backend::member_class::list_member_classes_response::Result { + member_classes: r.member_classes, + }, + ), + } + .encode_to_vec(), + ) + .await + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + } + + Ok::<(), bcr::error::rpc::Error>(()) + } + .await + { + message + .respond( + bpr::c2se::backend::member_class::ListMemberClassesResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } + + async fn update_member_class(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::backend::member_class::SUBJECT_UPDATE_MEMBER_CLASS, + self.queue_broker.as_str(), + ) + .await?; + + while let Some(message) = s.next().await { + if let Err(e) = async { + let client = self.get_client_in_header(&message)?; + + let req = bpr::c2se::backend::member_class::UpdateMemberClassRequest::decode( + message.data.as_slice(), + ) + .map_err(|e| { + bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest { + message: format!("invalid request: {}", e), + }) + })?; + + let ss_update_member_class_req = bpr::ss::member::member_class::UpdateMemberClassRequest { + client: Some(client), + id: req.id, + parent_id: req.parent_id, + name: req.name, + }; + + let ss_update_member_class_res_msg = self + .connection_broker + .request( + bpr::ss::member::member_class::SUBJECT_UPDATE_MEMBER_CLASS, + ss_update_member_class_req.encode_to_vec(), + ) + .await + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + let ss_update_member_class_res = + bpr::ss::member::member_class::UpdateMemberClassResponse::decode( + ss_update_member_class_res_msg.data.as_slice(), + ) + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + if let Some(e) = ss_update_member_class_res.error { + return Err(bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + })); + } + + if let Some(r) = ss_update_member_class_res.result { + message + .respond( + bpr::c2se::backend::member_class::UpdateMemberClassResponse { + error: None, + result: Some( + bpr::c2se::backend::member_class::update_member_class_response::Result { + member_class: r.member_class, + }, + ), + } + .encode_to_vec(), + ) + .await + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + } + + Ok::<(), bcr::error::rpc::Error>(()) + } + .await + { + message + .respond( + bpr::c2se::backend::member_class::UpdateMemberClassResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } + + async fn delete_member_class(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::backend::member_class::SUBJECT_DELETE_MEMBER_CLASS, + self.queue_broker.as_str(), + ) + .await?; + + while let Some(message) = s.next().await { + if let Err(e) = async { + let client = self.get_client_in_header(&message)?; + + let req = bpr::c2se::backend::member_class::DeleteMemberClassRequest::decode( + message.data.as_slice(), + ) + .map_err(|e| { + bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest { + message: format!("invalid request: {}", e), + }) + })?; + + let ss_delete_member_class_req = bpr::ss::member::member_class::DeleteMemberClassRequest { + client: Some(client), + id: req.id, + }; + + let ss_delete_member_class_res_msg = self + .connection_broker + .request( + bpr::ss::member::member_class::SUBJECT_DELETE_MEMBER_CLASS, + ss_delete_member_class_req.encode_to_vec(), + ) + .await + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + let ss_delete_member_class_res = + bpr::ss::member::member_class::DeleteMemberClassResponse::decode( + ss_delete_member_class_res_msg.data.as_slice(), + ) + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + if let Some(e) = ss_delete_member_class_res.error { + return Err(bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + })); + } + + if let Some(r) = ss_delete_member_class_res.result { + message + .respond( + bpr::c2se::backend::member_class::DeleteMemberClassResponse { + error: None, + result: Some( + bpr::c2se::backend::member_class::delete_member_class_response::Result {}, + ), + } + .encode_to_vec(), + ) + .await + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + } + + Ok::<(), bcr::error::rpc::Error>(()) + } + .await + { + message + .respond( + bpr::c2se::backend::member_class::DeleteMemberClassResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } +} diff --git a/src/member_level/mod.rs b/src/member_level/mod.rs new file mode 100644 index 0000000..1f278a4 --- /dev/null +++ b/src/member_level/mod.rs @@ -0,0 +1 @@ +pub mod service; diff --git a/src/member_level/service.rs b/src/member_level/service.rs new file mode 100644 index 0000000..0d47e44 --- /dev/null +++ b/src/member_level/service.rs @@ -0,0 +1,490 @@ +use beteran_common_rust as bcr; +use beteran_protobuf_rust as bpr; +use prost::Message; + +/// +pub struct Service { + connection_broker: nats::asynk::Connection, + queue_broker: String, +} + +impl std::fmt::Debug for Service { + fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { + f.debug_struct("beteran-backend-server-edge::member_level::service::Service") + .finish() + } +} + +impl Service { + /// + pub fn new(connection_broker: nats::asynk::Connection, queue_broker: String) -> Service { + Service { + connection_broker, + queue_broker, + } + } + + pub async fn subscribe(&self) -> std::result::Result<(), std::boxed::Box> { + futures::try_join!( + self.list_member_levels(), + self.create_member_level(), + self.update_member_level(), + self.delete_member_level(), + ) + .map(|_| ()) + } + + fn get_client_in_header( + &self, + message: &nats::asynk::Message, + ) -> Result { + match &message.headers { + Some(headers) => { + let client = match headers.get(bpr::c2se::core::network::HEADER_CLIENT) { + Some(c) => { + let msg = base64::decode(c).map_err(|e| { + bcr::error::rpc::Error::Parse(bcr::error::rpc::Parse { + message: format!("invalid header: {}", e), + }) + })?; + bpr::models::core::network::Client::decode(msg.as_slice()).map_err(|e| { + bcr::error::rpc::Error::Parse(bcr::error::rpc::Parse { + message: format!("invalid header: {}", e), + }) + })? + } + None => { + return Err(bcr::error::rpc::Error::Parse(bcr::error::rpc::Parse { + message: "invalid client information".to_string(), + })); + } + }; + + Ok(client) + } + None => Err(bcr::error::rpc::Error::Parse(bcr::error::rpc::Parse { + message: "invalid header".to_string(), + })), + } + } + + async fn create_member_level(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::backend::member_level::SUBJECT_CREATE_MEMBER_LEVEL, + self.queue_broker.as_str(), + ) + .await?; + + while let Some(message) = s.next().await { + if let Err(e) = async { + let client = self.get_client_in_header(&message)?; + + let req = bpr::c2se::backend::member_level::CreateMemberLevelRequest::decode( + message.data.as_slice(), + ) + .map_err(|e| { + bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest { + message: format!("invalid request: {}", e), + }) + })?; + + let ss_create_member_level_req = bpr::ss::member::member_level::CreateMemberLevelRequest { + client: Some(client), + name: req.name, + sort_order: req.sort_order, + }; + + let ss_create_member_level_res_msg = self + .connection_broker + .request( + bpr::ss::member::member_level::SUBJECT_CREATE_MEMBER_LEVEL, + ss_create_member_level_req.encode_to_vec(), + ) + .await + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + let ss_create_member_level_res = + bpr::ss::member::member_level::CreateMemberLevelResponse::decode( + ss_create_member_level_res_msg.data.as_slice(), + ) + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + if let Some(e) = ss_create_member_level_res.error { + return Err(bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + })); + } + + if let Some(r) = ss_create_member_level_res.result { + message + .respond( + bpr::c2se::backend::member_level::CreateMemberLevelResponse { + error: None, + result: Some( + bpr::c2se::backend::member_level::create_member_level_response::Result { + member_level: r.member_level, + }, + ), + } + .encode_to_vec(), + ) + .await + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + } + + Ok::<(), bcr::error::rpc::Error>(()) + } + .await + { + message + .respond( + bpr::c2se::backend::member_level::CreateMemberLevelResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } + + async fn list_member_levels(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::backend::member_level::SUBJECT_LIST_MEMBER_LEVELS, + self.queue_broker.as_str(), + ) + .await?; + + while let Some(message) = s.next().await { + if let Err(e) = async { + let client = self.get_client_in_header(&message)?; + + let req = bpr::c2se::backend::member_level::ListMemberLevelsRequest::decode( + message.data.as_slice(), + ) + .map_err(|e| { + bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest { + message: format!("invalid request: {}", e), + }) + })?; + + let ss_list_member_levels_req = bpr::ss::member::member_level::ListMemberLevelsRequest { + client: Some(client), + pagination: req.pagination, + searches: req.searches, + sorts: req.sorts, + name_like: req.name_like, + }; + + let ss_list_member_levels_res_msg = self + .connection_broker + .request( + bpr::ss::member::member_level::SUBJECT_LIST_MEMBER_LEVELS, + ss_list_member_levels_req.encode_to_vec(), + ) + .await + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + let ss_list_member_levels_res = + bpr::ss::member::member_level::ListMemberLevelsResponse::decode( + ss_list_member_levels_res_msg.data.as_slice(), + ) + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + if let Some(e) = ss_list_member_levels_res.error { + return Err(bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + })); + } + + if let Some(r) = ss_list_member_levels_res.result { + message + .respond( + bpr::c2se::backend::member_level::ListMemberLevelsResponse { + error: None, + result: Some( + bpr::c2se::backend::member_level::list_member_levels_response::Result { + member_levels: r.member_levels, + }, + ), + } + .encode_to_vec(), + ) + .await + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + } + + Ok::<(), bcr::error::rpc::Error>(()) + } + .await + { + message + .respond( + bpr::c2se::backend::member_level::ListMemberLevelsResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } + + async fn update_member_level(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::backend::member_level::SUBJECT_UPDATE_MEMBER_LEVEL, + self.queue_broker.as_str(), + ) + .await?; + + while let Some(message) = s.next().await { + if let Err(e) = async { + let client = self.get_client_in_header(&message)?; + + let req = bpr::c2se::backend::member_level::UpdateMemberLevelRequest::decode( + message.data.as_slice(), + ) + .map_err(|e| { + bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest { + message: format!("invalid request: {}", e), + }) + })?; + + let ss_update_member_level_req = bpr::ss::member::member_level::UpdateMemberLevelRequest { + client: Some(client), + id: req.id, + parent_id: req.parent_id, + name: req.name, + }; + + let ss_update_member_level_res_msg = self + .connection_broker + .request( + bpr::ss::member::member_level::SUBJECT_UPDATE_MEMBER_LEVEL, + ss_update_member_level_req.encode_to_vec(), + ) + .await + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + let ss_update_member_level_res = + bpr::ss::member::member_level::UpdateMemberLevelResponse::decode( + ss_update_member_level_res_msg.data.as_slice(), + ) + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + if let Some(e) = ss_update_member_level_res.error { + return Err(bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + })); + } + + if let Some(r) = ss_update_member_level_res.result { + message + .respond( + bpr::c2se::backend::member_level::UpdateMemberLevelResponse { + error: None, + result: Some( + bpr::c2se::backend::member_level::update_member_level_response::Result { + member_level: r.member_level, + }, + ), + } + .encode_to_vec(), + ) + .await + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + } + + Ok::<(), bcr::error::rpc::Error>(()) + } + .await + { + message + .respond( + bpr::c2se::backend::member_level::UpdateMemberLevelResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } + + async fn delete_member_level(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::backend::member_level::SUBJECT_DELETE_MEMBER_LEVEL, + self.queue_broker.as_str(), + ) + .await?; + + while let Some(message) = s.next().await { + if let Err(e) = async { + let client = self.get_client_in_header(&message)?; + + let req = bpr::c2se::backend::member_level::DeleteMemberLevelRequest::decode( + message.data.as_slice(), + ) + .map_err(|e| { + bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest { + message: format!("invalid request: {}", e), + }) + })?; + + let ss_delete_member_level_req = bpr::ss::member::member_level::DeleteMemberLevelRequest { + client: Some(client), + id: req.id, + }; + + let ss_delete_member_level_res_msg = self + .connection_broker + .request( + bpr::ss::member::member_level::SUBJECT_DELETE_MEMBER_LEVEL, + ss_delete_member_level_req.encode_to_vec(), + ) + .await + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + let ss_delete_member_level_res = + bpr::ss::member::member_level::DeleteMemberLevelResponse::decode( + ss_delete_member_level_res_msg.data.as_slice(), + ) + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + + if let Some(e) = ss_delete_member_level_res.error { + return Err(bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + })); + } + + if let Some(r) = ss_delete_member_level_res.result { + message + .respond( + bpr::c2se::backend::member_level::DeleteMemberLevelResponse { + error: None, + result: Some( + bpr::c2se::backend::member_level::delete_member_level_response::Result {}, + ), + } + .encode_to_vec(), + ) + .await + .map_err(|e| { + bcr::error::rpc::Error::Server(bcr::error::rpc::Server { + code: bpr::protobuf::rpc::Error::SERVER_00, + message: format!("server {}", e), + data: None, + }) + })?; + } + + Ok::<(), bcr::error::rpc::Error>(()) + } + .await + { + message + .respond( + bpr::c2se::backend::member_level::DeleteMemberLevelResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } +}