site is added

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.25-snapshot" }
-beteran-common-rust = { git = "https://gitlab.loafle.net/bet/beteran-common-rust.git", tag = "v0.1.5-snapshot" }
+beteran-protobuf-rust = { git = "https://gitlab.loafle.net/bet/beteran-protobuf-rust.git", tag = "v0.1.27-snapshot" }
+beteran-common-rust = { git = "https://gitlab.loafle.net/bet/beteran-common-rust.git", tag = "v0.1.10-snapshot" }
 
 [build-dependencies]

src/main.rs

@@ -1,6 +1,7 @@
 use std::env;
 
 mod identity;
+mod site;
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
@@ -16,11 +17,13 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
   let broker_opts = nats::asynk::Options::new();
   let connection_broker = broker_opts.connect(url_broker).await?;
 
-  let identity_service = identity::service::Service::new(connection_broker, queue_broker);
+  let identity_service =
+    identity::service::Service::new(connection_broker.clone(), queue_broker.clone());
+  let site_service = site::service::Service::new(connection_broker.clone(), queue_broker.clone());
 
   println!("Server edge[beteran-backend-server-edge] is started");
 
-  futures::try_join!(identity_service.subscribe())?;
+  futures::try_join!(site_service.subscribe(), identity_service.subscribe())?;
 
   Ok(())
 }

src/site/mod.rs

@@ -0,0 +1 @@
+pub mod service;

src/site/service.rs

@@ -0,0 +1,467 @@
+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::site::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<dyn std::error::Error>> {
+    futures::try_join!(
+      self.list_sites(),
+      self.create_site(),
+      self.update_site(),
+      self.delete_site(),
+    )
+    .map(|_| ())
+  }
+
+  fn get_client_in_header(
+    &self,
+    message: &nats::asynk::Message,
+  ) -> Result<bpr::models::core::network::Client, bcr::error::rpc::Error> {
+    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 list_sites(&self) -> Result<(), Box<dyn std::error::Error>> {
+    let s = self
+      .connection_broker
+      .queue_subscribe(
+        bpr::c2se::backend::site::SUBJECT_LIST_SITES,
+        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::site::ListSitesRequest::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_sites_req = bpr::ss::domain::site::ListSitesRequest {
+          client: Some(client),
+          pagination: req.pagination,
+          searches: req.searches,
+          sorts: req.sorts,
+        };
+
+        let ss_list_sites_res_msg = self
+          .connection_broker
+          .request(
+            bpr::ss::domain::site::SUBJECT_LIST_SITES,
+            ss_list_sites_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_sites_res =
+          bpr::ss::domain::site::ListSitesResponse::decode(ss_list_sites_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_sites_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_sites_res.result {
+          message
+            .respond(
+              bpr::c2se::backend::site::ListSitesResponse {
+                error: None,
+                result: Some(bpr::c2se::backend::site::list_sites_response::Result {
+                  sites: r.sites,
+                }),
+              }
+              .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::site::ListSitesResponse {
+              error: Some(bpr::protobuf::rpc::Error::from(e)),
+              result: None,
+            }
+            .encode_to_vec(),
+          )
+          .await?;
+      }
+    }
+
+    Ok(())
+  }
+
+  async fn create_site(&self) -> Result<(), Box<dyn std::error::Error>> {
+    let s = self
+      .connection_broker
+      .queue_subscribe(
+        bpr::c2se::backend::site::SUBJECT_CREATE_SITE,
+        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::site::CreateSiteRequest::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_site_req = bpr::ss::domain::site::CreateSiteRequest {
+          client: Some(client),
+          url: req.url,
+          show: req.show,
+          can_use: req.can_use,
+        };
+
+        let ss_create_site_res_msg = self
+          .connection_broker
+          .request(
+            bpr::ss::domain::site::SUBJECT_CREATE_SITE,
+            ss_create_site_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_site_res =
+          bpr::ss::domain::site::CreateSiteResponse::decode(ss_create_site_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_site_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_site_res.result {
+          message
+            .respond(
+              bpr::c2se::backend::site::CreateSiteResponse {
+                error: None,
+                result: Some(bpr::c2se::backend::site::create_site_response::Result {
+                  site: r.site,
+                }),
+              }
+              .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::site::CreateSiteResponse {
+              error: Some(bpr::protobuf::rpc::Error::from(e)),
+              result: None,
+            }
+            .encode_to_vec(),
+          )
+          .await?;
+      }
+    }
+
+    Ok(())
+  }
+
+  async fn update_site(&self) -> Result<(), Box<dyn std::error::Error>> {
+    let s = self
+      .connection_broker
+      .queue_subscribe(
+        bpr::c2se::backend::site::SUBJECT_UPDATE_SITE,
+        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::site::UpdateSiteRequest::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_site_req = bpr::ss::domain::site::UpdateSiteRequest {
+          client: Some(client),
+          id: req.id,
+          url: req.url,
+          show: req.show,
+          can_use: req.can_use,
+        };
+
+        let ss_update_site_res_msg = self
+          .connection_broker
+          .request(
+            bpr::ss::domain::site::SUBJECT_UPDATE_SITE,
+            ss_update_site_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_site_res =
+          bpr::ss::domain::site::UpdateSiteResponse::decode(ss_update_site_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_site_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_site_res.result {
+          message
+            .respond(
+              bpr::c2se::backend::site::UpdateSiteResponse {
+                error: None,
+                result: Some(bpr::c2se::backend::site::update_site_response::Result {
+                  site: r.site,
+                }),
+              }
+              .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::site::UpdateSiteResponse {
+              error: Some(bpr::protobuf::rpc::Error::from(e)),
+              result: None,
+            }
+            .encode_to_vec(),
+          )
+          .await?;
+      }
+    }
+
+    Ok(())
+  }
+
+  async fn delete_site(&self) -> Result<(), Box<dyn std::error::Error>> {
+    let s = self
+      .connection_broker
+      .queue_subscribe(
+        bpr::c2se::backend::site::SUBJECT_DELETE_SITE,
+        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::site::DeleteSiteRequest::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_site_req = bpr::ss::domain::site::DeleteSiteRequest {
+          client: Some(client),
+          id: req.id,
+        };
+
+        let ss_delete_site_res_msg = self
+          .connection_broker
+          .request(
+            bpr::ss::domain::site::SUBJECT_DELETE_SITE,
+            ss_delete_site_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_site_res =
+          bpr::ss::domain::site::DeleteSiteResponse::decode(ss_delete_site_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_site_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_site_res.result {
+          message
+            .respond(
+              bpr::c2se::backend::site::DeleteSiteResponse {
+                error: None,
+                result: Some(bpr::c2se::backend::site::delete_site_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::site::DeleteSiteResponse {
+              error: Some(bpr::protobuf::rpc::Error::from(e)),
+              result: None,
+            }
+            .encode_to_vec(),
+          )
+          .await?;
+      }
+    }
+
+    Ok(())
+  }
+}