731 lines
23 KiB
Rust

//!
//!
use std::str::FromStr;
use super::models;
use crate::compositions;
use crate::repositories;
use beteran_common_rust as bcr;
use beteran_protobuf_rust as bpr;
use diesel::{
r2d2::{ConnectionManager, Pool},
PgConnection,
};
use prost::Message;
///
pub struct Service {
connection_broker: nats::asynk::Connection,
queue_broker: String,
pool: Pool<ConnectionManager<PgConnection>>,
site_repository: repositories::site::repository::Repository,
site_composition: compositions::site::composition::Composition,
bank_repository: repositories::bank::repository::Repository,
}
impl std::fmt::Debug for Service {
fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
f.debug_struct("Service of service.member.service.identity")
.finish()
}
}
impl Service {
///
pub fn new(
connection_broker: nats::asynk::Connection,
queue_broker: String,
pool: Pool<ConnectionManager<PgConnection>>,
) -> Service {
Service {
connection_broker,
queue_broker,
pool,
site_repository: repositories::site::repository::Repository::new(),
site_composition: compositions::site::composition::Composition::new(),
bank_repository: repositories::bank::repository::Repository::new(),
}
}
pub async fn subscribe(&self) -> std::result::Result<(), std::boxed::Box<dyn std::error::Error>> {
futures::try_join!(
self.create_bank(),
self.list_banks(),
self.get_bank(),
self.update_bank(),
self.delete_bank(),
)
.map(|_| ())
}
fn check_site(
&self,
conn: &diesel::PgConnection,
url: Option<String>,
site_id: uuid::Uuid,
) -> Result<repositories::site::models::Site, bcr::error::rpc::Error> {
match self
.site_composition
.select_by_url(conn, url, site_id)
.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,
})
})? {
Some(s) => Ok(s),
None => Err(bcr::error::rpc::Error::InvalidParams(
bcr::error::rpc::InvalidParams {
message: "invalid site_url information".to_string(),
detail: bcr::error::rpc::InvalidParamsDetail {
location: "request".to_string(),
param: "client.site_url".to_string(),
value: "".to_string(),
error_type: bcr::error::rpc::InvalidParamsType::None,
message: "".to_string(),
},
},
)),
}
}
async fn create_bank(&self) -> Result<(), Box<dyn std::error::Error>> {
let s = self
.connection_broker
.queue_subscribe(
bpr::ss::bank::SUBJECT_CREATE_BANK,
self.queue_broker.as_str(),
)
.await?;
while let Some(message) = s.next().await {
if let Err(e) = async {
let req =
bpr::ss::bank::CreateBankRequest::decode(message.data.as_slice()).map_err(|e| {
bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest {
message: format!("invalid request: {}", e),
})
})?;
let client = match req.client {
Some(c) => c,
None => {
return Err(bcr::error::rpc::Error::InvalidParams(
bcr::error::rpc::InvalidParams {
message: "invalid client information".to_string(),
detail: bcr::error::rpc::InvalidParamsDetail {
location: "request".to_string(),
param: "client".to_string(),
value: "".to_string(),
error_type: bcr::error::rpc::InvalidParamsType::Required,
message: "".to_string(),
},
},
));
}
};
let request = match req.request {
Some(r) => r,
None => {
return Err(bcr::error::rpc::Error::InvalidParams(
bcr::error::rpc::InvalidParams {
message: "invalid request information".to_string(),
detail: bcr::error::rpc::InvalidParamsDetail {
location: "request".to_string(),
param: "request".to_string(),
value: "".to_string(),
error_type: bcr::error::rpc::InvalidParamsType::Required,
message: "".to_string(),
},
},
));
}
};
let conn = self.pool.get().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 s = self
.bank_repository
.insert(
&conn,
&repositories::bank::models::NewBank {
name: request.name,
sort_order: request.sort_order as i32,
show: request.show,
can_use: request.can_use,
memo: request.memo,
},
)
.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,
})
})?;
message
.respond(
bpr::ss::bank::CreateBankResponse {
error: None,
result: Some(bpr::ss::bank::create_bank_response::Result {
bank: Some(bpr::models::bank::Bank::from(&s)),
}),
}
.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::ss::bank::CreateBankResponse {
error: Some(bpr::protobuf::rpc::Error::from(e)),
result: None,
}
.encode_to_vec(),
)
.await?;
}
}
Ok(())
}
async fn list_banks(&self) -> Result<(), Box<dyn std::error::Error>> {
let s = self
.connection_broker
.queue_subscribe(
bpr::ss::bank::SUBJECT_LIST_BANKS,
self.queue_broker.as_str(),
)
.await?;
while let Some(message) = s.next().await {
if let Err(e) = async {
let req =
bpr::ss::bank::ListBanksRequest::decode(message.data.as_slice()).map_err(|e| {
bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest {
message: format!("invalid request: {}", e),
})
})?;
let client = match req.client {
Some(c) => c,
None => {
return Err(bcr::error::rpc::Error::InvalidParams(
bcr::error::rpc::InvalidParams {
message: "invalid client information".to_string(),
detail: bcr::error::rpc::InvalidParamsDetail {
location: "request".to_string(),
param: "client".to_string(),
value: "".to_string(),
error_type: bcr::error::rpc::InvalidParamsType::Required,
message: "".to_string(),
},
},
));
}
};
let request = match req.request {
Some(r) => r,
None => {
return Err(bcr::error::rpc::Error::InvalidParams(
bcr::error::rpc::InvalidParams {
message: "invalid request information".to_string(),
detail: bcr::error::rpc::InvalidParamsDetail {
location: "request".to_string(),
param: "request".to_string(),
value: "".to_string(),
error_type: bcr::error::rpc::InvalidParamsType::Required,
message: "".to_string(),
},
},
));
}
};
let conn = self.pool.get().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 find_all = repositories::bank::models::FindAll {
search: match request.search {
Some(s) => Some(repositories::bank::models::FindAllSearch {
name_like: s.name_like,
memo_like: s.memo_like,
show: s.show,
can_use: s.can_use,
}),
None => None,
},
pagination: request
.pagination
.as_ref()
.map(bcr::pagination::Pagination::from),
sorts: Some(
request
.sorts
.iter()
.map(bcr::pagination::Sort::from)
.collect(),
),
};
let total_count = self
.bank_repository
.select_all_count(&conn, &find_all)
.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 list = self
.bank_repository
.select_all(&conn, &find_all)
.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,
})
})?;
message
.respond(
bpr::ss::bank::ListBanksResponse {
error: None,
result: Some(bpr::ss::bank::list_banks_response::Result {
total_count: total_count as u64,
banks: list
.iter()
.map(|d| bpr::models::bank::Bank::from(d))
.collect(),
}),
}
.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::ss::bank::ListBanksResponse {
error: Some(bpr::protobuf::rpc::Error::from(e)),
result: None,
}
.encode_to_vec(),
)
.await?;
}
}
Ok(())
}
async fn get_bank(&self) -> Result<(), Box<dyn std::error::Error>> {
let s = self
.connection_broker
.queue_subscribe(bpr::ss::bank::SUBJECT_GET_BANK, self.queue_broker.as_str())
.await?;
while let Some(message) = s.next().await {
if let Err(e) = async {
let req = bpr::ss::bank::GetBankRequest::decode(message.data.as_slice()).map_err(|e| {
bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest {
message: format!("invalid request: {}", e),
})
})?;
let client = match req.client {
Some(c) => c,
None => {
return Err(bcr::error::rpc::Error::InvalidParams(
bcr::error::rpc::InvalidParams {
message: "invalid client information".to_string(),
detail: bcr::error::rpc::InvalidParamsDetail {
location: "request".to_string(),
param: "client".to_string(),
value: "".to_string(),
error_type: bcr::error::rpc::InvalidParamsType::Required,
message: "".to_string(),
},
},
));
}
};
let request = match req.request {
Some(r) => r,
None => {
return Err(bcr::error::rpc::Error::InvalidParams(
bcr::error::rpc::InvalidParams {
message: "invalid request information".to_string(),
detail: bcr::error::rpc::InvalidParamsDetail {
location: "request".to_string(),
param: "request".to_string(),
value: "".to_string(),
error_type: bcr::error::rpc::InvalidParamsType::Required,
message: "".to_string(),
},
},
));
}
};
let id = uuid::Uuid::from_str(request.id.as_str()).map_err(|e| {
bcr::error::rpc::Error::InvalidParams(bcr::error::rpc::InvalidParams {
message: "invalid id param".to_string(),
detail: bcr::error::rpc::InvalidParamsDetail {
location: "request".to_string(),
param: "id".to_string(),
value: request.id.clone(),
error_type: bcr::error::rpc::InvalidParamsType::Required,
message: e.to_string(),
},
})
})?;
let conn = self.pool.get().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 s = self.bank_repository.select(&conn, id).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,
})
})?;
message
.respond(
bpr::ss::bank::GetBankResponse {
error: None,
result: Some(bpr::ss::bank::get_bank_response::Result {
bank: s.map(|d| bpr::models::bank::Bank::from(&d)),
}),
}
.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::ss::bank::GetBankResponse {
error: Some(bpr::protobuf::rpc::Error::from(e)),
result: None,
}
.encode_to_vec(),
)
.await?;
}
}
Ok(())
}
async fn update_bank(&self) -> Result<(), Box<dyn std::error::Error>> {
let s = self
.connection_broker
.queue_subscribe(
bpr::ss::bank::SUBJECT_UPDATE_BANK,
self.queue_broker.as_str(),
)
.await?;
while let Some(message) = s.next().await {
if let Err(e) = async {
let req =
bpr::ss::bank::UpdateBankRequest::decode(message.data.as_slice()).map_err(|e| {
bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest {
message: format!("invalid request: {}", e),
})
})?;
let client = match req.client {
Some(c) => c,
None => {
return Err(bcr::error::rpc::Error::InvalidParams(
bcr::error::rpc::InvalidParams {
message: "invalid client information".to_string(),
detail: bcr::error::rpc::InvalidParamsDetail {
location: "request".to_string(),
param: "client".to_string(),
value: "".to_string(),
error_type: bcr::error::rpc::InvalidParamsType::Required,
message: "".to_string(),
},
},
));
}
};
let request = match req.request {
Some(r) => r,
None => {
return Err(bcr::error::rpc::Error::InvalidParams(
bcr::error::rpc::InvalidParams {
message: "invalid request information".to_string(),
detail: bcr::error::rpc::InvalidParamsDetail {
location: "request".to_string(),
param: "request".to_string(),
value: "".to_string(),
error_type: bcr::error::rpc::InvalidParamsType::Required,
message: "".to_string(),
},
},
));
}
};
let id = uuid::Uuid::from_str(request.id.as_str()).map_err(|e| {
bcr::error::rpc::Error::InvalidParams(bcr::error::rpc::InvalidParams {
message: "invalid id param".to_string(),
detail: bcr::error::rpc::InvalidParamsDetail {
location: "request".to_string(),
param: "id".to_string(),
value: request.id.clone(),
error_type: bcr::error::rpc::InvalidParamsType::Required,
message: e.to_string(),
},
})
})?;
let conn = self.pool.get().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 _affected = self
.bank_repository
.update(
&conn,
id,
&repositories::bank::models::ModifyBank {
name: request.name,
sort_order: request.sort_order as i32,
show: request.show,
can_use: request.can_use,
memo: request.memo,
},
)
.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 s = self.bank_repository.select(&conn, id).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,
})
})?;
message
.respond(
bpr::ss::bank::UpdateBankResponse {
error: None,
result: Some(bpr::ss::bank::update_bank_response::Result {
bank: s.map(|d| bpr::models::bank::Bank::from(&d)),
}),
}
.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::ss::bank::UpdateBankResponse {
error: Some(bpr::protobuf::rpc::Error::from(e)),
result: None,
}
.encode_to_vec(),
)
.await?;
}
}
Ok(())
}
async fn delete_bank(&self) -> Result<(), Box<dyn std::error::Error>> {
let s = self
.connection_broker
.queue_subscribe(
bpr::ss::bank::SUBJECT_DELETE_BANK,
self.queue_broker.as_str(),
)
.await?;
while let Some(message) = s.next().await {
if let Err(e) = async {
let req =
bpr::ss::bank::DeleteBankRequest::decode(message.data.as_slice()).map_err(|e| {
bcr::error::rpc::Error::InvalidRequest(bcr::error::rpc::InvalidRequest {
message: format!("invalid request: {}", e),
})
})?;
let client = match req.client {
Some(c) => c,
None => {
return Err(bcr::error::rpc::Error::InvalidParams(
bcr::error::rpc::InvalidParams {
message: "invalid client information".to_string(),
detail: bcr::error::rpc::InvalidParamsDetail {
location: "request".to_string(),
param: "client".to_string(),
value: "".to_string(),
error_type: bcr::error::rpc::InvalidParamsType::Required,
message: "".to_string(),
},
},
));
}
};
let request = match req.request {
Some(r) => r,
None => {
return Err(bcr::error::rpc::Error::InvalidParams(
bcr::error::rpc::InvalidParams {
message: "invalid request information".to_string(),
detail: bcr::error::rpc::InvalidParamsDetail {
location: "request".to_string(),
param: "request".to_string(),
value: "".to_string(),
error_type: bcr::error::rpc::InvalidParamsType::Required,
message: "".to_string(),
},
},
));
}
};
let id = uuid::Uuid::from_str(request.id.as_str()).map_err(|e| {
bcr::error::rpc::Error::InvalidParams(bcr::error::rpc::InvalidParams {
message: "invalid id param".to_string(),
detail: bcr::error::rpc::InvalidParamsDetail {
location: "request".to_string(),
param: "id".to_string(),
value: request.id.clone(),
error_type: bcr::error::rpc::InvalidParamsType::Required,
message: e.to_string(),
},
})
})?;
let conn = self.pool.get().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 _affected = self.bank_repository.delete(&conn, id).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,
})
})?;
message
.respond(
bpr::ss::bank::DeleteBankResponse {
error: None,
result: Some(bpr::ss::bank::delete_bank_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::ss::bank::DeleteBankResponse {
error: Some(bpr::protobuf::rpc::Error::from(e)),
result: None,
}
.encode_to_vec(),
)
.await?;
}
}
Ok(())
}
}