Lines
98.98 %
Functions
96.43 %
Branches
100 %
use std::{error::Error as StdError, sync::Arc};
use async_trait::async_trait;
use futures::TryStreamExt;
use mongodb::{
action::Find,
bson::{doc, DateTime, Document},
error::ErrorKind,
Cursor as MongoDbCursor, Database,
};
use serde::{Deserialize, Serialize};
use super::super::device_route::{
Cursor, DeviceRoute, DeviceRouteModel, ListOptions, ListQueryCond, QueryCond, SortKey,
UpdateQueryCond, Updates,
/// Model instance.
pub struct Model {
/// The associated database connection.
conn: Arc<Database>,
}
/// Cursor instance.
struct DbCursor {
/// The associated collection cursor.
cursor: MongoDbCursor<Schema>,
/// (Useless) only for Cursor trait implementation.
offset: u64,
/// MongoDB schema.
#[derive(Deserialize, Serialize)]
struct Schema {
#[serde(rename = "routeId")]
route_id: String,
#[serde(rename = "unitId")]
unit_id: String,
#[serde(rename = "unitCode")]
unit_code: String,
#[serde(rename = "applicationId")]
application_id: String,
#[serde(rename = "applicationCode")]
application_code: String,
#[serde(rename = "deviceId")]
device_id: String,
#[serde(rename = "networkId")]
network_id: String,
#[serde(rename = "networkCode")]
network_code: String,
#[serde(rename = "networkAddr")]
network_addr: String,
profile: String,
#[serde(rename = "createdAt")]
created_at: DateTime,
#[serde(rename = "modifiedAt")]
modified_at: DateTime,
const COL_NAME: &'static str = "deviceRoute";
impl Model {
/// To create the model instance with a database connection.
pub async fn new(conn: Arc<Database>) -> Result<Self, Box<dyn StdError>> {
let model = Model { conn };
model.init().await?;
Ok(model)
#[async_trait]
impl DeviceRouteModel for Model {
async fn init(&self) -> Result<(), Box<dyn StdError>> {
let indexes = vec![
doc! {"name": "routeId_1", "key": {"routeId": 1}, "unique": true},
doc! {
"name": "applicationId_1_deviceId_1",
"key": {"applicationId": 1, "deviceId": 1},
"unique": true
},
doc! {"name": "unitId_1", "key": {"unitId": 1}},
doc! {"name": "applicationId_1", "key": {"applicationId": 1}},
doc! {"name": "deviceId_1", "key": {"deviceId": 1}},
doc! {"name": "networkId_1", "key": {"networkId": 1}},
doc! {"name": "createdAt_1", "key": {"createdAt": 1}},
];
let command = doc! {
"createIndexes": COL_NAME,
"indexes": indexes,
self.conn.run_command(command).await?;
Ok(())
async fn count(&self, cond: &ListQueryCond) -> Result<u64, Box<dyn StdError>> {
let filter = get_list_query_filter(cond);
let count = self
.conn
.collection::<Schema>(COL_NAME)
.count_documents(filter)
.await?;
Ok(count)
async fn list(
&self,
opts: &ListOptions,
cursor: Option<Box<dyn Cursor>>,
) -> Result<(Vec<DeviceRoute>, Option<Box<dyn Cursor>>), Box<dyn StdError>> {
let mut cursor = match cursor {
None => {
let filter = get_list_query_filter(opts.cond);
Box::new(DbCursor::new(
build_find_options(opts, self.conn.collection::<Schema>(COL_NAME).find(filter))
.await?,
))
Some(cursor) => cursor,
let mut count: u64 = 0;
let mut list = Vec::new();
while let Some(item) = cursor.try_next().await? {
list.push(item);
if let Some(cursor_max) = opts.cursor_max {
count += 1;
if count >= cursor_max {
return Ok((list, Some(cursor)));
Ok((list, None))
async fn get(&self, route_id: &str) -> Result<Option<DeviceRoute>, Box<dyn StdError>> {
let filter = doc! {"routeId": route_id};
let mut cursor = self
.find(filter)
if let Some(route) = cursor.try_next().await? {
return Ok(Some(DeviceRoute {
route_id: route.route_id,
unit_id: route.unit_id,
unit_code: route.unit_code,
application_id: route.application_id,
application_code: route.application_code,
device_id: route.device_id,
network_id: route.network_id,
network_code: route.network_code,
network_addr: route.network_addr,
profile: route.profile,
created_at: route.created_at.into(),
modified_at: route.modified_at.into(),
}));
Ok(None)
async fn add(&self, route: &DeviceRoute) -> Result<(), Box<dyn StdError>> {
let item = Schema {
route_id: route.route_id.clone(),
unit_id: route.unit_id.clone(),
unit_code: route.unit_code.clone(),
application_id: route.application_id.clone(),
application_code: route.application_code.clone(),
device_id: route.device_id.clone(),
network_id: route.network_id.clone(),
network_code: route.network_code.clone(),
network_addr: route.network_addr.clone(),
profile: route.profile.clone(),
self.conn
.insert_one(item)
async fn add_bulk(&self, routes: &Vec<DeviceRoute>) -> Result<(), Box<dyn StdError>> {
let mut items = vec![];
for route in routes.iter() {
items.push(Schema {
});
if let Err(e) = self
.insert_many(items)
.ordered(false)
.await
{
match e.kind.as_ref() {
ErrorKind::InsertMany(imerr) => match imerr.write_errors.as_ref() {
None => return Err(Box::new(e)),
Some(errs) => {
for err in errs {
if err.code != 11000 {
return Err(Box::new(e));
()
_ => return Err(Box::new(e)),
async fn del(&self, cond: &QueryCond) -> Result<(), Box<dyn StdError>> {
let filter = get_query_filter(cond);
.delete_many(filter)
async fn update(
cond: &UpdateQueryCond,
updates: &Updates,
) -> Result<(), Box<dyn StdError>> {
let filter = get_update_query_filter(cond);
if let Some(updates) = get_update_doc(updates) {
.update_many(filter, updates)
return Ok(());
impl DbCursor {
/// To create the cursor instance with a collection cursor.
pub fn new(cursor: MongoDbCursor<Schema>) -> Self {
DbCursor { cursor, offset: 0 }
impl Cursor for DbCursor {
async fn try_next(&mut self) -> Result<Option<DeviceRoute>, Box<dyn StdError>> {
if let Some(item) = self.cursor.try_next().await? {
self.offset += 1;
route_id: item.route_id,
unit_id: item.unit_id,
unit_code: item.unit_code,
application_id: item.application_id,
application_code: item.application_code,
device_id: item.device_id,
network_id: item.network_id,
network_code: item.network_code,
network_addr: item.network_addr,
profile: item.profile,
created_at: item.created_at.into(),
modified_at: item.modified_at.into(),
fn offset(&self) -> u64 {
self.offset
/// Transforms query conditions to the MongoDB document.
fn get_query_filter(cond: &QueryCond) -> Document {
let mut filter = Document::new();
if let Some(value) = cond.route_id {
filter.insert("routeId", value);
if let Some(value) = cond.unit_id {
filter.insert("unitId", value);
if let Some(value) = cond.application_id {
filter.insert("applicationId", value);
if let Some(value) = cond.network_id {
filter.insert("networkId", value);
if let Some(value) = cond.device_id {
filter.insert("deviceId", value);
if let Some(value) = cond.network_addrs {
let mut in_cond = Document::new();
in_cond.insert("$in", value);
filter.insert("networkAddr", in_cond);
filter
fn get_list_query_filter(cond: &ListQueryCond) -> Document {
if let Some(value) = cond.unit_code {
filter.insert("unitCode", value);
if let Some(value) = cond.application_code {
filter.insert("applicationCode", value);
if let Some(value) = cond.network_code {
filter.insert("networkCode", value);
if let Some(value) = cond.network_addr {
filter.insert("networkAddr", value);
/// Transforms model options to the options.
fn build_find_options<'a, T>(opts: &ListOptions, mut find: Find<'a, T>) -> Find<'a, T>
where
T: Send + Sync,
if let Some(offset) = opts.offset {
find = find.skip(offset);
if let Some(limit) = opts.limit {
if limit > 0 {
find = find.limit(limit as i64);
if let Some(sort_list) = opts.sort.as_ref() {
if sort_list.len() > 0 {
let mut sort_opts = Document::new();
for cond in sort_list.iter() {
let key = match cond.key {
SortKey::CreatedAt => "createdAt",
SortKey::ModifiedAt => "modifiedAt",
SortKey::ApplicationCode => "applicationCode",
SortKey::NetworkCode => "networkCode",
SortKey::NetworkAddr => "networkAddr",
if cond.asc {
sort_opts.insert(key.to_string(), 1);
} else {
sort_opts.insert(key.to_string(), -1);
find = find.sort(sort_opts);
find
fn get_update_query_filter(cond: &UpdateQueryCond) -> Document {
doc! {"deviceId": cond.device_id}
/// Transforms the model object to the MongoDB document.
fn get_update_doc(updates: &Updates) -> Option<Document> {
let mut count = 0;
let mut document = Document::new();
if let Some(value) = updates.modified_at.as_ref() {
document.insert(
"modifiedAt",
DateTime::from_millis(value.timestamp_millis()),
);
if let Some(value) = updates.profile {
document.insert("profile", value);
if count == 0 {
return None;
Some(doc! {"$set": document})