diff --git a/Cargo.toml b/Cargo.toml index e82f374..51bf07b 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.72-snapshot" } -beteran-common-rust = { git = "https://gitlab.loafle.net/bet/beteran-common-rust.git", tag = "v0.1.56-snapshot" } +beteran-protobuf-rust = { git = "https://gitlab.loafle.net/bet/beteran-protobuf-rust.git", tag = "v0.1.74-snapshot" } +beteran-common-rust = { git = "https://gitlab.loafle.net/bet/beteran-common-rust.git", tag = "v0.1.58-snapshot" } [build-dependencies] diff --git a/src/main.rs b/src/main.rs index f298c1b..148fc2f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -3,6 +3,10 @@ use std::env; mod api; mod bank; mod identity; +mod member; +mod member_bank_account; +mod member_bank_deposit; +mod member_bank_withdraw; mod member_referrer; #[tokio::main] @@ -22,6 +26,14 @@ async fn main() -> Result<(), Box> { let bank_service = bank::service::Service::new(connection_broker.clone(), queue_broker.clone()); let identity_service = identity::service::Service::new(connection_broker.clone(), queue_broker.clone()); + let member_service = + member::service::Service::new(connection_broker.clone(), queue_broker.clone()); + let member_bank_account_service = + member_bank_account::service::Service::new(connection_broker.clone(), queue_broker.clone()); + let member_bank_deposit_service = + member_bank_deposit::service::Service::new(connection_broker.clone(), queue_broker.clone()); + let member_bank_withdraw_service = + member_bank_withdraw::service::Service::new(connection_broker.clone(), queue_broker.clone()); let member_referrer_service = member_referrer::service::Service::new(connection_broker.clone(), queue_broker.clone()); @@ -35,6 +47,10 @@ async fn main() -> Result<(), Box> { futures::try_join!( bank_service.subscribe(), identity_service.subscribe(), + member_service.subscribe(), + member_bank_account_service.subscribe(), + member_bank_deposit_service.subscribe(), + member_bank_withdraw_service.subscribe(), member_referrer_service.subscribe(), api_vendor_service.subscribe(), api_game_service.subscribe(), diff --git a/src/member/mod.rs b/src/member/mod.rs new file mode 100644 index 0000000..1f278a4 --- /dev/null +++ b/src/member/mod.rs @@ -0,0 +1 @@ +pub mod service; diff --git a/src/member/service.rs b/src/member/service.rs new file mode 100644 index 0000000..e537a2d --- /dev/null +++ b/src/member/service.rs @@ -0,0 +1,791 @@ +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-frontend-server-edge::identity::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.create_member(), + self.get_member(), + self.get_member_by_username(), + self.update_member(), + self.update_member_for_password(), + self.delete_member(), + ) + .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(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::frontend::member::SUBJECT_CREATE_MEMBER, + 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::member::CreateMemberRequest::decode(message.data.as_slice()).map_err(|e| { + bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest { + message: format!("invalid request: {}", e), + }) + })?; + + let bank_account = match req.bank_account { + Some(a) => Some( + bpr::ss::member::create_member_request::request::BankAccount { + bank_id: a.bank_id, + name: a.name, + account_number: a.account_number, + exchange_password: a.exchange_password, + }, + ), + None => None, + }; + + let game_setting = + req.game_setting.map( + |a| bpr::ss::member::create_member_request::request::GameSetting { + can_bet_casino: a.can_bet_casino, + can_bet_slot: a.can_bet_slot, + can_bet_powerball: a.can_bet_powerball, + can_bet_powerladder: a.can_bet_powerladder, + can_bet_eos: a.can_bet_eos, + can_bet_bogglepowerball: a.can_bet_bogglepowerball, + }, + ); + + let settlement_setting = req.settlement_setting.map(|a| { + bpr::ss::member::create_member_request::request::SettlementSetting { + can_exchange: a.can_exchange, + can_first_charge_comp: a.can_first_charge_comp, + can_per_charge_comp: a.can_per_charge_comp, + manual_payment_type_for_partner: a.manual_payment_type_for_partner, + settlement_type: a.settlement_type, + rate_casino: a.rate_casino, + rate_casino_loosing: a.rate_casino_loosing, + rate_casino_bacara: a.rate_casino_bacara, + rate_casino_roulette: a.rate_casino_roulette, + rate_casino_dragon_tiger: a.rate_casino_dragon_tiger, + rate_slot: a.rate_slot, + rate_slot_loosing: a.rate_slot_loosing, + rate_powerball_single: a.rate_powerball_single, + rate_powerball_combo: a.rate_powerball_combo, + rate_powerladder_single: a.rate_powerladder_single, + rate_powerladder_combo: a.rate_powerladder_combo, + rate_eos_single: a.rate_eos_single, + rate_eos_combo: a.rate_eos_combo, + rate_bogglepowerball_single: a.rate_bogglepowerball_single, + rate_bogglepowerball_combo: a.rate_bogglepowerball_combo, + } + }); + + let ss_create_member_req = bpr::ss::member::CreateMemberRequest { + client: Some(client), + request: Some(bpr::ss::member::create_member_request::Request { + site_id: req.site_id, + member_level_id: req.member_level_id, + mobile_phone_number: req.mobile_phone_number, + member_class_id: req.member_class_id, + referrer_member_username: req.referrer_member_username, + username: req.username, + password: req.password, + nickname: req.nickname, + state: bpr::models::member::MemberState::Pending as i32, + bank_account, + game_setting, + settlement_setting, + }), + }; + + let ss_create_member_res_msg = self + .connection_broker + .request( + bpr::ss::member::SUBJECT_CREATE_MEMBER, + ss_create_member_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_res = + bpr::ss::member::CreateMemberResponse::decode(ss_create_member_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_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_res.result { + message + .respond( + bpr::c2se::member::CreateMemberResponse { + error: None, + result: Some(bpr::c2se::member::create_member_response::Result { + member: r.member, + }), + } + .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::member::CreateMemberResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } + + async fn get_member(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::frontend::member::SUBJECT_GET_MEMBER, + 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::member::GetMemberRequest::decode(message.data.as_slice()).map_err(|e| { + bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest { + message: format!("invalid request: {}", e), + }) + })?; + + let ss_get_member_req = bpr::ss::member::GetMemberRequest { + client: Some(client), + request: Some(bpr::ss::member::get_member_request::Request { id: req.id }), + }; + + let ss_get_member_res_msg = self + .connection_broker + .request( + bpr::ss::member::SUBJECT_GET_MEMBER, + ss_get_member_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_get_member_res = + bpr::ss::member::GetMemberResponse::decode(ss_get_member_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_get_member_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_get_member_res.result { + message + .respond( + bpr::c2se::member::GetMemberResponse { + error: None, + result: Some(bpr::c2se::member::get_member_response::Result { member: r.member }), + } + .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::member::GetMemberResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } + + async fn get_member_by_username(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::frontend::member::SUBJECT_GET_MEMBER_BY_USERNAME, + 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::member::GetMemberByUsernameRequest::decode(message.data.as_slice()) + .map_err(|e| { + bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest { + message: format!("invalid request: {}", e), + }) + })?; + + let ss_get_member_by_username_req = bpr::ss::member::GetMemberByUsernameRequest { + client: Some(client), + request: Some(bpr::ss::member::get_member_by_username_request::Request { + username: req.username, + }), + }; + + let ss_get_member_by_username_res_msg = self + .connection_broker + .request( + bpr::ss::member::SUBJECT_GET_MEMBER_BY_USERNAME, + ss_get_member_by_username_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_get_member_by_username_res = bpr::ss::member::GetMemberByUsernameResponse::decode( + ss_get_member_by_username_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_get_member_by_username_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_get_member_by_username_res.result { + message + .respond( + bpr::c2se::member::GetMemberByUsernameResponse { + error: None, + result: Some(bpr::c2se::member::get_member_by_username_response::Result { + member: r.member, + }), + } + .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::member::GetMemberByUsernameResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } + + async fn update_member(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::frontend::member::SUBJECT_UPDATE_MEMBER, + 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::member::UpdateMemberRequest::decode(message.data.as_slice()).map_err(|e| { + bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest { + message: format!("invalid request: {}", e), + }) + })?; + + let bank_account = match req.bank_account { + Some(a) => Some( + bpr::ss::member::update_member_request::request::BankAccount { + id: a.id, + bank_id: a.bank_id, + name: a.name, + account_number: a.account_number, + memo: a.memo, + }, + ), + None => None, + }; + + let game_setting = match req.game_setting { + Some(a) => Some( + bpr::ss::member::update_member_request::request::GameSetting { + id: a.id, + can_bet_casino: a.can_bet_casino, + can_bet_slot: a.can_bet_slot, + can_bet_powerball: a.can_bet_powerball, + can_bet_powerladder: a.can_bet_powerladder, + can_bet_eos: a.can_bet_eos, + can_bet_bogglepowerball: a.can_bet_bogglepowerball, + }, + ), + None => None, + }; + + let settlement_setting = match req.settlement_setting { + Some(a) => Some( + bpr::ss::member::update_member_request::request::SettlementSetting { + id: a.id, + can_exchange: a.can_exchange, + can_first_charge_comp: a.can_first_charge_comp, + can_per_charge_comp: a.can_per_charge_comp, + manual_payment_type_for_partner: a.manual_payment_type_for_partner, + settlement_type: a.settlement_type, + rate_casino: a.rate_casino, + rate_casino_loosing: a.rate_casino_loosing, + rate_casino_bacara: a.rate_casino_bacara, + rate_casino_roulette: a.rate_casino_roulette, + rate_casino_dragon_tiger: a.rate_casino_dragon_tiger, + rate_slot: a.rate_slot, + rate_slot_loosing: a.rate_slot_loosing, + rate_powerball_single: a.rate_powerball_single, + rate_powerball_combo: a.rate_powerball_combo, + rate_powerladder_single: a.rate_powerladder_single, + rate_powerladder_combo: a.rate_powerladder_combo, + rate_eos_single: a.rate_eos_single, + rate_eos_combo: a.rate_eos_combo, + rate_bogglepowerball_single: a.rate_bogglepowerball_single, + rate_bogglepowerball_combo: a.rate_bogglepowerball_combo, + }, + ), + None => None, + }; + + let ss_update_member_req = bpr::ss::member::UpdateMemberRequest { + client: Some(client), + request: Some(bpr::ss::member::update_member_request::Request { + id: req.id, + site_id: req.site_id, + member_level_id: req.member_level_id, + mobile_phone_number: req.mobile_phone_number, + bank_account, + game_setting, + settlement_setting, + }), + }; + + let ss_update_member_res_msg = self + .connection_broker + .request( + bpr::ss::member::SUBJECT_UPDATE_MEMBER, + ss_update_member_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_res = + bpr::ss::member::UpdateMemberResponse::decode(ss_update_member_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_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_res.result { + message + .respond( + bpr::c2se::member::UpdateMemberResponse { + error: None, + result: Some(bpr::c2se::member::update_member_response::Result { + member: r.member, + }), + } + .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::member::UpdateMemberResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } + + async fn update_member_for_password(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::frontend::member::SUBJECT_UPDATE_MEMBER_FOR_PASSWORD, + 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::member::UpdateMemberForPasswordRequest::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_req = bpr::ss::member::UpdateMemberForPasswordRequest { + client: Some(client), + request: Some( + bpr::ss::member::update_member_for_password_request::Request { + id: req.id, + password: req.password, + }, + ), + }; + + let ss_update_member_res_msg = self + .connection_broker + .request( + bpr::ss::member::SUBJECT_UPDATE_MEMBER_FOR_PASSWORD, + ss_update_member_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_res = bpr::ss::member::UpdateMemberForPasswordResponse::decode( + ss_update_member_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_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_res.result { + message + .respond( + bpr::c2se::member::UpdateMemberForPasswordResponse { + error: None, + result: Some(bpr::c2se::member::update_member_for_password_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::member::UpdateMemberForPasswordResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } + + async fn delete_member(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::frontend::member::SUBJECT_DELETE_MEMBER, + 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::member::DeleteMemberRequest::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_req = bpr::ss::member::DeleteMemberRequest { + client: Some(client), + request: Some(bpr::ss::member::delete_member_request::Request { id: req.id }), + }; + + let ss_delete_member_res_msg = self + .connection_broker + .request( + bpr::ss::member::SUBJECT_DELETE_MEMBER, + ss_delete_member_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_res = + bpr::ss::member::DeleteMemberResponse::decode(ss_delete_member_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_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_res.result { + message + .respond( + bpr::c2se::member::DeleteMemberResponse { + error: None, + result: Some(bpr::c2se::member::delete_member_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::member::DeleteMemberResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } +} diff --git a/src/member_bank_account/mod.rs b/src/member_bank_account/mod.rs new file mode 100644 index 0000000..1f278a4 --- /dev/null +++ b/src/member_bank_account/mod.rs @@ -0,0 +1 @@ +pub mod service; diff --git a/src/member_bank_account/service.rs b/src/member_bank_account/service.rs new file mode 100644 index 0000000..f5e72ed --- /dev/null +++ b/src/member_bank_account/service.rs @@ -0,0 +1,508 @@ +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-frontend-server-edge::member_bank_account::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.create_member_bank_account(), + self.get_member_bank_account(), + self.update_member_bank_account(), + self.update_member_bank_account_for_exchange_password(), + ) + .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_bank_account(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::frontend::member_bank_account::SUBJECT_CREATE_MEMBER_BANK_ACCOUNT, + 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::member_bank_account::CreateMemberBankAccountRequest::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_bank_account_req = + bpr::ss::member_bank_account::CreateMemberBankAccountRequest { + client: Some(client), + request: Some( + bpr::ss::member_bank_account::create_member_bank_account_request::Request { + member_id: req.member_id, + bank_id: req.bank_id, + name: req.name, + account_number: req.account_number, + exchange_password: req.exchange_password, + memo: req.memo, + }, + ), + }; + + let ss_create_member_bank_account_res_msg = self + .connection_broker + .request( + bpr::ss::member_bank_account::SUBJECT_CREATE_MEMBER_BANK_ACCOUNT, + ss_create_member_bank_account_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_bank_account_res = + bpr::ss::member_bank_account::CreateMemberBankAccountResponse::decode( + ss_create_member_bank_account_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_bank_account_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_bank_account_res.result { + message + .respond( + bpr::c2se::member_bank_account::CreateMemberBankAccountResponse { + error: None, + result: Some( + bpr::c2se::member_bank_account::create_member_bank_account_response::Result { + member_bank_account: r.member_bank_account, + }, + ), + } + .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::member_bank_account::CreateMemberBankAccountResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } + + async fn get_member_bank_account(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::frontend::member_bank_account::SUBJECT_GET_MEMBER_BANK_ACCOUNT, + 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::member_bank_account::GetMemberBankAccountRequest::decode( + message.data.as_slice(), + ) + .map_err(|e| { + bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest { + message: format!("invalid request: {}", e), + }) + })?; + + let ss_get_member_bank_account_req = + bpr::ss::member_bank_account::GetMemberBankAccountRequest { + client: Some(client), + request: Some( + bpr::ss::member_bank_account::get_member_bank_account_request::Request { id: req.id }, + ), + }; + + let ss_get_member_bank_account_res_msg = self + .connection_broker + .request( + bpr::ss::member_bank_account::SUBJECT_GET_MEMBER_BANK_ACCOUNT, + ss_get_member_bank_account_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_get_member_bank_account_res = + bpr::ss::member_bank_account::GetMemberBankAccountResponse::decode( + ss_get_member_bank_account_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_get_member_bank_account_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_get_member_bank_account_res.result { + message + .respond( + bpr::c2se::member_bank_account::GetMemberBankAccountResponse { + error: None, + result: Some( + bpr::c2se::member_bank_account::get_member_bank_account_response::Result { + member_bank_account: r.member_bank_account, + }, + ), + } + .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::member_bank_account::GetMemberBankAccountResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } + + async fn update_member_bank_account(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::frontend::member_bank_account::SUBJECT_UPDATE_MEMBER_BANK_ACCOUNT, + 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::member_bank_account::UpdateMemberBankAccountRequest::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_bank_account_req = + bpr::ss::member_bank_account::UpdateMemberBankAccountRequest { + client: Some(client), + request: Some( + bpr::ss::member_bank_account::update_member_bank_account_request::Request { + id: req.id, + bank_id: req.bank_id, + name: req.name, + account_number: req.account_number, + memo: req.memo, + }, + ), + }; + + let ss_update_member_bank_account_res_msg = self + .connection_broker + .request( + bpr::ss::member_bank_account::SUBJECT_UPDATE_MEMBER_BANK_ACCOUNT, + ss_update_member_bank_account_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_bank_account_res = + bpr::ss::member_bank_account::UpdateMemberBankAccountResponse::decode( + ss_update_member_bank_account_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_bank_account_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_bank_account_res.result { + message + .respond( + bpr::c2se::member_bank_account::UpdateMemberBankAccountResponse { + error: None, + result: Some( + bpr::c2se::member_bank_account::update_member_bank_account_response::Result { + member_bank_account: r.member_bank_account, + }, + ), + } + .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::member_bank_account::UpdateMemberBankAccountResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } + + async fn update_member_bank_account_for_exchange_password( + &self, + ) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::frontend::member_bank_account::SUBJECT_UPDATE_MEMBER_BANK_ACCOUNT_FOR_EXCHANGE_PASSWORD, + 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::member_bank_account::UpdateMemberBankAccountForExchangePasswordRequest::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_bank_account_req = bpr::ss::member_bank_account::UpdateMemberBankAccountForExchangePasswordRequest { + client: Some(client), + request: Some(bpr::ss::member_bank_account::update_member_bank_account_for_exchange_password_request::Request{ + id: req.id, + exchange_password: req.exchange_password, + + }) + }; + + let ss_update_member_bank_account_res_msg = self + .connection_broker + .request( + bpr::ss::member_bank_account::SUBJECT_UPDATE_MEMBER_BANK_ACCOUNT_FOR_EXCHANGE_PASSWORD, + ss_update_member_bank_account_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_bank_account_res = bpr::ss::member_bank_account::UpdateMemberBankAccountForExchangePasswordResponse::decode( + ss_update_member_bank_account_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_bank_account_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_bank_account_res.result { + message + .respond( + bpr::c2se::member_bank_account::UpdateMemberBankAccountForExchangePasswordResponse { + error: None, + result: Some(bpr::c2se::member_bank_account::update_member_bank_account_for_exchange_password_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::member_bank_account::UpdateMemberBankAccountForExchangePasswordResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } +} diff --git a/src/member_bank_deposit/mod.rs b/src/member_bank_deposit/mod.rs new file mode 100644 index 0000000..1f278a4 --- /dev/null +++ b/src/member_bank_deposit/mod.rs @@ -0,0 +1 @@ +pub mod service; diff --git a/src/member_bank_deposit/service.rs b/src/member_bank_deposit/service.rs new file mode 100644 index 0000000..3b127f8 --- /dev/null +++ b/src/member_bank_deposit/service.rs @@ -0,0 +1,621 @@ +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-frontend-server-edge::member_bank_deposit::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.create_member_bank_deposit(), + self.list_member_bank_deposits(), + self.get_member_bank_deposit(), + self.update_member_bank_deposit(), + self.delete_member_bank_deposit(), + ) + .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_bank_deposit(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::frontend::member_bank_deposit::SUBJECT_CREATE_MEMBER_BANK_DEPOSIT, + 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::member_bank_deposit::CreateMemberBankDepositRequest::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_bank_deposit_req = + bpr::ss::member_bank_deposit::CreateMemberBankDepositRequest { + client: Some(client), + request: Some( + bpr::ss::member_bank_deposit::create_member_bank_deposit_request::Request { + member_id: "".to_string(), + name: req.name, + amount: req.amount, + memo: req.memo, + }, + ), + }; + + let ss_create_member_bank_deposit_res_msg = self + .connection_broker + .request( + bpr::ss::member_bank_deposit::SUBJECT_CREATE_MEMBER_BANK_DEPOSIT, + ss_create_member_bank_deposit_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_bank_deposit_res = + bpr::ss::member_bank_deposit::CreateMemberBankDepositResponse::decode( + ss_create_member_bank_deposit_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_bank_deposit_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_bank_deposit_res.result { + message + .respond( + bpr::c2se::member_bank_deposit::CreateMemberBankDepositResponse { + error: None, + result: Some( + bpr::c2se::member_bank_deposit::create_member_bank_deposit_response::Result { + member_bank_deposit: r.member_bank_deposit, + }, + ), + } + .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::member_bank_deposit::CreateMemberBankDepositResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } + + async fn list_member_bank_deposits(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::frontend::member_bank_deposit::SUBJECT_LIST_MEMBER_BANK_DEPOSITS, + 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::member_bank_deposit::ListMemberBankDepositsRequest::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_bank_deposits_req = bpr::ss::member_bank_deposit::ListMemberBankDepositsRequest { + client: Some(client), + request: Some(bpr::ss::member_bank_deposit::list_member_bank_deposits_request::Request { + pagination: req.pagination, + sorts: req.sorts, + search: match req.search { + Some(s) => Some(bpr::ss::member_bank_deposit::list_member_bank_deposits_request::request::Search { + member_id: s.member_id, + name_like: s.name_like, + amount: s.amount, + state: s.state, + memo_like: s.memo_like, + }), + None => None, + } + }) + + }; + + let ss_list_member_bank_deposits_res_msg = self + .connection_broker + .request( + bpr::ss::member_bank_deposit::SUBJECT_LIST_MEMBER_BANK_DEPOSITS, + ss_list_member_bank_deposits_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_bank_deposits_res = + bpr::ss::member_bank_deposit::ListMemberBankDepositsResponse::decode(ss_list_member_bank_deposits_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_bank_deposits_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_bank_deposits_res.result { + message + .respond( + bpr::c2se::member_bank_deposit::ListMemberBankDepositsResponse { + error: None, + result: Some(bpr::c2se::member_bank_deposit::list_member_bank_deposits_response::Result { + member_bank_deposits: r.member_bank_deposits, + }), + } + .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::member_bank_deposit::ListMemberBankDepositsResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } + + async fn get_member_bank_deposit(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::frontend::member_bank_deposit::SUBJECT_GET_MEMBER_BANK_DEPOSIT, + 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::member_bank_deposit::GetMemberBankDepositRequest::decode( + message.data.as_slice(), + ) + .map_err(|e| { + bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest { + message: format!("invalid request: {}", e), + }) + })?; + + let ss_get_member_bank_deposit_req = + bpr::ss::member_bank_deposit::GetMemberBankDepositRequest { + client: Some(client), + request: Some( + bpr::ss::member_bank_deposit::get_member_bank_deposit_request::Request { id: req.id }, + ), + }; + + let ss_get_member_bank_deposit_res_msg = self + .connection_broker + .request( + bpr::ss::member_bank_deposit::SUBJECT_GET_MEMBER_BANK_DEPOSIT, + ss_get_member_bank_deposit_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_get_member_bank_deposit_res = + bpr::ss::member_bank_deposit::GetMemberBankDepositResponse::decode( + ss_get_member_bank_deposit_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_get_member_bank_deposit_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_get_member_bank_deposit_res.result { + message + .respond( + bpr::c2se::member_bank_deposit::GetMemberBankDepositResponse { + error: None, + result: Some( + bpr::c2se::member_bank_deposit::get_member_bank_deposit_response::Result { + member_bank_deposit: r.member_bank_deposit, + }, + ), + } + .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::member_bank_deposit::GetMemberBankDepositResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } + + async fn update_member_bank_deposit(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::frontend::member_bank_deposit::SUBJECT_UPDATE_MEMBER_BANK_DEPOSIT, + 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::member_bank_deposit::UpdateMemberBankDepositRequest::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_bank_deposit_req = + bpr::ss::member_bank_deposit::UpdateMemberBankDepositRequest { + client: Some(client), + request: Some( + bpr::ss::member_bank_deposit::update_member_bank_deposit_request::Request { + id: req.id, + name: req.name, + amount: req.amount, + memo: req.memo, + }, + ), + }; + + let ss_update_member_bank_deposit_res_msg = self + .connection_broker + .request( + bpr::ss::member_bank_deposit::SUBJECT_UPDATE_MEMBER_BANK_DEPOSIT, + ss_update_member_bank_deposit_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_bank_deposit_res = + bpr::ss::member_bank_deposit::UpdateMemberBankDepositResponse::decode( + ss_update_member_bank_deposit_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_bank_deposit_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_bank_deposit_res.result { + message + .respond( + bpr::c2se::member_bank_deposit::UpdateMemberBankDepositResponse { + error: None, + result: Some( + bpr::c2se::member_bank_deposit::update_member_bank_deposit_response::Result { + member_bank_deposit: r.member_bank_deposit, + }, + ), + } + .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::member_bank_deposit::UpdateMemberBankDepositResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } + + async fn delete_member_bank_deposit(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::frontend::member_bank_deposit::SUBJECT_DELETE_MEMBER_BANK_DEPOSIT, + 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::member_bank_deposit::DeleteMemberBankDepositRequest::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_bank_deposit_req = + bpr::ss::member_bank_deposit::DeleteMemberBankDepositRequest { + client: Some(client), + request: Some( + bpr::ss::member_bank_deposit::delete_member_bank_deposit_request::Request { + id: req.id, + }, + ), + }; + + let ss_delete_member_bank_deposit_res_msg = self + .connection_broker + .request( + bpr::ss::member_bank_deposit::SUBJECT_DELETE_MEMBER_BANK_DEPOSIT, + ss_delete_member_bank_deposit_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_bank_deposit_res = + bpr::ss::member_bank_deposit::DeleteMemberBankDepositResponse::decode( + ss_delete_member_bank_deposit_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_bank_deposit_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_bank_deposit_res.result { + message + .respond( + bpr::c2se::member_bank_deposit::DeleteMemberBankDepositResponse { + error: None, + result: Some( + bpr::c2se::member_bank_deposit::delete_member_bank_deposit_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::member_bank_deposit::DeleteMemberBankDepositResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } +} diff --git a/src/member_bank_withdraw/mod.rs b/src/member_bank_withdraw/mod.rs new file mode 100644 index 0000000..1f278a4 --- /dev/null +++ b/src/member_bank_withdraw/mod.rs @@ -0,0 +1 @@ +pub mod service; diff --git a/src/member_bank_withdraw/service.rs b/src/member_bank_withdraw/service.rs new file mode 100644 index 0000000..7e0dd3d --- /dev/null +++ b/src/member_bank_withdraw/service.rs @@ -0,0 +1,604 @@ +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-frontend-server-edge::member_bank_withdraw::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.create_member_bank_withdraw(), + self.list_member_bank_withdraws(), + self.get_member_bank_withdraw(), + self.update_member_bank_withdraw(), + self.delete_member_bank_withdraw(), + ) + .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_bank_withdraw(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::frontend::member_bank_withdraw::SUBJECT_CREATE_MEMBER_BANK_WITHDRAW, + 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::member_bank_withdraw::CreateMemberBankWithdrawRequest::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_bank_withdraw_req = + bpr::ss::member_bank_withdraw::CreateMemberBankWithdrawRequest { + client: Some(client), + request: Some( + bpr::ss::member_bank_withdraw::create_member_bank_withdraw_request::Request { + member_id: "".to_string(), + bank_name: req.bank_name, + account_number: req.account_number, + amount: req.amount, + password: req.password, + name: req.name, + memo: req.memo, + }, + ), + }; + + let ss_create_member_bank_withdraw_res_msg = self + .connection_broker + .request( + bpr::ss::member_bank_withdraw::SUBJECT_CREATE_MEMBER_BANK_WITHDRAW, + ss_create_member_bank_withdraw_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_bank_withdraw_res = + bpr::ss::member_bank_withdraw::CreateMemberBankWithdrawResponse::decode( + ss_create_member_bank_withdraw_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_bank_withdraw_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_bank_withdraw_res.result { + message + .respond( + bpr::c2se::member_bank_withdraw::CreateMemberBankWithdrawResponse { + error: None, + result: Some( + bpr::c2se::member_bank_withdraw::create_member_bank_withdraw_response::Result { + member_bank_withdraw: r.member_bank_withdraw, + }, + ), + } + .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::member_bank_withdraw::CreateMemberBankWithdrawResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } + + async fn list_member_bank_withdraws(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::frontend::member_bank_withdraw::SUBJECT_LIST_MEMBER_BANK_WITHDRAWS, + 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::member_bank_withdraw::ListMemberBankWithdrawsRequest::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_bank_withdraws_req = bpr::ss::member_bank_withdraw::ListMemberBankWithdrawsRequest { + client: Some(client), + request: Some(bpr::ss::member_bank_withdraw::list_member_bank_withdraws_request::Request{ + pagination: req.pagination, + sorts: req.sorts, + search: match req.search { + Some(s) => Some(bpr::ss::member_bank_withdraw::list_member_bank_withdraws_request::request::Search { + name_like: s.name_like, + member_id: s.member_id, + bank_name_like: s.bank_name_like, + account_number_like: s.account_number_like, + amount: s.amount, + state: s.state, + memo_like: s.memo_like, + }), + None => None, + } + }) + + }; + + let ss_list_member_bank_withdraws_res_msg = self + .connection_broker + .request( + bpr::ss::member_bank_withdraw::SUBJECT_LIST_MEMBER_BANK_WITHDRAWS, + ss_list_member_bank_withdraws_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_bank_withdraws_res = + bpr::ss::member_bank_withdraw::ListMemberBankWithdrawsResponse::decode(ss_list_member_bank_withdraws_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_bank_withdraws_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_bank_withdraws_res.result { + message + .respond( + bpr::c2se::member_bank_withdraw::ListMemberBankWithdrawsResponse { + error: None, + result: Some(bpr::c2se::member_bank_withdraw::list_member_bank_withdraws_response::Result { + member_bank_withdraws: r.member_bank_withdraws, + }), + } + .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::member_bank_withdraw::ListMemberBankWithdrawsResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } + + async fn get_member_bank_withdraw(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::frontend::member_bank_withdraw::SUBJECT_GET_MEMBER_BANK_WITHDRAW, + 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::member_bank_withdraw::GetMemberBankWithdrawRequest::decode(message.data.as_slice()) + .map_err(|e| { + bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest { + message: format!("invalid request: {}", e), + }) + })?; + + let ss_get_member_bank_withdraw_req = bpr::ss::member_bank_withdraw::GetMemberBankWithdrawRequest { + client: Some(client), + request: Some(bpr::ss::member_bank_withdraw::get_member_bank_withdraw_request::Request{ + id: req.id, + }) + }; + + let ss_get_member_bank_withdraw_res_msg = self + .connection_broker + .request( + bpr::ss::member_bank_withdraw::SUBJECT_GET_MEMBER_BANK_WITHDRAW, + ss_get_member_bank_withdraw_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_get_member_bank_withdraw_res = + bpr::ss::member_bank_withdraw::GetMemberBankWithdrawResponse::decode(ss_get_member_bank_withdraw_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_get_member_bank_withdraw_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_get_member_bank_withdraw_res.result { + message + .respond( + bpr::c2se::member_bank_withdraw::GetMemberBankWithdrawResponse { + error: None, + result: Some(bpr::c2se::member_bank_withdraw::get_member_bank_withdraw_response::Result { member_bank_withdraw: r.member_bank_withdraw }), + } + .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::member_bank_withdraw::GetMemberBankWithdrawResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } + + async fn update_member_bank_withdraw(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::frontend::member_bank_withdraw::SUBJECT_UPDATE_MEMBER_BANK_WITHDRAW, + 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::member_bank_withdraw::UpdateMemberBankWithdrawRequest::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_bank_withdraw_req = bpr::ss::member_bank_withdraw::UpdateMemberBankWithdrawRequest { + client: Some(client), + request: Some(bpr::ss::member_bank_withdraw::update_member_bank_withdraw_request::Request{ + id: req.id, + bank_name: req.bank_name, + account_number: req.account_number, +amount: req.amount, +password: req.password, + name: req.name, + memo: req.memo, + }) + }; + + let ss_update_member_bank_withdraw_res_msg = self + .connection_broker + .request( + bpr::ss::member_bank_withdraw::SUBJECT_UPDATE_MEMBER_BANK_WITHDRAW, + ss_update_member_bank_withdraw_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_bank_withdraw_res = bpr::ss::member_bank_withdraw::UpdateMemberBankWithdrawResponse::decode( + ss_update_member_bank_withdraw_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_bank_withdraw_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_bank_withdraw_res.result { + message + .respond( + bpr::c2se::member_bank_withdraw::UpdateMemberBankWithdrawResponse { + error: None, + result: Some(bpr::c2se::member_bank_withdraw::update_member_bank_withdraw_response::Result { + member_bank_withdraw: r.member_bank_withdraw, + }), + } + .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::member_bank_withdraw::UpdateMemberBankWithdrawResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } + + async fn delete_member_bank_withdraw(&self) -> Result<(), Box> { + let s = self + .connection_broker + .queue_subscribe( + bpr::c2se::frontend::member_bank_withdraw::SUBJECT_DELETE_MEMBER_BANK_WITHDRAW, + 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::member_bank_withdraw::DeleteMemberBankWithdrawRequest::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_bank_withdraw_req = bpr::ss::member_bank_withdraw::DeleteMemberBankWithdrawRequest { + client: Some(client), + request: Some(bpr::ss::member_bank_withdraw::delete_member_bank_withdraw_request::Request{ + id: req.id, + }) + }; + + let ss_delete_member_bank_withdraw_res_msg = self + .connection_broker + .request( + bpr::ss::member_bank_withdraw::SUBJECT_DELETE_MEMBER_BANK_WITHDRAW, + ss_delete_member_bank_withdraw_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_bank_withdraw_res = bpr::ss::member_bank_withdraw::DeleteMemberBankWithdrawResponse::decode( + ss_delete_member_bank_withdraw_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_bank_withdraw_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_bank_withdraw_res.result { + message + .respond( + bpr::c2se::member_bank_withdraw::DeleteMemberBankWithdrawResponse { + error: None, + result: Some(bpr::c2se::member_bank_withdraw::delete_member_bank_withdraw_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::member_bank_withdraw::DeleteMemberBankWithdrawResponse { + error: Some(bpr::protobuf::rpc::Error::from(e)), + result: None, + } + .encode_to_vec(), + ) + .await?; + } + } + + Ok(()) + } +}