From fbcfe4636e7f0c1be207c55157cf44630c21bc9a Mon Sep 17 00:00:00 2001 From: Sergey Boyko Date: Fri, 10 Feb 2023 11:48:51 +0100 Subject: [PATCH 1/6] Save dev state --- .../src/indexed_db/drivers/cursor/cursor.rs | 194 ++++++-- .../drivers/cursor/multi_key_bound_cursor.rs | 34 +- .../drivers/cursor/multi_key_cursor.rs | 20 +- .../drivers/cursor/single_key_bound_cursor.rs | 12 - .../drivers/cursor/single_key_cursor.rs | 17 +- .../mm2_db/src/indexed_db/indexed_cursor.rs | 445 +++--------------- mm2src/mm2_db/src/indexed_db/indexed_db.rs | 34 +- 7 files changed, 241 insertions(+), 515 deletions(-) diff --git a/mm2src/mm2_db/src/indexed_db/drivers/cursor/cursor.rs b/mm2src/mm2_db/src/indexed_db/drivers/cursor/cursor.rs index 694fec6f16..40b697ccf1 100644 --- a/mm2src/mm2_db/src/indexed_db/drivers/cursor/cursor.rs +++ b/mm2src/mm2_db/src/indexed_db/drivers/cursor/cursor.rs @@ -81,6 +81,12 @@ pub enum CursorBoundValue { BigUint(BeBigUint), } +#[derive(Default)] +pub struct CursorFilters { + pub(crate) only_keys: Vec<(String, Json)>, + pub(crate) bound_keys: Vec<(String, CursorBoundValue, CursorBoundValue)>, +} + impl From for CursorBoundValue { fn from(uint: u32) -> Self { CursorBoundValue::Uint(uint) } } @@ -159,6 +165,7 @@ impl CursorBoundValue { } } +/// TODO rename to `CursorAction`. #[derive(Debug, PartialEq)] pub enum CollectCursorAction { Continue, @@ -166,12 +173,116 @@ pub enum CollectCursorAction { Stop, } +/// TODO rename to `IterItemAction`. #[derive(Debug, PartialEq)] pub enum CollectItemAction { Include, Skip, } +pub trait CursorDriverImpl: Sized { + fn key_range(&self) -> CursorResult>; + + fn on_iteration(&mut self, key: JsValue) -> CursorResult<(CollectItemAction, CollectCursorAction)>; +} + +pub(crate) struct CursorDriver { + /// An actual cursor implementation. + inner: IdbCursorEnum, + db_index: IdbIndex, + cursor_request: IdbRequest, + cursor_item_rx: mpsc::Receiver>, + finished: bool, + /// We need to hold the closures in memory till `cursor` exists. + _onsuccess_closure: Closure, + _onerror_closure: Closure, +} + +impl CursorDriver { + pub async fn init_cursor(db_index: IdbIndex, filters: CursorFilters) -> CursorResult { + let inner = IdbCursorEnum::new(filters); + + let cursor_request_result = match inner.key_range()? { + Some(key_range) => db_index.open_cursor_with_range(&key_range), + None => db_index.open_cursor(), + }; + let cursor_request = cursor_request_result.map_err(|e| CursorError::ErrorOpeningCursor { + description: stringify_js_error(&e), + })?; + + let (cursor_item_tx, cursor_item_rx) = mpsc::channel(1); + + let onsuccess_closure = construct_event_closure(Ok, cursor_item_tx.clone()); + let onerror_closure = construct_event_closure(Err, cursor_item_tx); + + cursor_request.set_onsuccess(Some(onsuccess_closure.as_ref().unchecked_ref())); + cursor_request.set_onerror(Some(onerror_closure.as_ref().unchecked_ref())); + + Ok(CursorDriver { + inner: IdbCursorEnum::new(filters), + db_index, + cursor_request, + cursor_item_rx, + _onsuccess_closure: onsuccess_closure, + _onerror_closure: onerror_closure, + }) + } + + pub(crate) async fn next(&mut self) -> CursorResult> { + while Some(event) = self.cursor_item_rx.next().await { + let _cursor_event = event.map_to_mm(|e| CursorError::ErrorOpeningCursor { + description: stringify_js_error(&e), + })?; + + let cursor = match cursor_from_request(&self.cursor_request)? { + Some(cursor) => cursor, + // No more items. + None => { + self.finished = true; + return Ok(None); + }, + }; + + let (key, js_value) = match (cursor.key(), cursor.value()) { + (Ok(key), Ok(js_value)) => (key, js_value), + // No more items. + _ => { + self.finished = true; + return Ok(None); + }, + }; + + let item: InternalItem = + deserialize_from_js(js_value).map_to_mm(|e| CursorError::ErrorDeserializingItem(e.to_string()))?; + + let (item_action, cursor_action) = self.on_collect_iter(key, &item.item)?; + + let to_return = match item_action { + CollectItemAction::Include => Some(item.into_pair()), + CollectItemAction::Skip => None, + }; + + match cursor_action { + CollectCursorAction::Continue => cursor.continue_().map_to_mm(|e| CursorError::AdvanceError { + description: stringify_js_error(&e), + })?, + CollectCursorAction::ContinueWithValue(next_value) => { + cursor + .continue_with_key(&next_value) + .map_to_mm(|e| CursorError::AdvanceError { + description: stringify_js_error(&e), + })? + }, + // don't advance the cursor, just stop the loop + CollectCursorAction::Stop => None, + } + } + + // No more items. + Ok(None) + } +} + #[async_trait(?Send)] pub trait CursorOps: Sized { fn db_index(&self) -> &IdbIndex; @@ -248,55 +359,54 @@ pub trait CursorOps: Sized { } } -pub struct IdbCursorBuilder { - db_index: IdbIndex, +pub(crate) enum IdbCursorEnum { + SingleKey(IdbSingleKeyCursor), + SingleKeyBound(IdbSingleKeyBoundCursor), + MultiKey(IdbMultiKeyCursor), + MultiKeyBound(IdbMultiKeyBoundCursor), } -impl IdbCursorBuilder { - pub fn new(db_index: IdbIndex) -> IdbCursorBuilder { IdbCursorBuilder { db_index } } - - /// Returns a cursor that is a representation of a range that includes records - /// whose value of the `field_name` field equals to the `field_value` value. - pub fn single_key_cursor( - self, - field_name: String, - field_value: Json, - collect_filter: Option, - ) -> IdbSingleKeyCursor { - IdbSingleKeyCursor::new(self.db_index, field_name, field_value, collect_filter) - } +impl IdbCursorEnum { + fn new(cursor_filters: CursorFilters) -> IdbCursorEnum { + if cursor_filters.only_keys.len() > 1 && cursor_filters.bound_keys.is_empty() { + return IdbCursorEnum::MultiKey(IdbMultiKeyCursor::new(cursor_filters.only_keys)); + } + if cursor_filters.only_keys.len() + cursor_filters.bound_keys.len() > 1 { + return IdbCursorEnum::MultiKeyBound(IdbMultiKeyBoundCursor::new( + cursor_filters.only_keys, + cursor_filters.bound_keys, + )); + } // otherwise we're sure that there is either one `only`, or one `bound`, or no constraint specified. + + if let Some((field_name, field_value)) = cursor_filters.only_keys.into_iter().next() { + return IdbCursorEnum::SingleKey(IdbSingleKeyCursor::new(field_name, field_value)); + } - /// Returns a cursor that is a representation of a range that includes records - /// whose value of the `field_name` field is lower than `lower_bound` and greater than `upper_bound`. - pub fn single_key_bound_cursor( - self, - field_name: String, - lower_bound: CursorBoundValue, - upper_bound: CursorBoundValue, - collect_filter: Option, - ) -> CursorResult { - IdbSingleKeyBoundCursor::new(self.db_index, field_name, lower_bound, upper_bound, collect_filter) + if let Some((field_name, lower_bound, upper_bound)) = cursor_filters.bound_keys.into_iter().next() { + IdbCursorEnum::SingleKeyBound(IdbSingleKeyBoundCursor::new(field_name, lower_bound, upper_bound)) + } + + todo!() } +} - /// Returns a cursor that is a representation of a range that includes records - /// whose fields have only the specified values `only_values`. - pub fn multi_key_cursor( - self, - only_values: Vec<(String, Json)>, - collect_filter: Option, - ) -> CursorResult { - IdbMultiKeyCursor::new(self.db_index, only_values, collect_filter) +impl CursorDriverImpl for IdbCursorEnum { + fn key_range(&self) -> CursorResult> { + match self { + IdbCursorEnum::SingleKey(single) => single.key_range(), + IdbCursorEnum::SingleKeyBound(single_bound) => single_bound.key_range(), + IdbCursorEnum::MultiKey(multi) => multi.key_range(), + IdbCursorEnum::MultiKeyBound(multi_bound) => multi_bound.key_range(), + } } - /// Returns a cursor that is a representation of a range that includes records - /// with the multiple `only` and `bound` restrictions. - pub fn multi_key_bound_cursor( - self, - only_values: Vec<(String, Json)>, - bound_values: Vec<(String, CursorBoundValue, CursorBoundValue)>, - collect_filter: Option, - ) -> CursorResult { - IdbMultiKeyBoundCursor::new(self.db_index, only_values, bound_values, collect_filter) + fn on_iteration(&mut self, key: JsValue) -> CursorResult<(CollectItemAction, CollectCursorAction)> { + match self { + IdbCursorEnum::SingleKey(single) => single.on_iteration(key), + IdbCursorEnum::SingleKeyBound(single_bound) => single_bound.on_iteration(key), + IdbCursorEnum::MultiKey(multi) => multi.on_iteration(key), + IdbCursorEnum::MultiKeyBound(multi_bound) => multi_bound.on_iteration(key), + } } } diff --git a/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_bound_cursor.rs b/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_bound_cursor.rs index 836204e94b..c6a6ae73b2 100644 --- a/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_bound_cursor.rs +++ b/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_bound_cursor.rs @@ -12,26 +12,19 @@ use web_sys::{IdbIndex, IdbKeyRange}; /// with the multiple `only` and `bound` restrictions. /// https://developer.mozilla.org/en-US/docs/Web/API/IDBKeyRange/bound pub struct IdbMultiKeyBoundCursor { - db_index: IdbIndex, only_values: Vec<(String, Json)>, bound_values: Vec<(String, CursorBoundValue, CursorBoundValue)>, - /// An additional predicate that may be used to filter records. - collect_filter: Option, } impl IdbMultiKeyBoundCursor { pub(super) fn new( - db_index: IdbIndex, only_values: Vec<(String, Json)>, bound_values: Vec<(String, CursorBoundValue, CursorBoundValue)>, - collect_filter: Option, ) -> CursorResult { Self::check_bounds(&only_values, &bound_values)?; Ok(IdbMultiKeyBoundCursor { - db_index, only_values, bound_values, - collect_filter, }) } @@ -85,8 +78,6 @@ impl IdbMultiKeyBoundCursor { #[async_trait(?Send)] impl CursorOps for IdbMultiKeyBoundCursor { - fn db_index(&self) -> &IdbIndex { &self.db_index } - fn key_range(&self) -> CursorResult> { let lower = Array::new(); let upper = Array::new(); @@ -209,11 +200,6 @@ impl CursorOps for IdbMultiKeyBoundCursor { idx_in_index += 1; idx_in_bounds += 1; } - - // `index_key` is in our expected bounds - if let Some(ref mut filter) = self.collect_filter { - return Ok(filter(value)); - } Ok((CollectItemAction::Include, CollectCursorAction::Continue)) } } @@ -311,10 +297,7 @@ mod tests { ), ]; - // Use [`wasm_bindgen::JsCast::unchecked_from_js`] to create not-valid `IdbIndex` - // that is not used by [`IdbMultiKeyBoundCursor::on_collect_iter`] anyway. - let db_index = IdbIndex::unchecked_from_js(JsValue::NULL); - let mut cursor = IdbMultiKeyBoundCursor::new(db_index, only_values, bound_values, None).unwrap(); + let mut cursor = IdbMultiKeyBoundCursor::new(only_values, bound_values).unwrap(); ////////////////// @@ -405,10 +388,7 @@ mod tests { ), ]; - // Use [`wasm_bindgen::JsCast::unchecked_from_js`] to create not-valid `IdbIndex` - // that is not used by [`IdbMultiKeyBoundCursor::on_collect_iter`] anyway. - let db_index = IdbIndex::unchecked_from_js(JsValue::NULL); - let mut cursor = IdbMultiKeyBoundCursor::new(db_index, only_values, bound_values, None).unwrap(); + let mut cursor = IdbMultiKeyBoundCursor::new(only_values, bound_values).unwrap(); ////////////////// @@ -463,10 +443,7 @@ mod tests { CursorBoundValue::Uint(5), // upper bound )]; - // Use [`wasm_bindgen::JsCast::unchecked_from_js`] to create not-valid `IdbIndex` - // that is not used by [`IdbMultiKeyBoundCursor::on_collect_iter`] anyway. - let db_index = IdbIndex::unchecked_from_js(JsValue::NULL); - let mut cursor = IdbMultiKeyBoundCursor::new(db_index, only_values, bound_values, None).unwrap(); + let mut cursor = IdbMultiKeyBoundCursor::new(only_values, bound_values).unwrap(); ////////////////// @@ -512,10 +489,7 @@ mod tests { CursorBoundValue::Uint(5), // upper bound )]; - // Use [`wasm_bindgen::JsCast::unchecked_from_js`] to create not-valid `IdbIndex` - // that is not used by [`IdbMultiKeyBoundCursor::on_collect_iter`] anyway. - let db_index = IdbIndex::unchecked_from_js(JsValue::NULL); - let mut cursor = IdbMultiKeyBoundCursor::new(db_index, only_values, bound_values, None).unwrap(); + let mut cursor = IdbMultiKeyBoundCursor::new(only_values, bound_values).unwrap(); ////////////////// diff --git a/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_cursor.rs b/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_cursor.rs index c161ada34e..0ca00b3687 100644 --- a/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_cursor.rs +++ b/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_cursor.rs @@ -11,24 +11,13 @@ use web_sys::{IdbIndex, IdbKeyRange}; /// whose fields have only the specified [`IdbSingleCursor::only_values`] values. /// https://developer.mozilla.org/en-US/docs/Web/API/IDBKeyRange/only pub struct IdbMultiKeyCursor { - db_index: IdbIndex, only_values: Vec<(String, Json)>, - /// An additional predicate that can be used to filter records. - collect_filter: Option, } impl IdbMultiKeyCursor { - pub(super) fn new( - db_index: IdbIndex, - only_values: Vec<(String, Json)>, - collect_filter: Option, - ) -> CursorResult { + pub(super) fn new(only_values: Vec<(String, Json)>) -> CursorResult { Self::check_only_values(&only_values)?; - Ok(IdbMultiKeyCursor { - db_index, - only_values, - collect_filter, - }) + Ok(IdbMultiKeyCursor { only_values }) } fn check_only_values(only_values: &Vec<(String, Json)>) -> CursorResult<()> { @@ -45,8 +34,6 @@ impl IdbMultiKeyCursor { #[async_trait(?Send)] impl CursorOps for IdbMultiKeyCursor { - fn db_index(&self) -> &IdbIndex { &self.db_index } - fn key_range(&self) -> CursorResult> { let only = Array::new(); @@ -70,9 +57,6 @@ impl CursorOps for IdbMultiKeyCursor { _key: JsValue, value: &Json, ) -> CursorResult<(CollectItemAction, CollectCursorAction)> { - if let Some(ref mut filter) = self.collect_filter { - return Ok(filter(value)); - } Ok((CollectItemAction::Include, CollectCursorAction::Continue)) } } diff --git a/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_bound_cursor.rs b/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_bound_cursor.rs index d3138f09e9..b457092dec 100644 --- a/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_bound_cursor.rs +++ b/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_bound_cursor.rs @@ -11,30 +11,23 @@ use web_sys::{IdbIndex, IdbKeyRange}; /// and greater than [`IdbSingleBoundCursor::upper_bound_value`]. /// https://developer.mozilla.org/en-US/docs/Web/API/IDBKeyRange/bound pub struct IdbSingleKeyBoundCursor { - db_index: IdbIndex, #[allow(dead_code)] field_name: String, lower_bound: CursorBoundValue, upper_bound: CursorBoundValue, - /// An additional predicate that may be used to filter records. - collect_filter: Option, } impl IdbSingleKeyBoundCursor { pub(super) fn new( - db_index: IdbIndex, field_name: String, lower_bound: CursorBoundValue, upper_bound: CursorBoundValue, - collect_filter: Option, ) -> CursorResult { Self::check_bounds(&lower_bound, &upper_bound)?; Ok(IdbSingleKeyBoundCursor { - db_index, field_name, lower_bound, upper_bound, - collect_filter, }) } } @@ -58,8 +51,6 @@ impl IdbSingleKeyBoundCursor { #[async_trait(?Send)] impl CursorOps for IdbSingleKeyBoundCursor { - fn db_index(&self) -> &IdbIndex { &self.db_index } - fn key_range(&self) -> CursorResult> { let key_range = IdbKeyRange::bound(&self.lower_bound.to_js_value()?, &self.upper_bound.to_js_value()?) .map_to_mm(|e| CursorError::InvalidKeyRange { @@ -73,9 +64,6 @@ impl CursorOps for IdbSingleKeyBoundCursor { _key: JsValue, value: &Json, ) -> CursorResult<(CollectItemAction, CollectCursorAction)> { - if let Some(ref mut filter) = self.collect_filter { - return Ok(filter(value)); - } Ok((CollectItemAction::Include, CollectCursorAction::Continue)) } } diff --git a/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_cursor.rs b/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_cursor.rs index f24f0da1f8..24f533e56f 100644 --- a/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_cursor.rs +++ b/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_cursor.rs @@ -10,37 +10,25 @@ use web_sys::{IdbIndex, IdbKeyRange}; /// whose value of the [`IdbSingleKeyCursor::field_name`] field equals to the [`IdbSingleKeyCursor::field_value`] value. /// https://developer.mozilla.org/en-US/docs/Web/API/IDBKeyRange/only pub struct IdbSingleKeyCursor { - db_index: IdbIndex, #[allow(dead_code)] field_name: String, field_value: Json, - /// An additional predicate that may be used to filter records. - collect_filter: Option, } impl IdbSingleKeyCursor { - pub(super) fn new( - db_index: IdbIndex, - field_name: String, - field_value: Json, - filter: Option, - ) -> IdbSingleKeyCursor { + pub(super) fn new(field_name: String, field_value: Json, filter: Option) -> IdbSingleKeyCursor { if filter.is_none() { warn!("Consider using 'IdbObjectStoreImpl::get_items' instead of 'IdbSingleKeyCursor'"); } IdbSingleKeyCursor { - db_index, field_name, field_value, - collect_filter: None, } } } #[async_trait(?Send)] impl CursorOps for IdbSingleKeyCursor { - fn db_index(&self) -> &IdbIndex { &self.db_index } - fn key_range(&self) -> CursorResult> { let js_value = serialize_to_js(&self.field_value).map_to_mm(|e| CursorError::ErrorSerializingIndexFieldValue { @@ -60,9 +48,6 @@ impl CursorOps for IdbSingleKeyCursor { _key: JsValue, value: &Json, ) -> CursorResult<(CollectItemAction, CollectCursorAction)> { - if let Some(ref mut filter) = self.collect_filter { - return Ok(filter(value)); - } Ok((CollectItemAction::Include, CollectCursorAction::Continue)) } } diff --git a/mm2src/mm2_db/src/indexed_db/indexed_cursor.rs b/mm2src/mm2_db/src/indexed_db/indexed_cursor.rs index e04c622ddf..14e55c36e2 100644 --- a/mm2src/mm2_db/src/indexed_db/indexed_cursor.rs +++ b/mm2src/mm2_db/src/indexed_db/indexed_cursor.rs @@ -51,8 +51,8 @@ use crate::indexed_db::db_driver::cursor::{CollectCursorAction, CollectItemAction, CursorBoundValue, CursorOps, DbFilter, IdbCursorBuilder}; -pub use crate::indexed_db::db_driver::cursor::{CursorError, CursorResult}; -use crate::indexed_db::{ItemId, TableSignature}; +pub use crate::indexed_db::db_driver::cursor::{CursorError, CursorResult,CursorFilters}; +use crate::indexed_db::{DbTable, ItemId, TableSignature}; use async_trait::async_trait; use futures::channel::{mpsc, oneshot}; use futures::StreamExt; @@ -65,33 +65,66 @@ use std::marker::PhantomData; pub(super) type DbCursorEventTx = mpsc::UnboundedSender; pub(super) type DbCursorEventRx = mpsc::UnboundedReceiver; -pub enum DbCursorEvent { - Collect { - options: CursorCollectOptions, - result_tx: oneshot::Sender>>, - }, +pub struct CursorBuilder<'a, Table> { + index: String, + db_table: &'a DbTable<'a, Table>, + filters: CursorFilters, } -pub enum CursorCollectOptions { - SingleKey { - field_name: String, - field_value: Json, - filter: Option, - }, - SingleKeyBound { - field_name: String, - lower_bound: CursorBoundValue, - upper_bound: CursorBoundValue, - filter: Option, - }, - MultiKey { - keys: Vec<(String, Json)>, - filter: Option, - }, - MultiKeyBound { - only_keys: Vec<(String, Json)>, - bound_keys: Vec<(String, CursorBoundValue, CursorBoundValue)>, - filter: Option, +impl<'a, Table: TableSignature> CursorBuilder<'a, Table> { + pub(crate) fn new(index: &str, db_table: &'a DbTable<'a, Table>) -> Self { + CursorBuilder { + index: index.to_string(), + db_table, + filters: CursorFilters::default(), + } + } + + pub fn only(&mut self, field_name: &str, field_value: Value) -> CursorResult<()> + where + Value: Serialize + fmt::Debug, + { + let field_value_str = format!("{:?}", field_value); + let field_value = json::to_value(field_value).map_to_mm(|e| CursorError::ErrorSerializingIndexFieldValue { + field: field_name.to_owned(), + value: field_value_str, + description: e.to_string(), + })?; + + self.filters.only_keys.push((field_name.to_owned(), field_value)); + Ok(()) + } + + pub fn bound(&mut self, field_name: &str, lower_bound: Value, upper_bound: Value) + where + CursorBoundValue: From, + { + let lower_bound = CursorBoundValue::from(lower_bound); + let upper_bound = CursorBoundValue::from(upper_bound); + self.filters + .bound_keys + .push((field_name.to_owned(), lower_bound, upper_bound)); + } + + /// Opens a cursor by the specified `index`. + /// https://developer.mozilla.org/en-US/docs/Web/API/IDBObjectStore/openCursor + pub async fn open_cursor(self) -> CursorResult> { + let event_tx = self.db_table.open_cursor(&self.index, self.filters).await?; + Ok(CursorIter { + event_tx, + phantom: PhantomData::default(), + }) + } +} + +pub struct CursorIter<'a, Table> { + event_tx: DbCursorEventTx, + phantom: PhantomData<&'a Table>, +} + +pub enum DbCursorEvent { + NextItem { + result_tx: oneshot::Sender>>, }, } @@ -157,364 +190,6 @@ async fn on_collect_cursor_event( result_tx.send(result).ok(); } -pub trait WithOnly: Sized { - type ResultCursor; - - fn only(self, field_name: &str, field_value: Value) -> CursorResult - where - Value: Serialize + fmt::Debug, - { - let field_value_str = format!("{:?}", field_value); - let field_value = json::to_value(field_value).map_to_mm(|e| CursorError::ErrorSerializingIndexFieldValue { - field: field_name.to_owned(), - value: field_value_str, - description: e.to_string(), - })?; - Ok(self.only_json(field_name, field_value)) - } - - fn only_json(self, field_name: &str, field_value: Json) -> Self::ResultCursor; -} - -pub trait WithBound: Sized { - type ResultCursor; - - fn bound(self, field_name: &str, lower_bound: Value, upper_bound: Value) -> Self::ResultCursor - where - CursorBoundValue: From, - { - let lower_bound = CursorBoundValue::from(lower_bound); - let upper_bound = CursorBoundValue::from(upper_bound); - self.bound_values(field_name, lower_bound, upper_bound) - } - - fn bound_values( - self, - field_name: &str, - lower_bound: CursorBoundValue, - upper_bound: CursorBoundValue, - ) -> Self::ResultCursor; -} - -pub trait WithFilter: Sized { - fn filter(self, filter: F) -> Self - where - F: FnMut(&Json) -> (CollectItemAction, CollectCursorAction) + Send + 'static, - { - let filter = Box::new(filter); - self.filter_boxed(filter) - } - - fn filter_boxed(self, filter: DbFilter) -> Self; -} - -#[async_trait] -pub trait CollectCursor { - async fn collect(self) -> CursorResult>; -} - -#[async_trait] -impl + Send> CollectCursor for T { - async fn collect(self) -> CursorResult> { self.collect_impl().await } -} - -#[async_trait] -pub(crate) trait CollectCursorImpl: Sized { - fn into_collect_options(self) -> CursorCollectOptions; - - fn event_tx(&self) -> DbCursorEventTx; - - async fn collect_impl(self) -> CursorResult> { - let event_tx = self.event_tx(); - let options = self.into_collect_options(); - - let (result_tx, result_rx) = oneshot::channel(); - let event = DbCursorEvent::Collect { result_tx, options }; - let items: Vec<(ItemId, Json)> = send_event_recv_response(&event_tx, event, result_rx).await?; - - items - .into_iter() - .map(|(item_id, item)| json::from_value(item).map(|item| (item_id, item))) - .map(|res| res.map_to_mm(|e| CursorError::ErrorDeserializingItem(e.to_string()))) - // Item = CursorResult<(ItemId, Table)> - .collect() - } -} - -pub struct DbEmptyCursor<'a, Table: TableSignature> { - event_tx: DbCursorEventTx, - filter: Option, - phantom: PhantomData<&'a Table>, -} - -impl<'a, Table: TableSignature> WithOnly for DbEmptyCursor<'a, Table> { - type ResultCursor = DbSingleKeyCursor<'a, Table>; - - fn only_json(self, field_name: &str, field_value: Json) -> Self::ResultCursor { - DbSingleKeyCursor { - event_tx: self.event_tx, - field_name: field_name.to_owned(), - field_value, - filter: self.filter, - phantom: PhantomData::default(), - } - } -} - -impl<'a, Table: TableSignature> WithBound for DbEmptyCursor<'a, Table> { - type ResultCursor = DbSingleKeyBoundCursor<'a, Table>; - - fn bound_values( - self, - field_name: &str, - lower_bound: CursorBoundValue, - upper_bound: CursorBoundValue, - ) -> Self::ResultCursor { - DbSingleKeyBoundCursor { - event_tx: self.event_tx, - field_name: field_name.to_owned(), - lower_bound, - upper_bound, - filter: self.filter, - phantom: PhantomData::default(), - } - } -} - -impl<'a, Table: TableSignature> WithFilter for DbEmptyCursor<'a, Table> { - fn filter_boxed(mut self, filter: DbFilter) -> Self { - self.filter = Some(filter); - self - } -} - -impl<'a, Table: TableSignature> DbEmptyCursor<'a, Table> { - pub(super) fn new(event_tx: DbCursorEventTx) -> DbEmptyCursor<'a, Table> { - DbEmptyCursor { - event_tx, - filter: None, - phantom: PhantomData::default(), - } - } -} - -pub struct DbSingleKeyCursor<'a, Table: TableSignature> { - event_tx: DbCursorEventTx, - field_name: String, - field_value: Json, - filter: Option, - phantom: PhantomData<&'a Table>, -} - -impl<'a, Table: TableSignature> WithOnly for DbSingleKeyCursor<'a, Table> { - type ResultCursor = DbMultiKeyCursor<'a, Table>; - - fn only_json(self, field_name: &str, field_value: Json) -> Self::ResultCursor { - let keys = vec![ - (self.field_name, self.field_value), - (field_name.to_owned(), field_value), - ]; - DbMultiKeyCursor { - event_tx: self.event_tx, - keys, - filter: self.filter, - phantom: PhantomData::default(), - } - } -} - -impl<'a, Table: TableSignature> WithBound for DbSingleKeyCursor<'a, Table> { - type ResultCursor = DbMultiKeyBoundCursor<'a, Table>; - - fn bound_values( - self, - field_name: &str, - lower_bound: CursorBoundValue, - upper_bound: CursorBoundValue, - ) -> Self::ResultCursor { - let only_keys = vec![(self.field_name, self.field_value)]; - let bound_keys = vec![(field_name.to_owned(), lower_bound, upper_bound)]; - DbMultiKeyBoundCursor { - event_tx: self.event_tx, - only_keys, - bound_keys, - filter: self.filter, - phantom: PhantomData::default(), - } - } -} - -impl<'a, Table: TableSignature> WithFilter for DbSingleKeyCursor<'a, Table> { - fn filter_boxed(mut self, filter: DbFilter) -> Self { - self.filter = Some(filter); - self - } -} - -#[async_trait] -impl<'a, Table: TableSignature> CollectCursorImpl
for DbSingleKeyCursor<'a, Table> { - fn into_collect_options(self) -> CursorCollectOptions { - CursorCollectOptions::SingleKey { - field_name: self.field_name, - field_value: self.field_value, - filter: self.filter, - } - } - - fn event_tx(&self) -> DbCursorEventTx { self.event_tx.clone() } -} - -/// `DbSingleKeyBoundCursor` doesn't implement `WithOnly` trait, because indexes MUST start with `only` values. -pub struct DbSingleKeyBoundCursor<'a, Table: TableSignature> { - event_tx: DbCursorEventTx, - field_name: String, - lower_bound: CursorBoundValue, - upper_bound: CursorBoundValue, - filter: Option, - phantom: PhantomData<&'a Table>, -} - -impl<'a, Table: TableSignature> WithBound for DbSingleKeyBoundCursor<'a, Table> { - type ResultCursor = DbMultiKeyBoundCursor<'a, Table>; - - fn bound_values( - self, - field_name: &str, - lower_bound: CursorBoundValue, - upper_bound: CursorBoundValue, - ) -> Self::ResultCursor { - let bound_keys = vec![ - (self.field_name, self.lower_bound, self.upper_bound), - (field_name.to_owned(), lower_bound, upper_bound), - ]; - DbMultiKeyBoundCursor { - event_tx: self.event_tx, - only_keys: Vec::new(), - bound_keys, - filter: self.filter, - phantom: PhantomData::default(), - } - } -} - -impl<'a, Table: TableSignature> WithFilter for DbSingleKeyBoundCursor<'a, Table> { - fn filter_boxed(mut self, filter: DbFilter) -> Self { - self.filter = Some(filter); - self - } -} - -#[async_trait] -impl<'a, Table: TableSignature> CollectCursorImpl
for DbSingleKeyBoundCursor<'a, Table> { - fn into_collect_options(self) -> CursorCollectOptions { - CursorCollectOptions::SingleKeyBound { - field_name: self.field_name, - lower_bound: self.lower_bound, - upper_bound: self.upper_bound, - filter: self.filter, - } - } - - fn event_tx(&self) -> DbCursorEventTx { self.event_tx.clone() } -} - -pub struct DbMultiKeyCursor<'a, Table: TableSignature> { - event_tx: DbCursorEventTx, - keys: Vec<(String, Json)>, - filter: Option, - phantom: PhantomData<&'a Table>, -} - -impl<'a, Table: TableSignature> WithOnly for DbMultiKeyCursor<'a, Table> { - type ResultCursor = DbMultiKeyCursor<'a, Table>; - - fn only_json(mut self, field_name: &str, field_value: Json) -> Self::ResultCursor { - self.keys.push((field_name.to_owned(), field_value)); - self - } -} - -impl<'a, Table: TableSignature> WithBound for DbMultiKeyCursor<'a, Table> { - type ResultCursor = DbMultiKeyBoundCursor<'a, Table>; - - fn bound_values( - self, - field_name: &str, - lower_bound: CursorBoundValue, - upper_bound: CursorBoundValue, - ) -> Self::ResultCursor { - DbMultiKeyBoundCursor { - event_tx: self.event_tx, - only_keys: self.keys, - bound_keys: vec![(field_name.to_owned(), lower_bound, upper_bound)], - filter: self.filter, - phantom: PhantomData::default(), - } - } -} - -impl<'a, Table: TableSignature> WithFilter for DbMultiKeyCursor<'a, Table> { - fn filter_boxed(mut self, filter: DbFilter) -> Self { - self.filter = Some(filter); - self - } -} - -#[async_trait] -impl<'a, Table: TableSignature> CollectCursorImpl
for DbMultiKeyCursor<'a, Table> { - fn into_collect_options(self) -> CursorCollectOptions { - CursorCollectOptions::MultiKey { - keys: self.keys, - filter: self.filter, - } - } - - fn event_tx(&self) -> DbCursorEventTx { self.event_tx.clone() } -} - -/// `DbMultiKeyBoundCursor` doesn't implement `WithOnly` trait, because indexes MUST start with `only` values. -pub struct DbMultiKeyBoundCursor<'a, Table: TableSignature> { - event_tx: DbCursorEventTx, - only_keys: Vec<(String, Json)>, - bound_keys: Vec<(String, CursorBoundValue, CursorBoundValue)>, - filter: Option, - phantom: PhantomData<&'a Table>, -} - -impl<'a, Table: TableSignature> WithBound for DbMultiKeyBoundCursor<'a, Table> { - type ResultCursor = DbMultiKeyBoundCursor<'a, Table>; - - fn bound_values( - mut self, - field_name: &str, - lower_bound: CursorBoundValue, - upper_bound: CursorBoundValue, - ) -> Self::ResultCursor { - self.bound_keys.push((field_name.to_owned(), lower_bound, upper_bound)); - self - } -} - -impl<'a, Table: TableSignature> WithFilter for DbMultiKeyBoundCursor<'a, Table> { - fn filter_boxed(mut self, filter: DbFilter) -> Self { - self.filter = Some(filter); - self - } -} - -#[async_trait] -impl<'a, Table: TableSignature> CollectCursorImpl
for DbMultiKeyBoundCursor<'a, Table> { - fn into_collect_options(self) -> CursorCollectOptions { - CursorCollectOptions::MultiKeyBound { - only_keys: self.only_keys, - bound_keys: self.bound_keys, - filter: self.filter, - } - } - - fn event_tx(&self) -> DbCursorEventTx { self.event_tx.clone() } -} - async fn send_event_recv_response( event_tx: &mpsc::UnboundedSender, event: Event, diff --git a/mm2src/mm2_db/src/indexed_db/indexed_db.rs b/mm2src/mm2_db/src/indexed_db/indexed_db.rs index 897161464f..f5ad0a3dcd 100644 --- a/mm2src/mm2_db/src/indexed_db/indexed_db.rs +++ b/mm2src/mm2_db/src/indexed_db/indexed_db.rs @@ -48,7 +48,7 @@ pub use db_driver::{DbTransactionError, DbTransactionResult, DbUpgrader, InitDbE pub use db_lock::{ConstructibleDb, DbLocked, SharedDb, WeakDb}; use db_driver::{IdbDatabaseBuilder, IdbDatabaseImpl, IdbObjectStoreImpl, IdbTransactionImpl, OnUpgradeNeededCb}; -use indexed_cursor::{cursor_event_loop, DbCursorEventTx, DbEmptyCursor}; +use indexed_cursor::{cursor_event_loop, CursorBuilder, CursorFilters, DbCursorEventTx}; type DbEventTx = mpsc::UnboundedSender; type DbTransactionEventTx = mpsc::UnboundedSender; @@ -307,7 +307,7 @@ pub enum AddOrIgnoreResult { ExistAlready(ItemId), } -impl DbTable<'_, Table> { +impl<'a, Table: TableSignature> DbTable<'a, Table> { /// Adds the given item to the table. /// https://developer.mozilla.org/en-US/docs/Web/API/IDBObjectStore/add pub async fn add_item(&self, item: &Table) -> DbTransactionResult { @@ -636,23 +636,27 @@ impl DbTable<'_, Table> { send_event_recv_response(&self.event_tx, event, result_rx).await } + /// TODO add a comment. + pub fn cursor_builder(&self, index: &str) -> CursorBuilder> { CursorBuilder::new(index, self) } + + /// Whether the transaction is aborted. + pub async fn aborted(&self) -> DbTransactionResult { + let (result_tx, result_rx) = oneshot::channel(); + let event = internal::DbTableEvent::IsAborted { result_tx }; + send_event_recv_response(&self.event_tx, event, result_rx).await + } + /// Opens a cursor by the specified `index`. /// https://developer.mozilla.org/en-US/docs/Web/API/IDBObjectStore/openCursor - pub async fn open_cursor(&self, index: &str) -> DbTransactionResult> { + async fn open_cursor(&self, index: &str, filters: CursorFilters) -> DbTransactionResult { let (result_tx, result_rx) = oneshot::channel(); let event = internal::DbTableEvent::OpenCursor { index: index.to_owned(), + filters, result_tx, }; let cursor_event_tx = send_event_recv_response(&self.event_tx, event, result_rx).await?; - Ok(DbEmptyCursor::new(cursor_event_tx)) - } - - /// Whether the transaction is aborted. - pub async fn aborted(&self) -> DbTransactionResult { - let (result_tx, result_rx) = oneshot::channel(); - let event = internal::DbTableEvent::IsAborted { result_tx }; - send_event_recv_response(&self.event_tx, event, result_rx).await + Ok(cursor_event_tx) } fn deserialize_items(items: Vec<(ItemId, Json)>) -> DbTransactionResult> { @@ -726,7 +730,11 @@ async fn table_event_loop(mut rx: mpsc::UnboundedReceiver { result_tx.send(Ok(table.aborted())).ok(); }, - internal::DbTableEvent::OpenCursor { index, result_tx } => { + internal::DbTableEvent::OpenCursor { + index, + filters, + result_tx, + } => { open_cursor(&table, index, result_tx); }, } @@ -761,6 +769,7 @@ impl MultiIndex { fn open_cursor( table: &IdbObjectStoreImpl, index: String, + filters: CursorFilters, result_tx: oneshot::Sender>, ) { let cursor_builder = match table.cursor_builder(&index) { @@ -843,6 +852,7 @@ mod internal { }, OpenCursor { index: String, + filters: CursorFilters, result_tx: oneshot::Sender>, }, } From c7792c294e5359bef4e77a69e76e94480ebf91ca Mon Sep 17 00:00:00 2001 From: Sergey Boyko Date: Thu, 23 Feb 2023 19:07:45 +0100 Subject: [PATCH 2/6] Refactor IndexedDB Cursor * Add `CursorIter::next` * Add `IdbEmptyCursor` --- Cargo.lock | 1 + mm2src/mm2_db/Cargo.toml | 1 + .../src/indexed_db/drivers/cursor/cursor.rs | 173 ++++----- .../indexed_db/drivers/cursor/empty_cursor.rs | 14 + .../drivers/cursor/multi_key_bound_cursor.rs | 35 +- .../drivers/cursor/multi_key_cursor.rs | 30 +- .../drivers/cursor/single_key_bound_cursor.rs | 15 +- .../drivers/cursor/single_key_cursor.rs | 21 +- .../src/indexed_db/drivers/object_store.rs | 35 +- .../src/indexed_db/drivers/transaction.rs | 4 +- .../mm2_db/src/indexed_db/indexed_cursor.rs | 339 +++++++++++------- mm2src/mm2_db/src/indexed_db/indexed_db.rs | 69 ++-- .../mm2_main/src/lp_swap/my_swaps_storage.rs | 18 +- 13 files changed, 395 insertions(+), 360 deletions(-) create mode 100644 mm2src/mm2_db/src/indexed_db/drivers/cursor/empty_cursor.rs diff --git a/Cargo.lock b/Cargo.lock index d82d83b56c..efc9c71e0f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4333,6 +4333,7 @@ dependencies = [ "async-trait", "common", "derive_more", + "enum_from", "futures 0.3.15", "hex 0.4.2", "itertools", diff --git a/mm2src/mm2_db/Cargo.toml b/mm2src/mm2_db/Cargo.toml index 5c07fcd407..152bbaec65 100644 --- a/mm2src/mm2_db/Cargo.toml +++ b/mm2src/mm2_db/Cargo.toml @@ -10,6 +10,7 @@ doctest = false async-trait = "0.1" common = { path = "../common" } derive_more = "0.99" +enum_from = { path = "../derives/enum_from" } futures = { version = "0.3", package = "futures", features = ["compat", "async-await", "thread-pool"] } itertools = "0.10" hex = "0.4.2" diff --git a/mm2src/mm2_db/src/indexed_db/drivers/cursor/cursor.rs b/mm2src/mm2_db/src/indexed_db/drivers/cursor/cursor.rs index 40b697ccf1..b748b6939d 100644 --- a/mm2src/mm2_db/src/indexed_db/drivers/cursor/cursor.rs +++ b/mm2src/mm2_db/src/indexed_db/drivers/cursor/cursor.rs @@ -1,9 +1,9 @@ use super::construct_event_closure; use crate::indexed_db::db_driver::{InternalItem, ItemId}; use crate::indexed_db::BeBigUint; -use async_trait::async_trait; use common::wasm::{deserialize_from_js, serialize_to_js, stringify_js_error}; use derive_more::Display; +use enum_from::EnumFromTrait; use futures::channel::mpsc; use futures::StreamExt; use js_sys::Array; @@ -14,20 +14,21 @@ use wasm_bindgen::prelude::*; use wasm_bindgen::JsCast; use web_sys::{IdbCursorWithValue, IdbIndex, IdbKeyRange, IdbRequest}; +mod empty_cursor; mod multi_key_bound_cursor; mod multi_key_cursor; mod single_key_bound_cursor; mod single_key_cursor; -pub use multi_key_bound_cursor::IdbMultiKeyBoundCursor; -pub use multi_key_cursor::IdbMultiKeyCursor; -pub use single_key_bound_cursor::IdbSingleKeyBoundCursor; -pub use single_key_cursor::IdbSingleKeyCursor; +use empty_cursor::IdbEmptyCursor; +use multi_key_bound_cursor::IdbMultiKeyBoundCursor; +use multi_key_cursor::IdbMultiKeyCursor; +use single_key_bound_cursor::IdbSingleKeyBoundCursor; +use single_key_cursor::IdbSingleKeyCursor; pub type CursorResult = Result>; -pub type DbFilter = Box (CollectItemAction, CollectCursorAction) + Send>; -#[derive(Debug, Display, PartialEq)] +#[derive(Debug, Display, EnumFromTrait, PartialEq)] pub enum CursorError { #[display( fmt = "Error serializing the '{}' value of the index field '{}' : {:?}", @@ -59,6 +60,7 @@ pub enum CursorError { )] IncorrectNumberOfKeysPerIndex { expected: usize, found: usize }, #[display(fmt = "Error occurred due to an unexpected state: {:?}", _0)] + #[from_trait(WithInternal::internal)] UnexpectedState(String), #[display(fmt = "Incorrect usage of the cursor: {:?}", description)] IncorrectUsage { description: String }, @@ -189,18 +191,18 @@ pub trait CursorDriverImpl: Sized { pub(crate) struct CursorDriver { /// An actual cursor implementation. inner: IdbCursorEnum, - db_index: IdbIndex, cursor_request: IdbRequest, cursor_item_rx: mpsc::Receiver>, - finished: bool, + /// Whether we got `CollectCursorAction::Stop` at the last iteration or not. + stopped: bool, /// We need to hold the closures in memory till `cursor` exists. - _onsuccess_closure: Closure, - _onerror_closure: Closure, + _onsuccess_closure: Closure, + _onerror_closure: Closure, } impl CursorDriver { - pub async fn init_cursor(db_index: IdbIndex, filters: CursorFilters) -> CursorResult { - let inner = IdbCursorEnum::new(filters); + pub(crate) fn init_cursor(db_index: IdbIndex, filters: CursorFilters) -> CursorResult { + let inner = IdbCursorEnum::new(filters)?; let cursor_request_result = match inner.key_range()? { Some(key_range) => db_index.open_cursor_with_range(&key_range), @@ -219,17 +221,30 @@ impl CursorDriver { cursor_request.set_onerror(Some(onerror_closure.as_ref().unchecked_ref())); Ok(CursorDriver { - inner: IdbCursorEnum::new(filters), - db_index, + inner, cursor_request, cursor_item_rx, + stopped: false, _onsuccess_closure: onsuccess_closure, _onerror_closure: onerror_closure, }) } pub(crate) async fn next(&mut self) -> CursorResult> { - while Some(event) = self.cursor_item_rx.next().await { + loop { + // Check if we got `CollectCursorAction::Stop` at the last iteration. + if self.stopped { + return Ok(None); + } + + let event = match self.cursor_item_rx.next().await { + Some(event) => event, + None => { + self.stopped = true; + return Ok(None); + }, + }; + let _cursor_event = event.map_to_mm(|e| CursorError::ErrorOpeningCursor { description: stringify_js_error(&e), })?; @@ -238,7 +253,7 @@ impl CursorDriver { Some(cursor) => cursor, // No more items. None => { - self.finished = true; + self.stopped = true; return Ok(None); }, }; @@ -247,7 +262,7 @@ impl CursorDriver { (Ok(key), Ok(js_value)) => (key, js_value), // No more items. _ => { - self.finished = true; + self.stopped = true; return Ok(None); }, }; @@ -255,12 +270,7 @@ impl CursorDriver { let item: InternalItem = deserialize_from_js(js_value).map_to_mm(|e| CursorError::ErrorDeserializingItem(e.to_string()))?; - let (item_action, cursor_action) = self.on_collect_iter(key, &item.item)?; - - let to_return = match item_action { - CollectItemAction::Include => Some(item.into_pair()), - CollectItemAction::Skip => None, - }; + let (item_action, cursor_action) = self.inner.on_iteration(key)?; match cursor_action { CollectCursorAction::Continue => cursor.continue_().map_to_mm(|e| CursorError::AdvanceError { @@ -273,93 +283,24 @@ impl CursorDriver { description: stringify_js_error(&e), })? }, - // don't advance the cursor, just stop the loop - CollectCursorAction::Stop => None, + // Don't advance the cursor. + // Here we set the `stopped` flag so we return `Ok(None)` at the next iteration immediately. + // This is required because `item_action` can be `CollectItemAction::Include`, + // and at this iteration we will return `Ok(Some)`. + CollectCursorAction::Stop => self.stopped = true, } - } - - // No more items. - Ok(None) - } -} - -#[async_trait(?Send)] -pub trait CursorOps: Sized { - fn db_index(&self) -> &IdbIndex; - - fn key_range(&self) -> CursorResult>; - - fn on_collect_iter(&mut self, key: JsValue, value: &Json) - -> CursorResult<(CollectItemAction, CollectCursorAction)>; - - /// Collect items that match the specified bounds. - async fn collect(mut self) -> CursorResult> { - let (tx, mut rx) = mpsc::channel(1); - - let db_index = self.db_index(); - let cursor_request_result = match self.key_range()? { - Some(key_range) => db_index.open_cursor_with_range(&key_range), - None => db_index.open_cursor(), - }; - let cursor_request = cursor_request_result.map_err(|e| CursorError::ErrorOpeningCursor { - description: stringify_js_error(&e), - })?; - - let onsuccess_closure = construct_event_closure(Ok, tx.clone()); - let onerror_closure = construct_event_closure(Err, tx); - - cursor_request.set_onsuccess(Some(onsuccess_closure.as_ref().unchecked_ref())); - cursor_request.set_onerror(Some(onerror_closure.as_ref().unchecked_ref())); - - let mut collected_items = Vec::new(); - while let Some(event) = rx.next().await { - let _cursor_event = event.map_to_mm(|e| CursorError::ErrorOpeningCursor { - description: stringify_js_error(&e), - })?; - - let cursor = match cursor_from_request(&cursor_request)? { - Some(cursor) => cursor, - // no more items, stop the loop - None => break, - }; - - let (key, js_value) = match (cursor.key(), cursor.value()) { - (Ok(key), Ok(js_value)) => (key, js_value), - // no more items, stop the loop - _ => break, - }; - - let item: InternalItem = - deserialize_from_js(js_value).map_to_mm(|e| CursorError::ErrorDeserializingItem(e.to_string()))?; - - let (item_action, cursor_action) = self.on_collect_iter(key, &item.item)?; match item_action { - CollectItemAction::Include => collected_items.push(item.into_pair()), + CollectItemAction::Include => return Ok(Some(item.into_pair())), + // Try to fetch the next item. CollectItemAction::Skip => (), } - - match cursor_action { - CollectCursorAction::Continue => cursor.continue_().map_to_mm(|e| CursorError::AdvanceError { - description: stringify_js_error(&e), - })?, - CollectCursorAction::ContinueWithValue(next_value) => { - cursor - .continue_with_key(&next_value) - .map_to_mm(|e| CursorError::AdvanceError { - description: stringify_js_error(&e), - })? - }, - // don't advance the cursor, just stop the loop - CollectCursorAction::Stop => break, - } } - - Ok(collected_items) } } pub(crate) enum IdbCursorEnum { + Empty(IdbEmptyCursor), SingleKey(IdbSingleKeyCursor), SingleKeyBound(IdbSingleKeyBoundCursor), MultiKey(IdbMultiKeyCursor), @@ -367,32 +308,45 @@ pub(crate) enum IdbCursorEnum { } impl IdbCursorEnum { - fn new(cursor_filters: CursorFilters) -> IdbCursorEnum { + fn new(cursor_filters: CursorFilters) -> CursorResult { if cursor_filters.only_keys.len() > 1 && cursor_filters.bound_keys.is_empty() { - return IdbCursorEnum::MultiKey(IdbMultiKeyCursor::new(cursor_filters.only_keys)); + return Ok(IdbCursorEnum::MultiKey(IdbMultiKeyCursor::new( + cursor_filters.only_keys, + ))); } - if cursor_filters.only_keys.len() + cursor_filters.bound_keys.len() > 1 { - return IdbCursorEnum::MultiKeyBound(IdbMultiKeyBoundCursor::new( + if !cursor_filters.bound_keys.is_empty() + && (cursor_filters.only_keys.len() + cursor_filters.bound_keys.len() > 1) + { + return Ok(IdbCursorEnum::MultiKeyBound(IdbMultiKeyBoundCursor::new( cursor_filters.only_keys, cursor_filters.bound_keys, - )); + )?)); } // otherwise we're sure that there is either one `only`, or one `bound`, or no constraint specified. if let Some((field_name, field_value)) = cursor_filters.only_keys.into_iter().next() { - return IdbCursorEnum::SingleKey(IdbSingleKeyCursor::new(field_name, field_value)); + return Ok(IdbCursorEnum::SingleKey(IdbSingleKeyCursor::new( + field_name, + field_value, + ))); } if let Some((field_name, lower_bound, upper_bound)) = cursor_filters.bound_keys.into_iter().next() { - IdbCursorEnum::SingleKeyBound(IdbSingleKeyBoundCursor::new(field_name, lower_bound, upper_bound)) + return Ok(IdbCursorEnum::SingleKeyBound(IdbSingleKeyBoundCursor::new( + field_name, + lower_bound, + upper_bound, + )?)); } - todo!() + // There are no constraint specified. + Ok(IdbCursorEnum::Empty(IdbEmptyCursor)) } } impl CursorDriverImpl for IdbCursorEnum { fn key_range(&self) -> CursorResult> { match self { + IdbCursorEnum::Empty(empty) => empty.key_range(), IdbCursorEnum::SingleKey(single) => single.key_range(), IdbCursorEnum::SingleKeyBound(single_bound) => single_bound.key_range(), IdbCursorEnum::MultiKey(multi) => multi.key_range(), @@ -402,6 +356,7 @@ impl CursorDriverImpl for IdbCursorEnum { fn on_iteration(&mut self, key: JsValue) -> CursorResult<(CollectItemAction, CollectCursorAction)> { match self { + IdbCursorEnum::Empty(empty) => empty.on_iteration(key), IdbCursorEnum::SingleKey(single) => single.on_iteration(key), IdbCursorEnum::SingleKeyBound(single_bound) => single_bound.on_iteration(key), IdbCursorEnum::MultiKey(multi) => multi.on_iteration(key), diff --git a/mm2src/mm2_db/src/indexed_db/drivers/cursor/empty_cursor.rs b/mm2src/mm2_db/src/indexed_db/drivers/cursor/empty_cursor.rs new file mode 100644 index 0000000000..ca0e9d213a --- /dev/null +++ b/mm2src/mm2_db/src/indexed_db/drivers/cursor/empty_cursor.rs @@ -0,0 +1,14 @@ +use super::{CollectCursorAction, CollectItemAction, CursorDriverImpl, CursorResult}; +use wasm_bindgen::prelude::*; +use web_sys::IdbKeyRange; + +/// The representation of a range that includes all records. +pub struct IdbEmptyCursor; + +impl CursorDriverImpl for IdbEmptyCursor { + fn key_range(&self) -> CursorResult> { Ok(None) } + + fn on_iteration(&mut self, _key: JsValue) -> CursorResult<(CollectItemAction, CollectCursorAction)> { + Ok((CollectItemAction::Include, CollectCursorAction::Continue)) + } +} diff --git a/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_bound_cursor.rs b/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_bound_cursor.rs index c6a6ae73b2..170b3c516f 100644 --- a/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_bound_cursor.rs +++ b/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_bound_cursor.rs @@ -1,12 +1,11 @@ -use super::{index_key_as_array, CollectCursorAction, CollectItemAction, CursorBoundValue, CursorError, CursorOps, - CursorResult, DbFilter}; -use async_trait::async_trait; +use super::{index_key_as_array, CollectCursorAction, CollectItemAction, CursorBoundValue, CursorDriverImpl, + CursorError, CursorResult}; use common::{deserialize_from_js, serialize_to_js, stringify_js_error}; use js_sys::Array; use mm2_err_handle::prelude::*; use serde_json::{json, Value as Json}; use wasm_bindgen::prelude::*; -use web_sys::{IdbIndex, IdbKeyRange}; +use web_sys::IdbKeyRange; /// The representation of a range that includes records /// with the multiple `only` and `bound` restrictions. @@ -76,8 +75,7 @@ impl IdbMultiKeyBoundCursor { } } -#[async_trait(?Send)] -impl CursorOps for IdbMultiKeyBoundCursor { +impl CursorDriverImpl for IdbMultiKeyBoundCursor { fn key_range(&self) -> CursorResult> { let lower = Array::new(); let upper = Array::new(); @@ -110,11 +108,7 @@ impl CursorOps for IdbMultiKeyBoundCursor { /// The range `IDBKeyRange.bound([2,2], [4,4])` includes values like `[3,0]` and `[3,5]` as `[2,2] < [3,0] < [3,5] < [4,4]`, /// so we need to do additional filtering. /// For more information on why it's required, see https://stackoverflow.com/a/32976384. - fn on_collect_iter( - &mut self, - index_key: JsValue, - value: &Json, - ) -> CursorResult<(CollectItemAction, CollectCursorAction)> { + fn on_iteration(&mut self, index_key: JsValue) -> CursorResult<(CollectItemAction, CollectCursorAction)> { let index_keys_js_array = index_key_as_array(index_key)?; let index_keys: Vec = index_keys_js_array .iter() @@ -207,7 +201,6 @@ impl CursorOps for IdbMultiKeyBoundCursor { mod tests { use super::*; use common::log::wasm_log::register_wasm_log; - use wasm_bindgen::JsCast; use wasm_bindgen_test::*; wasm_bindgen_test_configure!(run_in_browser); @@ -216,7 +209,7 @@ mod tests { fn test_in_bound_indexes(cursor: &mut IdbMultiKeyBoundCursor, input_indexes: Vec) { for input_index in input_indexes { let input_index_js_value = serialize_to_js(&input_index).unwrap(); - let result = cursor.on_collect_iter(input_index_js_value, &Json::Null); + let result = cursor.on_iteration(input_index_js_value); assert_eq!( result, Ok((CollectItemAction::Include, CollectCursorAction::Continue)), @@ -233,7 +226,7 @@ mod tests { for (input_index, expected_next) in input_indexes { let input_index_js_value = serialize_to_js(&input_index).unwrap(); let (item_action, cursor_action) = cursor - .on_collect_iter(input_index_js_value, &Json::Null) + .on_iteration(input_index_js_value) .expect(&format!("Error due to the index '{:?}'", input_index)); let actual_next: Json = match cursor_action { @@ -254,7 +247,7 @@ mod tests { fn test_out_of_bound_indexes(cursor: &mut IdbMultiKeyBoundCursor, input_indexes: Vec) { for input_index in input_indexes { let input_index_js_value = serialize_to_js(&input_index).unwrap(); - let result = cursor.on_collect_iter(input_index_js_value, &Json::Null); + let result = cursor.on_iteration(input_index_js_value); assert_eq!( result, Ok((CollectItemAction::Skip, CollectCursorAction::Stop)), @@ -266,7 +259,7 @@ mod tests { /// This test doesn't check [`IdbMultiKeyBoundCursor::filter`]. #[wasm_bindgen_test] - fn test_on_collect_iter_multiple_only_and_bound_values() { + fn test_on_iteration_multiple_only_and_bound_values() { register_wasm_log(); let only_values = vec![ @@ -371,7 +364,7 @@ mod tests { /// This test doesn't check [`IdbMultiKeyBoundCursor::filter`]. #[wasm_bindgen_test] - fn test_on_collect_iter_multiple_bound_values() { + fn test_on_iteration_multiple_bound_values() { register_wasm_log(); let only_values = Vec::new(); @@ -433,7 +426,7 @@ mod tests { /// This test doesn't check [`IdbMultiKeyBoundCursor::filter`]. #[wasm_bindgen_test] - fn test_on_collect_iter_single_only_and_bound_values() { + fn test_on_iteration_single_only_and_bound_values() { register_wasm_log(); let only_values = vec![("field1".to_owned(), json!(2))]; @@ -479,7 +472,7 @@ mod tests { } #[wasm_bindgen_test] - fn test_on_collect_iter_error() { + fn test_on_iteration_error() { register_wasm_log(); let only_values = vec![("field1".to_owned(), json!(2u32))]; @@ -505,7 +498,7 @@ mod tests { for (input_index, expected_type) in input_indexes { let input_index_js_value = serialize_to_js(&input_index).unwrap(); let error = cursor - .on_collect_iter(input_index_js_value, &Json::Null) + .on_iteration(input_index_js_value) .expect_err(&format!("'{:?}' must lead to 'CursorError::TypeMismatch'", input_index)); match error.into_inner() { CursorError::TypeMismatch { expected, .. } => assert_eq!(expected, expected_type), @@ -523,7 +516,7 @@ mod tests { for input_index in input_indexes { let input_index_js_value = serialize_to_js(&input_index).unwrap(); let error = cursor - .on_collect_iter(input_index_js_value, &Json::Null) + .on_iteration(input_index_js_value) .expect_err(&format!("'{:?}' must lead to 'CursorError::TypeMismatch'", input_index)); match error.into_inner() { CursorError::IncorrectNumberOfKeysPerIndex { .. } => (), diff --git a/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_cursor.rs b/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_cursor.rs index 0ca00b3687..ddd2cd8fa0 100644 --- a/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_cursor.rs +++ b/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_cursor.rs @@ -1,11 +1,10 @@ -use super::{CollectCursorAction, CollectItemAction, CursorError, CursorOps, CursorResult, DbFilter}; -use async_trait::async_trait; +use super::{CollectCursorAction, CollectItemAction, CursorDriverImpl, CursorError, CursorResult}; use common::{serialize_to_js, stringify_js_error}; use js_sys::Array; use mm2_err_handle::prelude::*; use serde_json::Value as Json; use wasm_bindgen::prelude::*; -use web_sys::{IdbIndex, IdbKeyRange}; +use web_sys::IdbKeyRange; /// The representation of a range that includes records /// whose fields have only the specified [`IdbSingleCursor::only_values`] values. @@ -15,25 +14,10 @@ pub struct IdbMultiKeyCursor { } impl IdbMultiKeyCursor { - pub(super) fn new(only_values: Vec<(String, Json)>) -> CursorResult { - Self::check_only_values(&only_values)?; - Ok(IdbMultiKeyCursor { only_values }) - } - - fn check_only_values(only_values: &Vec<(String, Json)>) -> CursorResult<()> { - if only_values.len() < 2 { - let description = format!( - "Incorrect usage of 'IdbMultiKeyCursor': expected more than one cursor bound, found '{}'", - only_values.len(), - ); - return MmError::err(CursorError::IncorrectUsage { description }); - } - Ok(()) - } + pub(super) fn new(only_values: Vec<(String, Json)>) -> IdbMultiKeyCursor { IdbMultiKeyCursor { only_values } } } -#[async_trait(?Send)] -impl CursorOps for IdbMultiKeyCursor { +impl CursorDriverImpl for IdbMultiKeyCursor { fn key_range(&self) -> CursorResult> { let only = Array::new(); @@ -52,11 +36,7 @@ impl CursorOps for IdbMultiKeyCursor { Ok(Some(key_range)) } - fn on_collect_iter( - &mut self, - _key: JsValue, - value: &Json, - ) -> CursorResult<(CollectItemAction, CollectCursorAction)> { + fn on_iteration(&mut self, _key: JsValue) -> CursorResult<(CollectItemAction, CollectCursorAction)> { Ok((CollectItemAction::Include, CollectCursorAction::Continue)) } } diff --git a/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_bound_cursor.rs b/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_bound_cursor.rs index b457092dec..d868984526 100644 --- a/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_bound_cursor.rs +++ b/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_bound_cursor.rs @@ -1,10 +1,8 @@ -use super::{CollectCursorAction, CollectItemAction, CursorBoundValue, CursorError, CursorOps, CursorResult, DbFilter}; -use async_trait::async_trait; +use super::{CollectCursorAction, CollectItemAction, CursorBoundValue, CursorDriverImpl, CursorError, CursorResult}; use common::{log::warn, stringify_js_error}; use mm2_err_handle::prelude::*; -use serde_json::Value as Json; use wasm_bindgen::prelude::*; -use web_sys::{IdbIndex, IdbKeyRange}; +use web_sys::IdbKeyRange; /// The representation of a range that includes records /// whose value of the [`IdbSingleBoundCursor::field_name`] field is lower than [`IdbSingleBoundCursor::lower_bound_value`] @@ -49,8 +47,7 @@ impl IdbSingleKeyBoundCursor { } } -#[async_trait(?Send)] -impl CursorOps for IdbSingleKeyBoundCursor { +impl CursorDriverImpl for IdbSingleKeyBoundCursor { fn key_range(&self) -> CursorResult> { let key_range = IdbKeyRange::bound(&self.lower_bound.to_js_value()?, &self.upper_bound.to_js_value()?) .map_to_mm(|e| CursorError::InvalidKeyRange { @@ -59,11 +56,7 @@ impl CursorOps for IdbSingleKeyBoundCursor { Ok(Some(key_range)) } - fn on_collect_iter( - &mut self, - _key: JsValue, - value: &Json, - ) -> CursorResult<(CollectItemAction, CollectCursorAction)> { + fn on_iteration(&mut self, _key: JsValue) -> CursorResult<(CollectItemAction, CollectCursorAction)> { Ok((CollectItemAction::Include, CollectCursorAction::Continue)) } } diff --git a/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_cursor.rs b/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_cursor.rs index 24f533e56f..6416317943 100644 --- a/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_cursor.rs +++ b/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_cursor.rs @@ -1,10 +1,9 @@ -use super::{CollectCursorAction, CollectItemAction, CursorError, CursorOps, CursorResult, DbFilter}; -use async_trait::async_trait; -use common::{log::warn, serialize_to_js, stringify_js_error}; +use super::{CollectCursorAction, CollectItemAction, CursorDriverImpl, CursorError, CursorResult}; +use common::{serialize_to_js, stringify_js_error}; use mm2_err_handle::prelude::*; use serde_json::Value as Json; use wasm_bindgen::prelude::*; -use web_sys::{IdbIndex, IdbKeyRange}; +use web_sys::IdbKeyRange; /// The representation of a range that includes records /// whose value of the [`IdbSingleKeyCursor::field_name`] field equals to the [`IdbSingleKeyCursor::field_value`] value. @@ -16,10 +15,7 @@ pub struct IdbSingleKeyCursor { } impl IdbSingleKeyCursor { - pub(super) fn new(field_name: String, field_value: Json, filter: Option) -> IdbSingleKeyCursor { - if filter.is_none() { - warn!("Consider using 'IdbObjectStoreImpl::get_items' instead of 'IdbSingleKeyCursor'"); - } + pub(super) fn new(field_name: String, field_value: Json) -> IdbSingleKeyCursor { IdbSingleKeyCursor { field_name, field_value, @@ -27,8 +23,7 @@ impl IdbSingleKeyCursor { } } -#[async_trait(?Send)] -impl CursorOps for IdbSingleKeyCursor { +impl CursorDriverImpl for IdbSingleKeyCursor { fn key_range(&self) -> CursorResult> { let js_value = serialize_to_js(&self.field_value).map_to_mm(|e| CursorError::ErrorSerializingIndexFieldValue { @@ -43,11 +38,7 @@ impl CursorOps for IdbSingleKeyCursor { Ok(Some(key_range)) } - fn on_collect_iter( - &mut self, - _key: JsValue, - value: &Json, - ) -> CursorResult<(CollectItemAction, CollectCursorAction)> { + fn on_iteration(&mut self, _key: JsValue) -> CursorResult<(CollectItemAction, CollectCursorAction)> { Ok((CollectItemAction::Include, CollectCursorAction::Continue)) } } diff --git a/mm2src/mm2_db/src/indexed_db/drivers/object_store.rs b/mm2src/mm2_db/src/indexed_db/drivers/object_store.rs index 53c9be6104..a7b8081e9c 100644 --- a/mm2src/mm2_db/src/indexed_db/drivers/object_store.rs +++ b/mm2src/mm2_db/src/indexed_db/drivers/object_store.rs @@ -1,5 +1,4 @@ use super::{construct_event_closure, DbTransactionError, DbTransactionResult, InternalItem, ItemId}; -use crate::indexed_db::db_driver::cursor::IdbCursorBuilder; use common::{deserialize_from_js, serialize_to_js, stringify_js_error}; use futures::channel::mpsc; use futures::StreamExt; @@ -9,7 +8,7 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use wasm_bindgen::prelude::*; use wasm_bindgen::JsCast; -use web_sys::{IdbObjectStore, IdbRequest}; +use web_sys::{IdbIndex, IdbObjectStore, IdbRequest}; pub struct IdbObjectStoreImpl { pub(crate) object_store: IdbObjectStore, @@ -58,10 +57,7 @@ impl IdbObjectStoreImpl { let index = index_str.to_owned(); let index_value_js = try_serialize_index_value!(serialize_to_js(&index_value), index_str); - let db_index = match self.object_store.index(index_str) { - Ok(index) => index, - Err(_) => return MmError::err(DbTransactionError::NoSuchIndex { index }), - }; + let db_index = self.open_index(index_str)?; let get_request = match db_index.get_all_with_key(&index_value_js) { Ok(request) => request, Err(e) => { @@ -90,10 +86,7 @@ impl IdbObjectStoreImpl { let index = index_str.to_owned(); let index_value_js = try_serialize_index_value!(serialize_to_js(&index_value), index); - let db_index = match self.object_store.index(index_str) { - Ok(index) => index, - Err(_) => return MmError::err(DbTransactionError::NoSuchIndex { index }), - }; + let db_index = self.open_index(index_str)?; let get_request = match db_index.get_all_keys_with_key(&index_value_js) { Ok(request) => request, Err(e) => { @@ -141,10 +134,7 @@ impl IdbObjectStoreImpl { let index = index_str.to_owned(); let index_value_js = try_serialize_index_value!(serialize_to_js(&index_value), index); - let db_index = match self.object_store.index(index_str) { - Ok(index) => index, - Err(_) => return MmError::err(DbTransactionError::NoSuchIndex { index }), - }; + let db_index = self.open_index(index_str)?; let count_request = match db_index.count_with_key(&index_value_js) { Ok(request) => request, Err(e) => { @@ -248,16 +238,13 @@ impl IdbObjectStoreImpl { Ok(()) } - pub(crate) fn cursor_builder(&self, index_str: &str) -> DbTransactionResult { - let db_index = match self.object_store.index(index_str) { - Ok(index) => index, - Err(_) => { - return MmError::err(DbTransactionError::NoSuchIndex { - index: index_str.to_owned(), - }) - }, - }; - Ok(IdbCursorBuilder::new(db_index)) + pub(crate) fn open_index(&self, index_str: &str) -> DbTransactionResult { + match self.object_store.index(index_str) { + Ok(index) => Ok(index), + Err(_) => MmError::err(DbTransactionError::NoSuchIndex { + index: index_str.to_owned(), + }), + } } async fn wait_for_request_complete(request: &IdbRequest) -> Result { diff --git a/mm2src/mm2_db/src/indexed_db/drivers/transaction.rs b/mm2src/mm2_db/src/indexed_db/drivers/transaction.rs index aef31a4e7d..f76794b2de 100644 --- a/mm2src/mm2_db/src/indexed_db/drivers/transaction.rs +++ b/mm2src/mm2_db/src/indexed_db/drivers/transaction.rs @@ -1,6 +1,7 @@ use super::IdbObjectStoreImpl; use common::wasm::stringify_js_error; use derive_more::Display; +use enum_from::EnumFromTrait; use mm2_err_handle::prelude::*; use serde_json::Value as Json; use std::collections::HashSet; @@ -11,7 +12,7 @@ use web_sys::IdbTransaction; pub type DbTransactionResult = Result>; -#[derive(Debug, Display, PartialEq)] +#[derive(Debug, Display, EnumFromTrait, PartialEq)] pub enum DbTransactionError { #[display(fmt = "No such table '{}'", table)] NoSuchTable { table: String }, @@ -44,6 +45,7 @@ pub enum DbTransactionError { description: String, }, #[display(fmt = "Error occurred due to an unexpected state: {:?}", _0)] + #[from_trait(WithInternal::internal)] UnexpectedState(String), #[display(fmt = "Transaction was aborted")] TransactionAborted, diff --git a/mm2src/mm2_db/src/indexed_db/indexed_cursor.rs b/mm2src/mm2_db/src/indexed_db/indexed_cursor.rs index 14e55c36e2..1c16cc69a6 100644 --- a/mm2src/mm2_db/src/indexed_db/indexed_cursor.rs +++ b/mm2src/mm2_db/src/indexed_db/indexed_cursor.rs @@ -18,10 +18,13 @@ //! you can use [`WithBound::bound`] along with [`WithOnly::only`]: //! ```rust //! let table = open_table_somehow(); -//! let all_rick_morty_swaps = table.open_cursor("search_index") +//! let all_rick_morty_swaps = table +//! .cursor_builder() //! .only("base_coin", "RICK", "MORTY")? //! .bound("base_coin_value", 10, 13) //! .bound("started_at", 1000000030.into(), u32::MAX.into()) +//! .open_cursor("search_index") +//! .await? //! .collect() //! .await?; //! ``` @@ -49,13 +52,12 @@ //! because ['RICK', 'MORTY', 13] < ['RICK', 'MORTY', 13, 1000000030], //! although it is expected to be within the specified bounds. -use crate::indexed_db::db_driver::cursor::{CollectCursorAction, CollectItemAction, CursorBoundValue, CursorOps, - DbFilter, IdbCursorBuilder}; -pub use crate::indexed_db::db_driver::cursor::{CursorError, CursorResult,CursorFilters}; +use crate::indexed_db::db_driver::cursor::CursorBoundValue; +pub(crate) use crate::indexed_db::db_driver::cursor::{CursorDriver, CursorFilters}; +pub use crate::indexed_db::db_driver::cursor::{CursorError, CursorResult}; use crate::indexed_db::{DbTable, ItemId, TableSignature}; -use async_trait::async_trait; use futures::channel::{mpsc, oneshot}; -use futures::StreamExt; +use futures::{SinkExt, StreamExt}; use mm2_err_handle::prelude::*; use serde::Serialize; use serde_json::{self as json, Value as Json}; @@ -65,22 +67,20 @@ use std::marker::PhantomData; pub(super) type DbCursorEventTx = mpsc::UnboundedSender; pub(super) type DbCursorEventRx = mpsc::UnboundedReceiver; -pub struct CursorBuilder<'a, Table> { - index: String, - db_table: &'a DbTable<'a, Table>, +pub struct CursorBuilder<'transaction, 'reference, Table: TableSignature> { + db_table: &'reference DbTable<'transaction, Table>, filters: CursorFilters, } -impl<'a, Table: TableSignature> CursorBuilder<'a, Table> { - pub(crate) fn new(index: &str, db_table: &'a DbTable<'a, Table>) -> Self { +impl<'transaction, 'reference, Table: TableSignature> CursorBuilder<'transaction, 'reference, Table> { + pub(crate) fn new(db_table: &'reference DbTable<'transaction, Table>) -> Self { CursorBuilder { - index: index.to_string(), db_table, filters: CursorFilters::default(), } } - pub fn only(&mut self, field_name: &str, field_value: Value) -> CursorResult<()> + pub fn only(mut self, field_name: &str, field_value: Value) -> CursorResult where Value: Serialize + fmt::Debug, { @@ -92,10 +92,10 @@ impl<'a, Table: TableSignature> CursorBuilder<'a, Table> { })?; self.filters.only_keys.push((field_name.to_owned(), field_value)); - Ok(()) + Ok(self) } - pub fn bound(&mut self, field_name: &str, lower_bound: Value, upper_bound: Value) + pub fn bound(mut self, field_name: &str, lower_bound: Value, upper_bound: Value) -> Self where CursorBoundValue: From, { @@ -104,12 +104,19 @@ impl<'a, Table: TableSignature> CursorBuilder<'a, Table> { self.filters .bound_keys .push((field_name.to_owned(), lower_bound, upper_bound)); + self } /// Opens a cursor by the specified `index`. /// https://developer.mozilla.org/en-US/docs/Web/API/IDBObjectStore/openCursor - pub async fn open_cursor(self) -> CursorResult> { - let event_tx = self.db_table.open_cursor(&self.index, self.filters).await?; + pub async fn open_cursor(self, index: &str) -> CursorResult> { + let event_tx = + self.db_table + .open_cursor(index, self.filters) + .await + .mm_err(|e| CursorError::ErrorOpeningCursor { + description: e.to_string(), + })?; Ok(CursorIter { event_tx, phantom: PhantomData::default(), @@ -117,97 +124,56 @@ impl<'a, Table: TableSignature> CursorBuilder<'a, Table> { } } -pub struct CursorIter<'a, Table> { +pub struct CursorIter<'transaction, Table> { event_tx: DbCursorEventTx, - phantom: PhantomData<&'a Table>, + phantom: PhantomData<&'transaction Table>, +} + +impl<'transaction, Table: TableSignature> CursorIter<'transaction, Table> { + /// Advances the iterator and returns the next value. + /// Please note that the items are sorted by the index keys. + pub async fn next(&mut self) -> CursorResult> { + let (result_tx, result_rx) = oneshot::channel(); + self.event_tx + .send(DbCursorEvent::NextItem { result_tx }) + .await + .map_to_mm(|e| CursorError::UnexpectedState(format!("Error sending cursor event: {e}")))?; + let maybe_item = result_rx + .await + .map_to_mm(|e| CursorError::UnexpectedState(format!("Error receiving cursor item: {e}")))??; + let (item_id, item) = match maybe_item { + Some((item_id, item)) => (item_id, item), + None => return Ok(None), + }; + let item = json::from_value(item).map_to_mm(|e| CursorError::ErrorDeserializingItem(e.to_string()))?; + Ok(Some((item_id, item))) + } + + pub async fn collect(mut self) -> CursorResult> { + let mut result = Vec::new(); + while let Some((item_id, item)) = self.next().await? { + result.push((item_id, item)); + } + Ok(result) + } } pub enum DbCursorEvent { NextItem { - result_tx: oneshot::Sender>>, + result_tx: oneshot::Sender>>, }, } -pub async fn cursor_event_loop(mut rx: DbCursorEventRx, cursor_builder: IdbCursorBuilder) { +pub(crate) async fn cursor_event_loop(mut rx: DbCursorEventRx, mut cursor: CursorDriver) { while let Some(event) = rx.next().await { match event { - DbCursorEvent::Collect { options, result_tx } => { - on_collect_cursor_event(result_tx, options, cursor_builder).await; - return; + DbCursorEvent::NextItem { result_tx } => { + result_tx.send(cursor.next().await).ok(); }, } } } -async fn on_collect_cursor_event( - result_tx: oneshot::Sender>>, - options: CursorCollectOptions, - cursor_builder: IdbCursorBuilder, -) { - async fn on_collect_cursor_event_impl( - options: CursorCollectOptions, - cursor_builder: IdbCursorBuilder, - ) -> CursorResult> { - match options { - CursorCollectOptions::SingleKey { - field_name, - field_value, - filter, - } => { - cursor_builder - .single_key_cursor(field_name, field_value, filter) - .collect() - .await - }, - CursorCollectOptions::SingleKeyBound { - field_name, - lower_bound, - upper_bound, - filter, - } => { - cursor_builder - .single_key_bound_cursor(field_name, lower_bound, upper_bound, filter)? - .collect() - .await - }, - CursorCollectOptions::MultiKey { keys, filter } => { - cursor_builder.multi_key_cursor(keys, filter)?.collect().await - }, - CursorCollectOptions::MultiKeyBound { - only_keys, - bound_keys, - filter, - } => { - cursor_builder - .multi_key_bound_cursor(only_keys, bound_keys, filter)? - .collect() - .await - }, - } - } - - let result = on_collect_cursor_event_impl(options, cursor_builder).await; - result_tx.send(result).ok(); -} - -async fn send_event_recv_response( - event_tx: &mpsc::UnboundedSender, - event: Event, - result_rx: oneshot::Receiver>, -) -> CursorResult { - if let Err(e) = event_tx.unbounded_send(event) { - let error = format!("Error sending event: {}", e); - return MmError::err(CursorError::UnexpectedState(error)); - } - match result_rx.await { - Ok(result) => result, - Err(e) => { - let error = format!("Error receiving result: {}", e); - MmError::err(CursorError::UnexpectedState(error)) - }, - } -} - mod tests { use super::*; use crate::indexed_db::{BeBigUint, DbIdentifier, DbTable, DbUpgrader, IndexedDbBuilder, OnUpgradeResult}; @@ -380,13 +346,14 @@ mod tests { // Get every item that satisfies the following [num_x, num_y] bound. let actual_items = table + .cursor_builder() + .bound("timestamp_x", num_x.clone(), num_y.clone()) .open_cursor("timestamp_x") .await - .expect("!DbTable::open_cursor") - .bound("timestamp_x", num_x.clone(), num_y.clone()) + .expect("!CursorBuilder::open_cursor") .collect() .await - .expect("!DbSingleKeyBoundCursor::collect") + .expect("!CursorIter::collect") .into_iter() // Map `(ItemId, TimestampTable)` into `BeBigUint`. .map(|(_item_id, item)| item.timestamp_x) @@ -434,14 +401,15 @@ mod tests { fill_table(&table, items).await; let mut actual_items = table + .cursor_builder() + .only("base_coin", "RICK") + .expect("!CursorBuilder::only") .open_cursor("base_coin") .await - .expect("!DbTable::open_cursor") - .only("base_coin", "RICK") - .expect("!DbEmptyCursor::only") + .expect("!CursorBuilder::open_cursor") .collect() .await - .expect("!DbSingleKeyCursor::collect") + .expect("!CursorIter::collect") .into_iter() .map(|(_item_id, item)| item) .collect::>(); @@ -487,13 +455,14 @@ mod tests { fill_table(&table, items).await; let mut actual_items = table + .cursor_builder() + .bound("rel_coin_value", 5u32, u32::MAX) .open_cursor("rel_coin_value") .await - .expect("!DbTable::open_cursor") - .bound("rel_coin_value", 5u32, u32::MAX) + .expect("!CursorBuilder::open_cursor") .collect() .await - .expect("!DbSingleKeyCursor::collect") + .expect("!CursorIter::collect") .into_iter() .map(|(_item_id, item)| item) .collect::>(); @@ -545,18 +514,19 @@ mod tests { fill_table(&table, items).await; let mut actual_items = table - .open_cursor("basecoin_basecoinvalue_startedat_index") - .await - .expect("!DbTable::open_cursor") + .cursor_builder() .only("base_coin", "RICK") - .expect("!DbEmptyCursor::only") + .expect("!CursorBuilder::only") .only("base_coin_value", 12) - .expect("!DbSingleKeyCursor::only") + .expect("!CursorBuilder::only") .only("started_at", 721) - .expect("!DbMultiKeyCursor::only") + .expect("!CursorBuilder::only") + .open_cursor("basecoin_basecoinvalue_startedat_index") + .await + .expect("!CursorBuilder::open_cursor") .collect() .await - .expect("!DbMultiKeyCursor::collect") + .expect("!CursorIter::collect") .into_iter() .map(|(_item_id, item)| item) .collect::>(); @@ -621,19 +591,20 @@ mod tests { fill_table(&table, items).await; let actual_items = table - .open_cursor("all_fields_index") - .await - .expect("!DbTable::open_cursor") + .cursor_builder() .only("base_coin", "RICK") - .expect("!DbEmptyCursor::only") + .expect("!CursorBuilder::only") .only("rel_coin", "QRC20") - .expect("!DbEmptyCursor::only") + .expect("!CursorBuilder::only") .bound("base_coin_value", 3u32, 8u32) .bound("rel_coin_value", 10u32, 12u32) .bound("started_at", 600i32, 800i32) + .open_cursor("all_fields_index") + .await + .expect("!CursorBuilder::open_cursor") .collect() .await - .expect("!DbMultiKeyCursor::collect") + .expect("!CursorIter::collect") .into_iter() .map(|(_item_id, item)| item) .collect::>(); @@ -681,15 +652,16 @@ mod tests { fill_table(&table, items).await; let actual_items = table - .open_cursor("timestamp_xyz") - .await - .expect("!DbTable::open_cursor") + .cursor_builder() .bound("timestamp_x", BeBigUint::from(u64::MAX - 1), BeBigUint::from(u128::MAX)) .bound("timestamp_y", 0u32, 5u32) .bound("timestamp_z", BeBigUint::from(u64::MAX), BeBigUint::from(u128::MAX - 2)) + .open_cursor("timestamp_xyz") + .await + .expect("!CursorBuilder::open_cursor") .collect() .await - .expect("!DbMultiKeyCursor::collect") + .expect("!CursorIter::collect") .into_iter() .map(|(_item_id, item)| item) .collect::>(); @@ -704,4 +676,131 @@ mod tests { assert_eq!(actual_items, expected_items); } + + #[wasm_bindgen_test] + async fn test_iter_without_constraints() { + const DB_NAME: &str = "TEST_ITER_WITHOUT_CONSTRAINTS"; + const DB_VERSION: u32 = 1; + + register_wasm_log(); + + let items = vec![ + swap_item!("uuid1", "RICK", "MORTY", 10, 3, 700), + swap_item!("uuid2", "MORTY", "KMD", 95000, 1, 721), + swap_item!("uuid3", "RICK", "XYZ", 7, u32::MAX, 1281), + swap_item!("uuid4", "RICK", "MORTY", 8, 6, 92), + ]; + + let db = IndexedDbBuilder::new(DbIdentifier::for_test(DB_NAME)) + .with_version(DB_VERSION) + .with_table::() + .build() + .await + .expect("!IndexedDb::init"); + let transaction = db.transaction().await.expect("!IndexedDb::transaction"); + let table = transaction + .table::() + .await + .expect("!DbTransaction::open_table"); + fill_table(&table, items).await; + + let mut cursor_iter = table + .cursor_builder() + .open_cursor("rel_coin_value") + .await + .expect("!CursorBuilder::open_cursor"); + + #[track_caller] + async fn next_item(cursor_iter: &mut CursorIter<'_, Table>) -> Option
{ + cursor_iter + .next() + .await + .expect("!CursorIter::next") + .map(|(_item_id, item)| item) + } + + // The items must be sorted by `rel_coin_value`. + assert_eq!( + next_item(&mut cursor_iter).await, + Some(swap_item!("uuid2", "MORTY", "KMD", 95000, 1, 721)) + ); + assert_eq!( + next_item(&mut cursor_iter).await, + Some(swap_item!("uuid1", "RICK", "MORTY", 10, 3, 700)) + ); + assert_eq!( + next_item(&mut cursor_iter).await, + Some(swap_item!("uuid4", "RICK", "MORTY", 8, 6, 92)) + ); + assert_eq!( + next_item(&mut cursor_iter).await, + Some(swap_item!("uuid3", "RICK", "XYZ", 7, u32::MAX, 1281)) + ); + assert!(next_item(&mut cursor_iter).await.is_none()); + // Try to poll one more time. This should not fail but return `None`. + assert!(next_item(&mut cursor_iter).await.is_none()); + } + + #[wasm_bindgen_test] + async fn test_iter_single_key_bound_cursor() { + const DB_NAME: &str = "TEST_ITER_SINGLE_KEY_BOUND_CURSOR"; + const DB_VERSION: u32 = 1; + + register_wasm_log(); + + let items = vec![ + swap_item!("uuid1", "RICK", "MORTY", 10, 3, 700), + swap_item!("uuid2", "MORTY", "KMD", 95000, 1, 721), + swap_item!("uuid3", "RICK", "XYZ", 7, u32::MAX, 1281), // + + swap_item!("uuid4", "RICK", "MORTY", 8, 6, 92), // + + swap_item!("uuid5", "QRC20", "RICK", 2, 4, 721), + swap_item!("uuid6", "KMD", "MORTY", 12, 3124, 214), // + + ]; + + let db = IndexedDbBuilder::new(DbIdentifier::for_test(DB_NAME)) + .with_version(DB_VERSION) + .with_table::() + .build() + .await + .expect("!IndexedDb::init"); + let transaction = db.transaction().await.expect("!IndexedDb::transaction"); + let table = transaction + .table::() + .await + .expect("!DbTransaction::open_table"); + fill_table(&table, items).await; + + let mut cursor_iter = table + .cursor_builder() + .bound("rel_coin_value", 5u32, u32::MAX) + .open_cursor("rel_coin_value") + .await + .expect("!CursorBuilder::open_cursor"); + + #[track_caller] + async fn next_item(cursor_iter: &mut CursorIter<'_, Table>) -> Option
{ + cursor_iter + .next() + .await + .expect("!CursorIter::next") + .map(|(_item_id, item)| item) + } + + // The items must be sorted by `rel_coin_value`. + assert_eq!( + next_item(&mut cursor_iter).await, + Some(swap_item!("uuid4", "RICK", "MORTY", 8, 6, 92)) + ); + assert_eq!( + next_item(&mut cursor_iter).await, + Some(swap_item!("uuid6", "KMD", "MORTY", 12, 3124, 214)) + ); + assert_eq!( + next_item(&mut cursor_iter).await, + Some(swap_item!("uuid3", "RICK", "XYZ", 7, u32::MAX, 1281)) + ); + assert!(next_item(&mut cursor_iter).await.is_none()); + // Try to poll one more time. This should not fail but return `None`. + assert!(next_item(&mut cursor_iter).await.is_none()); + } } diff --git a/mm2src/mm2_db/src/indexed_db/indexed_db.rs b/mm2src/mm2_db/src/indexed_db/indexed_db.rs index f5ad0a3dcd..97fc4d684d 100644 --- a/mm2src/mm2_db/src/indexed_db/indexed_db.rs +++ b/mm2src/mm2_db/src/indexed_db/indexed_db.rs @@ -48,15 +48,15 @@ pub use db_driver::{DbTransactionError, DbTransactionResult, DbUpgrader, InitDbE pub use db_lock::{ConstructibleDb, DbLocked, SharedDb, WeakDb}; use db_driver::{IdbDatabaseBuilder, IdbDatabaseImpl, IdbObjectStoreImpl, IdbTransactionImpl, OnUpgradeNeededCb}; -use indexed_cursor::{cursor_event_loop, CursorBuilder, CursorFilters, DbCursorEventTx}; +use indexed_cursor::{cursor_event_loop, CursorBuilder, CursorDriver, CursorError, CursorFilters, CursorResult, + DbCursorEventTx}; type DbEventTx = mpsc::UnboundedSender; type DbTransactionEventTx = mpsc::UnboundedSender; type DbTableEventTx = mpsc::UnboundedSender; pub mod cursor_prelude { - pub use crate::indexed_db::indexed_cursor::{CollectCursor, CursorError, CursorResult, WithBound, WithFilter, - WithOnly}; + pub use crate::indexed_db::indexed_cursor::{CursorError, CursorResult}; } pub trait TableSignature: DeserializeOwned + Serialize + 'static { @@ -174,21 +174,20 @@ pub struct IndexedDb { event_tx: DbEventTx, } -async fn send_event_recv_response( +async fn send_event_recv_response( event_tx: &mpsc::UnboundedSender, event: Event, - result_rx: oneshot::Receiver>, -) -> DbTransactionResult { + result_rx: oneshot::Receiver>, +) -> MmResult +where + Error: WithInternal + NotMmError, +{ if let Err(e) = event_tx.unbounded_send(event) { - let error = format!("Error sending event: {}", e); - return MmError::err(DbTransactionError::UnexpectedState(error)); + return MmError::err(Error::internal(format!("Error sending event: {}", e))); } match result_rx.await { Ok(result) => result, - Err(e) => { - let error = format!("Error receiving result: {}", e); - MmError::err(DbTransactionError::UnexpectedState(error)) - }, + Err(e) => MmError::err(Error::internal(format!("Error receiving result: {}", e))), } } @@ -232,9 +231,9 @@ impl IndexedDb { } } -pub struct DbTransaction<'a> { +pub struct DbTransaction<'transaction> { event_tx: DbTransactionEventTx, - phantom: PhantomData<&'a ()>, + phantom: PhantomData<&'transaction ()>, } impl DbTransaction<'_> { @@ -297,9 +296,9 @@ impl DbTransaction<'_> { } } -pub struct DbTable<'a, Table: TableSignature> { +pub struct DbTable<'transaction, Table: TableSignature> { event_tx: DbTableEventTx, - phantom: PhantomData<&'a Table>, + phantom: PhantomData<&'transaction Table>, } pub enum AddOrIgnoreResult { @@ -307,7 +306,7 @@ pub enum AddOrIgnoreResult { ExistAlready(ItemId), } -impl<'a, Table: TableSignature> DbTable<'a, Table> { +impl<'transaction, Table: TableSignature> DbTable<'transaction, Table> { /// Adds the given item to the table. /// https://developer.mozilla.org/en-US/docs/Web/API/IDBObjectStore/add pub async fn add_item(&self, item: &Table) -> DbTransactionResult { @@ -636,8 +635,11 @@ impl<'a, Table: TableSignature> DbTable<'a, Table> { send_event_recv_response(&self.event_tx, event, result_rx).await } - /// TODO add a comment. - pub fn cursor_builder(&self, index: &str) -> CursorBuilder> { CursorBuilder::new(index, self) } + /// Returns a `CursorBuilder` builder. It can be used to open a cursor at the specified with specific key bounds. + /// See [`CursorBuilder::open_cursor`]. + pub fn cursor_builder<'reference>(&'reference self) -> CursorBuilder<'transaction, 'reference, Table> { + CursorBuilder::new(self) + } /// Whether the transaction is aborted. pub async fn aborted(&self) -> DbTransactionResult { @@ -648,14 +650,16 @@ impl<'a, Table: TableSignature> DbTable<'a, Table> { /// Opens a cursor by the specified `index`. /// https://developer.mozilla.org/en-US/docs/Web/API/IDBObjectStore/openCursor - async fn open_cursor(&self, index: &str, filters: CursorFilters) -> DbTransactionResult { + async fn open_cursor(&self, index: &str, filters: CursorFilters) -> CursorResult { let (result_tx, result_rx) = oneshot::channel(); let event = internal::DbTableEvent::OpenCursor { index: index.to_owned(), filters, result_tx, }; - let cursor_event_tx = send_event_recv_response(&self.event_tx, event, result_rx).await?; + let cursor_event_tx = send_event_recv_response(&self.event_tx, event, result_rx) + .await + .mm_err(|e| CursorError::UnexpectedState(e.to_string()))?; Ok(cursor_event_tx) } @@ -735,7 +739,7 @@ async fn table_event_loop(mut rx: mpsc::UnboundedReceiver { - open_cursor(&table, index, result_tx); + open_cursor(&table, index, filters, result_tx); }, } } @@ -770,18 +774,29 @@ fn open_cursor( table: &IdbObjectStoreImpl, index: String, filters: CursorFilters, - result_tx: oneshot::Sender>, + result_tx: oneshot::Sender>, ) { - let cursor_builder = match table.cursor_builder(&index) { - Ok(builder) => builder, + let db_index = match table.open_index(&index) { + Ok(db_index) => db_index, + Err(tr_err) => { + let cursor_err = tr_err.map(|tr_err| CursorError::ErrorOpeningCursor { + description: tr_err.to_string(), + }); + result_tx.send(Err(cursor_err)).ok(); + return; + }, + }; + let cursor = match CursorDriver::init_cursor(db_index, filters) { + Ok(cursor) => cursor, Err(e) => { result_tx.send(Err(e)).ok(); return; }, }; + let (event_tx, event_rx) = mpsc::unbounded(); - let fut = async move { cursor_event_loop(event_rx, cursor_builder).await }; + let fut = async move { cursor_event_loop(event_rx, cursor).await }; // `cursor_event_loop` will finish almost immediately once `event_tx` is dropped. spawn_local(fut); @@ -853,7 +868,7 @@ mod internal { OpenCursor { index: String, filters: CursorFilters, - result_tx: oneshot::Sender>, + result_tx: oneshot::Sender>, }, } } diff --git a/mm2src/mm2_main/src/lp_swap/my_swaps_storage.rs b/mm2src/mm2_main/src/lp_swap/my_swaps_storage.rs index 6d2dc630bb..e38b8c5419 100644 --- a/mm2src/mm2_main/src/lp_swap/my_swaps_storage.rs +++ b/mm2src/mm2_main/src/lp_swap/my_swaps_storage.rs @@ -201,37 +201,41 @@ mod wasm_impl { let items = match (&filter.my_coin, &filter.other_coin) { (Some(my_coin), Some(other_coin)) => { my_swaps_table - .open_cursor("with_my_other_coins") - .await? + .cursor_builder() .only("my_coin", my_coin)? .only("other_coin", other_coin)? .bound("started_at", from_timestamp, to_timestamp) + .open_cursor("with_my_other_coins") + .await? .collect() .await? }, (Some(my_coin), None) => { my_swaps_table - .open_cursor("with_my_coin") - .await? + .cursor_builder() .only("my_coin", my_coin)? .bound("started_at", from_timestamp, to_timestamp) + .open_cursor("with_my_coin") + .await? .collect() .await? }, (None, Some(other_coin)) => { my_swaps_table - .open_cursor("with_other_coin") - .await? + .cursor_builder() .only("other_coin", other_coin)? .bound("started_at", from_timestamp, to_timestamp) + .open_cursor("with_other_coin") + .await? .collect() .await? }, (None, None) => { my_swaps_table + .cursor_builder() + .bound("started_at", from_timestamp, to_timestamp) .open_cursor("started_at") .await? - .bound("started_at", from_timestamp, to_timestamp) .collect() .await? }, From b4acf2c1ba28065202caae989bd744ba9ffba0c1 Mon Sep 17 00:00:00 2001 From: Sergey Boyko Date: Thu, 23 Feb 2023 19:15:15 +0100 Subject: [PATCH 3/6] Fix merge conflicts --- .../wasm/indexeddb_block_header_storage.rs | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/mm2src/coins/utxo/utxo_block_header_storage/wasm/indexeddb_block_header_storage.rs b/mm2src/coins/utxo/utxo_block_header_storage/wasm/indexeddb_block_header_storage.rs index 63d2fa9950..8d908d801b 100644 --- a/mm2src/coins/utxo/utxo_block_header_storage/wasm/indexeddb_block_header_storage.rs +++ b/mm2src/coins/utxo/utxo_block_header_storage/wasm/indexeddb_block_header_storage.rs @@ -3,7 +3,6 @@ use super::BlockHeaderStorageTable; use async_trait::async_trait; use chain::BlockHeader; use mm2_core::mm_ctx::MmArc; -use mm2_db::indexed_db::cursor_prelude::{CollectCursor, WithOnly}; use mm2_db::indexed_db::{ConstructibleDb, DbIdentifier, DbInstance, DbLocked, IndexedDb, IndexedDbBuilder, InitDbResult, MultiIndex, SharedDb}; use mm2_err_handle::prelude::*; @@ -180,11 +179,12 @@ impl BlockHeaderStorageOps for IDBBlockHeadersStorage { // Todo: use open_cursor with direction to optimze this process. let res = block_headers_db + .cursor_builder() + .only("ticker", ticker.clone()) + .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))? .open_cursor("ticker") .await .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))? - .only("ticker", ticker.clone()) - .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))? .collect() .await .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))? @@ -216,11 +216,12 @@ impl BlockHeaderStorageOps for IDBBlockHeadersStorage { // Todo: use open_cursor with direction to optimze this process. let res = block_headers_db + .cursor_builder() + .only("ticker", ticker.clone()) + .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))? .open_cursor("ticker") .await .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))? - .only("ticker", ticker.clone()) - .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))? .collect() .await .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))? From 4deacf51c40b98bbf71d0bea4e1dec99e429bcca Mon Sep 17 00:00:00 2001 From: Sergey Boyko Date: Fri, 24 Feb 2023 17:14:18 +0100 Subject: [PATCH 4/6] Fix and optimize `IndexedDbBlockHeaderStorage` * Make `height: BeBigUint` instead of `u64` * Fix `BlockHeaderStorageTable::TICKER_HEIGHT_INDEX` index --- .../utxo/utxo_block_header_storage/mod.rs | 7 +- .../wasm/block_header_table.rs | 8 +- .../wasm/indexeddb_block_header_storage.rs | 95 ++++++++------ mm2src/mm2_db/Cargo.toml | 2 +- .../src/indexed_db/drivers/cursor/cursor.rs | 14 ++- .../mm2_db/src/indexed_db/indexed_cursor.rs | 117 +++++++++++++++--- 6 files changed, 178 insertions(+), 65 deletions(-) diff --git a/mm2src/coins/utxo/utxo_block_header_storage/mod.rs b/mm2src/coins/utxo/utxo_block_header_storage/mod.rs index 1932d9afda..888d067f02 100644 --- a/mm2src/coins/utxo/utxo_block_header_storage/mod.rs +++ b/mm2src/coins/utxo/utxo_block_header_storage/mod.rs @@ -58,7 +58,6 @@ impl BlockHeaderStorage { }) } - #[cfg(test)] pub(crate) fn into_inner(self) -> Box { self.inner } } @@ -108,7 +107,7 @@ impl BlockHeaderStorageOps for BlockHeaderStorage { async fn is_table_empty(&self) -> Result<(), BlockHeaderStorageError> { self.inner.is_table_empty().await } } -#[cfg(test)] +#[cfg(any(test, target_arch = "wasm32"))] mod block_headers_storage_tests { use super::*; use chain::BlockHeaderBits; @@ -313,7 +312,7 @@ mod native_tests { fn test_remove_headers_up_to_height() { block_on(test_remove_headers_up_to_height_impl(FOR_COIN_GET)) } } -#[cfg(all(test, target_arch = "wasm32"))] +#[cfg(target_arch = "wasm32")] mod wasm_test { use super::*; use crate::utxo::utxo_block_header_storage::block_headers_storage_tests::*; @@ -323,7 +322,7 @@ mod wasm_test { wasm_bindgen_test_configure!(run_in_browser); - const FOR_COIN: &str = "RICK"; + const FOR_COIN: &str = "tBTC"; #[wasm_bindgen_test] async fn test_storage_init() { diff --git a/mm2src/coins/utxo/utxo_block_header_storage/wasm/block_header_table.rs b/mm2src/coins/utxo/utxo_block_header_storage/wasm/block_header_table.rs index 79121043d6..951d0aa938 100644 --- a/mm2src/coins/utxo/utxo_block_header_storage/wasm/block_header_table.rs +++ b/mm2src/coins/utxo/utxo_block_header_storage/wasm/block_header_table.rs @@ -1,8 +1,8 @@ -use mm2_db::indexed_db::{DbUpgrader, OnUpgradeResult, TableSignature}; +use mm2_db::indexed_db::{BeBigUint, DbUpgrader, OnUpgradeResult, TableSignature}; #[derive(Clone, Debug, Deserialize, Serialize)] pub struct BlockHeaderStorageTable { - pub height: u64, + pub height: BeBigUint, pub bits: u32, pub hash: String, pub raw_header: String, @@ -10,7 +10,7 @@ pub struct BlockHeaderStorageTable { } impl BlockHeaderStorageTable { - pub const HEIGHT_TICKER_INDEX: &str = "block_height_ticker_index"; + pub const TICKER_HEIGHT_INDEX: &str = "block_height_ticker_index"; pub const HASH_TICKER_INDEX: &str = "block_hash_ticker_index"; } @@ -21,7 +21,7 @@ impl TableSignature for BlockHeaderStorageTable { match (old_version, new_version) { (0, 1) => { let table = upgrader.create_table(Self::table_name())?; - table.create_multi_index(Self::HEIGHT_TICKER_INDEX, &["height", "ticker"], true)?; + table.create_multi_index(Self::TICKER_HEIGHT_INDEX, &["ticker", "height"], true)?; table.create_multi_index(Self::HASH_TICKER_INDEX, &["hash", "ticker"], true)?; table.create_index("ticker", false)?; }, diff --git a/mm2src/coins/utxo/utxo_block_header_storage/wasm/indexeddb_block_header_storage.rs b/mm2src/coins/utxo/utxo_block_header_storage/wasm/indexeddb_block_header_storage.rs index 8d908d801b..5b0b4b0f6c 100644 --- a/mm2src/coins/utxo/utxo_block_header_storage/wasm/indexeddb_block_header_storage.rs +++ b/mm2src/coins/utxo/utxo_block_header_storage/wasm/indexeddb_block_header_storage.rs @@ -3,9 +3,10 @@ use super::BlockHeaderStorageTable; use async_trait::async_trait; use chain::BlockHeader; use mm2_core::mm_ctx::MmArc; -use mm2_db::indexed_db::{ConstructibleDb, DbIdentifier, DbInstance, DbLocked, IndexedDb, IndexedDbBuilder, +use mm2_db::indexed_db::{BeBigUint, ConstructibleDb, DbIdentifier, DbInstance, DbLocked, IndexedDb, IndexedDbBuilder, InitDbResult, MultiIndex, SharedDb}; use mm2_err_handle::prelude::*; +use num_traits::ToPrimitive; use primitives::hash::H256; use serialization::Reader; use spv_validation::storage::{BlockHeaderStorageError, BlockHeaderStorageOps}; @@ -92,15 +93,15 @@ impl BlockHeaderStorageOps for IDBBlockHeadersStorage { let bits: u32 = header.bits.into(); let headers_to_store = BlockHeaderStorageTable { ticker: ticker.clone(), - height, + height: BeBigUint::from(height), bits, hash, raw_header, }; - let index_keys = MultiIndex::new(BlockHeaderStorageTable::HEIGHT_TICKER_INDEX) - .with_value(&height) - .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))? + let index_keys = MultiIndex::new(BlockHeaderStorageTable::TICKER_HEIGHT_INDEX) .with_value(&ticker) + .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))? + .with_value(&BeBigUint::from(height)) .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))?; block_headers_db @@ -148,10 +149,10 @@ impl BlockHeaderStorageOps for IDBBlockHeadersStorage { .table::() .await .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))?; - let index_keys = MultiIndex::new(BlockHeaderStorageTable::HEIGHT_TICKER_INDEX) - .with_value(&height) - .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))? + let index_keys = MultiIndex::new(BlockHeaderStorageTable::TICKER_HEIGHT_INDEX) .with_value(&ticker) + .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))? + .with_value(&BeBigUint::from(height)) .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))?; Ok(block_headers_db @@ -177,22 +178,31 @@ impl BlockHeaderStorageOps for IDBBlockHeadersStorage { .await .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))?; - // Todo: use open_cursor with direction to optimze this process. - let res = block_headers_db + let maybe_item = block_headers_db .cursor_builder() .only("ticker", ticker.clone()) .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))? - .open_cursor("ticker") + // We need to provide any constraint on the `height` property + // since `ticker_height` consists of both `ticker` and `height` properties. + .bound("height", BeBigUint::from(0u64), BeBigUint::from(u64::MAX)) + // Cursor returns values from the lowest to highest key indexes. + // But we need to get the most highest height, so reverse the cursor direction. + .reverse() + .open_cursor(BlockHeaderStorageTable::TICKER_HEIGHT_INDEX) .await .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))? - .collect() + .next() .await - .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))? - .into_iter() - .map(|(_item_id, item)| item.height) - .collect::>(); + .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))?; - Ok(res.into_iter().max()) + if let Some((_item_id, item)) = maybe_item { + let height = item + .height + .to_u64() + .ok_or_else(|| BlockHeaderStorageError::get_err(&ticker, "height is too large".to_string()))?; + return Ok(Some(height)); + } + Ok(None) } async fn get_last_block_header_with_non_max_bits( @@ -214,26 +224,29 @@ impl BlockHeaderStorageOps for IDBBlockHeadersStorage { .await .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))?; - // Todo: use open_cursor with direction to optimze this process. - let res = block_headers_db + let mut cursor = block_headers_db .cursor_builder() .only("ticker", ticker.clone()) .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))? - .open_cursor("ticker") + // We need to provide any constraint on the `height` property + // since `ticker_height` consists of both `ticker` and `height` properties. + .bound("height", BeBigUint::from(0u64), BeBigUint::from(u64::MAX)) + // Cursor returns values from the lowest to highest key indexes. + // But we need to get the most highest height, so reverse the cursor direction. + .reverse() + .open_cursor(BlockHeaderStorageTable::TICKER_HEIGHT_INDEX) .await - .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))? - .collect() + .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))?; + + while let Some((_item_id, header)) = cursor + .next() .await .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))? - .into_iter() - .map(|(_item_id, item)| item) - .collect::>(); - let res = res - .into_iter() - .filter_map(|e| if e.bits != max_bits { Some(e) } else { None }) - .collect::>(); - - for header in res { + { + if header.bits == max_bits { + continue; + } + let serialized = &hex::decode(header.raw_header).map_err(|e| BlockHeaderStorageError::DecodeError { coin: ticker.clone(), reason: e.to_string(), @@ -274,11 +287,19 @@ impl BlockHeaderStorageOps for IDBBlockHeadersStorage { .with_value(&ticker) .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))?; - Ok(block_headers_db + let maybe_item = block_headers_db .get_item_by_unique_multi_index(index_keys) .await - .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))? - .map(|raw| raw.1.height as i64)) + .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))?; + + if let Some((_item_id, header)) = maybe_item { + let height = header + .height + .to_i64() + .ok_or_else(|| BlockHeaderStorageError::get_err(&ticker, "height is too large".to_string()))?; + return Ok(Some(height)); + } + Ok(None) } async fn remove_headers_up_to_height(&self, to_height: u64) -> Result<(), BlockHeaderStorageError> { @@ -298,10 +319,10 @@ impl BlockHeaderStorageOps for IDBBlockHeadersStorage { .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))?; for height in 0..=to_height { - let index_keys = MultiIndex::new(BlockHeaderStorageTable::HEIGHT_TICKER_INDEX) - .with_value(&height) - .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))? + let index_keys = MultiIndex::new(BlockHeaderStorageTable::TICKER_HEIGHT_INDEX) .with_value(&ticker) + .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))? + .with_value(&BeBigUint::from(height)) .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))?; block_headers_db diff --git a/mm2src/mm2_db/Cargo.toml b/mm2src/mm2_db/Cargo.toml index 152bbaec65..95baa9715f 100644 --- a/mm2src/mm2_db/Cargo.toml +++ b/mm2src/mm2_db/Cargo.toml @@ -27,4 +27,4 @@ serde_json = { version = "1.0", features = ["preserve_order", "raw_value"] } wasm-bindgen = { version = "0.2.50", features = ["nightly"] } wasm-bindgen-futures = { version = "0.4.1" } wasm-bindgen-test = { version = "0.3.2" } -web-sys = { version = "0.3.55", features = ["console", "CloseEvent", "DomException", "ErrorEvent", "IdbDatabase", "IdbCursor", "IdbCursorWithValue", "IdbFactory", "IdbIndex", "IdbIndexParameters", "IdbObjectStore", "IdbObjectStoreParameters", "IdbOpenDbRequest", "IdbKeyRange", "IdbTransaction", "IdbTransactionMode", "IdbVersionChangeEvent", "MessageEvent", "WebSocket"] } +web-sys = { version = "0.3.55", features = ["console", "CloseEvent", "DomException", "ErrorEvent", "IdbDatabase", "IdbCursor", "IdbCursorWithValue", "IdbCursorDirection", "IdbFactory", "IdbIndex", "IdbIndexParameters", "IdbObjectStore", "IdbObjectStoreParameters", "IdbOpenDbRequest", "IdbKeyRange", "IdbTransaction", "IdbTransactionMode", "IdbVersionChangeEvent", "MessageEvent", "WebSocket"] } diff --git a/mm2src/mm2_db/src/indexed_db/drivers/cursor/cursor.rs b/mm2src/mm2_db/src/indexed_db/drivers/cursor/cursor.rs index b748b6939d..6eb10b754b 100644 --- a/mm2src/mm2_db/src/indexed_db/drivers/cursor/cursor.rs +++ b/mm2src/mm2_db/src/indexed_db/drivers/cursor/cursor.rs @@ -12,7 +12,7 @@ use serde_json::{self as json, Value as Json}; use std::convert::TryInto; use wasm_bindgen::prelude::*; use wasm_bindgen::JsCast; -use web_sys::{IdbCursorWithValue, IdbIndex, IdbKeyRange, IdbRequest}; +use web_sys::{IdbCursorDirection, IdbCursorWithValue, IdbIndex, IdbKeyRange, IdbRequest}; mod empty_cursor; mod multi_key_bound_cursor; @@ -87,6 +87,7 @@ pub enum CursorBoundValue { pub struct CursorFilters { pub(crate) only_keys: Vec<(String, Json)>, pub(crate) bound_keys: Vec<(String, CursorBoundValue, CursorBoundValue)>, + pub(crate) reverse: bool, } impl From for CursorBoundValue { @@ -202,10 +203,21 @@ pub(crate) struct CursorDriver { impl CursorDriver { pub(crate) fn init_cursor(db_index: IdbIndex, filters: CursorFilters) -> CursorResult { + let reverse = filters.reverse; let inner = IdbCursorEnum::new(filters)?; let cursor_request_result = match inner.key_range()? { + Some(key_range) if reverse => { + db_index.open_cursor_with_range_and_direction(&key_range, IdbCursorDirection::Prev) + }, Some(key_range) => db_index.open_cursor_with_range(&key_range), + // Please note that `IndexedDb` doesn't allow to open a cursor with a direction + // but without a key range. + None if reverse => { + return MmError::err(CursorError::ErrorOpeningCursor { + description: format!("Direction cannot be specified without a range"), + }) + }, None => db_index.open_cursor(), }; let cursor_request = cursor_request_result.map_err(|e| CursorError::ErrorOpeningCursor { diff --git a/mm2src/mm2_db/src/indexed_db/indexed_cursor.rs b/mm2src/mm2_db/src/indexed_db/indexed_cursor.rs index 1c16cc69a6..04dd7c37cb 100644 --- a/mm2src/mm2_db/src/indexed_db/indexed_cursor.rs +++ b/mm2src/mm2_db/src/indexed_db/indexed_cursor.rs @@ -107,6 +107,11 @@ impl<'transaction, 'reference, Table: TableSignature> CursorBuilder<'transaction self } + pub fn reverse(mut self) -> Self { + self.filters.reverse = true; + self + } + /// Opens a cursor by the specified `index`. /// https://developer.mozilla.org/en-US/docs/Web/API/IDBObjectStore/openCursor pub async fn open_cursor(self, index: &str) -> CursorResult> { @@ -250,6 +255,15 @@ mod tests { } } + #[track_caller] + async fn next_item(cursor_iter: &mut CursorIter<'_, Table>) -> Option
{ + cursor_iter + .next() + .await + .expect("!CursorIter::next") + .map(|(_item_id, item)| item) + } + /// The table with `BeBigUint` parameters. #[derive(Clone, Debug, Default, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)] #[serde(deny_unknown_fields)] @@ -710,15 +724,6 @@ mod tests { .await .expect("!CursorBuilder::open_cursor"); - #[track_caller] - async fn next_item(cursor_iter: &mut CursorIter<'_, Table>) -> Option
{ - cursor_iter - .next() - .await - .expect("!CursorIter::next") - .map(|(_item_id, item)| item) - } - // The items must be sorted by `rel_coin_value`. assert_eq!( next_item(&mut cursor_iter).await, @@ -741,6 +746,36 @@ mod tests { assert!(next_item(&mut cursor_iter).await.is_none()); } + #[wasm_bindgen_test] + async fn test_rev_iter_without_constraints() { + const DB_NAME: &str = "TEST_REV_ITER_WITHOUT_CONSTRAINTS"; + const DB_VERSION: u32 = 1; + + register_wasm_log(); + + let db = IndexedDbBuilder::new(DbIdentifier::for_test(DB_NAME)) + .with_version(DB_VERSION) + .with_table::() + .build() + .await + .expect("!IndexedDb::init"); + let transaction = db.transaction().await.expect("!IndexedDb::transaction"); + let table = transaction + .table::() + .await + .expect("!DbTransaction::open_table"); + + table + .cursor_builder() + .reverse() + .open_cursor("rel_coin_value") + .await + .map(|_| ()) + .expect_err( + "CursorBuilder::open_cursor should have failed because 'reverse' can be used with key range only", + ); + } + #[wasm_bindgen_test] async fn test_iter_single_key_bound_cursor() { const DB_NAME: &str = "TEST_ITER_SINGLE_KEY_BOUND_CURSOR"; @@ -777,15 +812,6 @@ mod tests { .await .expect("!CursorBuilder::open_cursor"); - #[track_caller] - async fn next_item(cursor_iter: &mut CursorIter<'_, Table>) -> Option
{ - cursor_iter - .next() - .await - .expect("!CursorIter::next") - .map(|(_item_id, item)| item) - } - // The items must be sorted by `rel_coin_value`. assert_eq!( next_item(&mut cursor_iter).await, @@ -803,4 +829,59 @@ mod tests { // Try to poll one more time. This should not fail but return `None`. assert!(next_item(&mut cursor_iter).await.is_none()); } + + #[wasm_bindgen_test] + async fn test_rev_iter_single_key_bound_cursor() { + const DB_NAME: &str = "TEST_REV_ITER_SINGLE_KEY_BOUND_CURSOR"; + const DB_VERSION: u32 = 1; + + register_wasm_log(); + + let items = vec![ + swap_item!("uuid1", "RICK", "MORTY", 10, 3, 700), + swap_item!("uuid2", "MORTY", "KMD", 95000, 1, 721), + swap_item!("uuid3", "RICK", "XYZ", 7, u32::MAX, 1281), // + + swap_item!("uuid4", "RICK", "MORTY", 8, 6, 92), // + + swap_item!("uuid5", "QRC20", "RICK", 2, 4, 721), + swap_item!("uuid6", "KMD", "MORTY", 12, 3124, 214), // + + ]; + + let db = IndexedDbBuilder::new(DbIdentifier::for_test(DB_NAME)) + .with_version(DB_VERSION) + .with_table::() + .build() + .await + .expect("!IndexedDb::init"); + let transaction = db.transaction().await.expect("!IndexedDb::transaction"); + let table = transaction + .table::() + .await + .expect("!DbTransaction::open_table"); + fill_table(&table, items).await; + + let mut cursor_iter = table + .cursor_builder() + .bound("rel_coin_value", 5u32, u32::MAX) + .reverse() + .open_cursor("rel_coin_value") + .await + .expect("!CursorBuilder::open_cursor"); + + // The items must be sorted in reverse order by `rel_coin_value`. + assert_eq!( + next_item(&mut cursor_iter).await, + Some(swap_item!("uuid3", "RICK", "XYZ", 7, u32::MAX, 1281)) + ); + assert_eq!( + next_item(&mut cursor_iter).await, + Some(swap_item!("uuid6", "KMD", "MORTY", 12, 3124, 214)) + ); + assert_eq!( + next_item(&mut cursor_iter).await, + Some(swap_item!("uuid4", "RICK", "MORTY", 8, 6, 92)) + ); + assert!(next_item(&mut cursor_iter).await.is_none()); + // Try to poll one more time. This should not fail but return `None`. + assert!(next_item(&mut cursor_iter).await.is_none()); + } } From 3b4a6e09c4484c323e36d6484f89b04925b236a1 Mon Sep 17 00:00:00 2001 From: Sergey Boyko Date: Fri, 24 Feb 2023 17:28:38 +0100 Subject: [PATCH 5/6] Fix TODOs --- .../src/indexed_db/drivers/cursor/cursor.rs | 24 +++++++------- .../indexed_db/drivers/cursor/empty_cursor.rs | 6 ++-- .../drivers/cursor/multi_key_bound_cursor.rs | 31 +++++++++---------- .../drivers/cursor/multi_key_cursor.rs | 6 ++-- .../drivers/cursor/single_key_bound_cursor.rs | 6 ++-- .../drivers/cursor/single_key_cursor.rs | 6 ++-- 6 files changed, 37 insertions(+), 42 deletions(-) diff --git a/mm2src/mm2_db/src/indexed_db/drivers/cursor/cursor.rs b/mm2src/mm2_db/src/indexed_db/drivers/cursor/cursor.rs index 6eb10b754b..0fdc26261b 100644 --- a/mm2src/mm2_db/src/indexed_db/drivers/cursor/cursor.rs +++ b/mm2src/mm2_db/src/indexed_db/drivers/cursor/cursor.rs @@ -168,17 +168,15 @@ impl CursorBoundValue { } } -/// TODO rename to `CursorAction`. #[derive(Debug, PartialEq)] -pub enum CollectCursorAction { +pub enum CursorAction { Continue, ContinueWithValue(JsValue), Stop, } -/// TODO rename to `IterItemAction`. #[derive(Debug, PartialEq)] -pub enum CollectItemAction { +pub enum CursorItemAction { Include, Skip, } @@ -186,7 +184,7 @@ pub enum CollectItemAction { pub trait CursorDriverImpl: Sized { fn key_range(&self) -> CursorResult>; - fn on_iteration(&mut self, key: JsValue) -> CursorResult<(CollectItemAction, CollectCursorAction)>; + fn on_iteration(&mut self, key: JsValue) -> CursorResult<(CursorItemAction, CursorAction)>; } pub(crate) struct CursorDriver { @@ -194,7 +192,7 @@ pub(crate) struct CursorDriver { inner: IdbCursorEnum, cursor_request: IdbRequest, cursor_item_rx: mpsc::Receiver>, - /// Whether we got `CollectCursorAction::Stop` at the last iteration or not. + /// Whether we got `CursorAction::Stop` at the last iteration or not. stopped: bool, /// We need to hold the closures in memory till `cursor` exists. _onsuccess_closure: Closure, @@ -244,7 +242,7 @@ impl CursorDriver { pub(crate) async fn next(&mut self) -> CursorResult> { loop { - // Check if we got `CollectCursorAction::Stop` at the last iteration. + // Check if we got `CursorAction::Stop` at the last iteration. if self.stopped { return Ok(None); } @@ -285,10 +283,10 @@ impl CursorDriver { let (item_action, cursor_action) = self.inner.on_iteration(key)?; match cursor_action { - CollectCursorAction::Continue => cursor.continue_().map_to_mm(|e| CursorError::AdvanceError { + CursorAction::Continue => cursor.continue_().map_to_mm(|e| CursorError::AdvanceError { description: stringify_js_error(&e), })?, - CollectCursorAction::ContinueWithValue(next_value) => { + CursorAction::ContinueWithValue(next_value) => { cursor .continue_with_key(&next_value) .map_to_mm(|e| CursorError::AdvanceError { @@ -299,13 +297,13 @@ impl CursorDriver { // Here we set the `stopped` flag so we return `Ok(None)` at the next iteration immediately. // This is required because `item_action` can be `CollectItemAction::Include`, // and at this iteration we will return `Ok(Some)`. - CollectCursorAction::Stop => self.stopped = true, + CursorAction::Stop => self.stopped = true, } match item_action { - CollectItemAction::Include => return Ok(Some(item.into_pair())), + CursorItemAction::Include => return Ok(Some(item.into_pair())), // Try to fetch the next item. - CollectItemAction::Skip => (), + CursorItemAction::Skip => (), } } } @@ -366,7 +364,7 @@ impl CursorDriverImpl for IdbCursorEnum { } } - fn on_iteration(&mut self, key: JsValue) -> CursorResult<(CollectItemAction, CollectCursorAction)> { + fn on_iteration(&mut self, key: JsValue) -> CursorResult<(CursorItemAction, CursorAction)> { match self { IdbCursorEnum::Empty(empty) => empty.on_iteration(key), IdbCursorEnum::SingleKey(single) => single.on_iteration(key), diff --git a/mm2src/mm2_db/src/indexed_db/drivers/cursor/empty_cursor.rs b/mm2src/mm2_db/src/indexed_db/drivers/cursor/empty_cursor.rs index ca0e9d213a..441180d8bb 100644 --- a/mm2src/mm2_db/src/indexed_db/drivers/cursor/empty_cursor.rs +++ b/mm2src/mm2_db/src/indexed_db/drivers/cursor/empty_cursor.rs @@ -1,4 +1,4 @@ -use super::{CollectCursorAction, CollectItemAction, CursorDriverImpl, CursorResult}; +use super::{CursorAction, CursorDriverImpl, CursorItemAction, CursorResult}; use wasm_bindgen::prelude::*; use web_sys::IdbKeyRange; @@ -8,7 +8,7 @@ pub struct IdbEmptyCursor; impl CursorDriverImpl for IdbEmptyCursor { fn key_range(&self) -> CursorResult> { Ok(None) } - fn on_iteration(&mut self, _key: JsValue) -> CursorResult<(CollectItemAction, CollectCursorAction)> { - Ok((CollectItemAction::Include, CollectCursorAction::Continue)) + fn on_iteration(&mut self, _key: JsValue) -> CursorResult<(CursorItemAction, CursorAction)> { + Ok((CursorItemAction::Include, CursorAction::Continue)) } } diff --git a/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_bound_cursor.rs b/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_bound_cursor.rs index 170b3c516f..6a6b224fcd 100644 --- a/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_bound_cursor.rs +++ b/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_bound_cursor.rs @@ -1,5 +1,5 @@ -use super::{index_key_as_array, CollectCursorAction, CollectItemAction, CursorBoundValue, CursorDriverImpl, - CursorError, CursorResult}; +use super::{index_key_as_array, CursorAction, CursorBoundValue, CursorDriverImpl, CursorError, CursorItemAction, + CursorResult}; use common::{deserialize_from_js, serialize_to_js, stringify_js_error}; use js_sys::Array; use mm2_err_handle::prelude::*; @@ -108,7 +108,7 @@ impl CursorDriverImpl for IdbMultiKeyBoundCursor { /// The range `IDBKeyRange.bound([2,2], [4,4])` includes values like `[3,0]` and `[3,5]` as `[2,2] < [3,0] < [3,5] < [4,4]`, /// so we need to do additional filtering. /// For more information on why it's required, see https://stackoverflow.com/a/32976384. - fn on_iteration(&mut self, index_key: JsValue) -> CursorResult<(CollectItemAction, CollectCursorAction)> { + fn on_iteration(&mut self, index_key: JsValue) -> CursorResult<(CursorItemAction, CursorAction)> { let index_keys_js_array = index_key_as_array(index_key)?; let index_keys: Vec = index_keys_js_array .iter() @@ -157,8 +157,8 @@ impl CursorDriverImpl for IdbMultiKeyBoundCursor { return Ok(( // the `actual_index_value` is not in our expected bounds - CollectItemAction::Skip, - CollectCursorAction::ContinueWithValue(new_index.into()), + CursorItemAction::Skip, + CursorAction::ContinueWithValue(new_index.into()), )); } if &actual_index_value > upper_bound { @@ -177,13 +177,13 @@ impl CursorDriverImpl for IdbMultiKeyBoundCursor { return Ok(( // the `actual_index_value` is not in our expected bounds - CollectItemAction::Skip, - CollectCursorAction::ContinueWithValue(new_index.into()), + CursorItemAction::Skip, + CursorAction::ContinueWithValue(new_index.into()), )); } // otherwise there is no an index greater than actual `index`, stop the cursor - return Ok((CollectItemAction::Skip, CollectCursorAction::Stop)); + return Ok((CursorItemAction::Skip, CursorAction::Stop)); } let increased_index_key = actual_index_value.next(); @@ -194,7 +194,7 @@ impl CursorDriverImpl for IdbMultiKeyBoundCursor { idx_in_index += 1; idx_in_bounds += 1; } - Ok((CollectItemAction::Include, CollectCursorAction::Continue)) + Ok((CursorItemAction::Include, CursorAction::Continue)) } } @@ -212,7 +212,7 @@ mod tests { let result = cursor.on_iteration(input_index_js_value); assert_eq!( result, - Ok((CollectItemAction::Include, CollectCursorAction::Continue)), + Ok((CursorItemAction::Include, CursorAction::Continue)), "'{}' index is expected to be in a bound", input_index ); @@ -230,15 +230,12 @@ mod tests { .expect(&format!("Error due to the index '{:?}'", input_index)); let actual_next: Json = match cursor_action { - CollectCursorAction::ContinueWithValue(next_index_js_value) => { + CursorAction::ContinueWithValue(next_index_js_value) => { deserialize_from_js(next_index_js_value).expect("Error deserializing next index}") }, - action => panic!( - "Expected 'CollectCursorAction::ContinueWithValue', found '{:?}'", - action - ), + action => panic!("Expected 'CursorAction::ContinueWithValue', found '{:?}'", action), }; - assert_eq!(item_action, CollectItemAction::Skip); + assert_eq!(item_action, CursorItemAction::Skip); assert_eq!(actual_next, expected_next); } } @@ -250,7 +247,7 @@ mod tests { let result = cursor.on_iteration(input_index_js_value); assert_eq!( result, - Ok((CollectItemAction::Skip, CollectCursorAction::Stop)), + Ok((CursorItemAction::Skip, CursorAction::Stop)), "'{}' index is expected to be out of bound", input_index ); diff --git a/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_cursor.rs b/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_cursor.rs index ddd2cd8fa0..f6b128aa1b 100644 --- a/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_cursor.rs +++ b/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_cursor.rs @@ -1,4 +1,4 @@ -use super::{CollectCursorAction, CollectItemAction, CursorDriverImpl, CursorError, CursorResult}; +use super::{CursorAction, CursorDriverImpl, CursorError, CursorItemAction, CursorResult}; use common::{serialize_to_js, stringify_js_error}; use js_sys::Array; use mm2_err_handle::prelude::*; @@ -36,7 +36,7 @@ impl CursorDriverImpl for IdbMultiKeyCursor { Ok(Some(key_range)) } - fn on_iteration(&mut self, _key: JsValue) -> CursorResult<(CollectItemAction, CollectCursorAction)> { - Ok((CollectItemAction::Include, CollectCursorAction::Continue)) + fn on_iteration(&mut self, _key: JsValue) -> CursorResult<(CursorItemAction, CursorAction)> { + Ok((CursorItemAction::Include, CursorAction::Continue)) } } diff --git a/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_bound_cursor.rs b/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_bound_cursor.rs index d868984526..92e7ee087b 100644 --- a/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_bound_cursor.rs +++ b/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_bound_cursor.rs @@ -1,4 +1,4 @@ -use super::{CollectCursorAction, CollectItemAction, CursorBoundValue, CursorDriverImpl, CursorError, CursorResult}; +use super::{CursorAction, CursorBoundValue, CursorDriverImpl, CursorError, CursorItemAction, CursorResult}; use common::{log::warn, stringify_js_error}; use mm2_err_handle::prelude::*; use wasm_bindgen::prelude::*; @@ -56,7 +56,7 @@ impl CursorDriverImpl for IdbSingleKeyBoundCursor { Ok(Some(key_range)) } - fn on_iteration(&mut self, _key: JsValue) -> CursorResult<(CollectItemAction, CollectCursorAction)> { - Ok((CollectItemAction::Include, CollectCursorAction::Continue)) + fn on_iteration(&mut self, _key: JsValue) -> CursorResult<(CursorItemAction, CursorAction)> { + Ok((CursorItemAction::Include, CursorAction::Continue)) } } diff --git a/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_cursor.rs b/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_cursor.rs index 6416317943..b3d6efa476 100644 --- a/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_cursor.rs +++ b/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_cursor.rs @@ -1,4 +1,4 @@ -use super::{CollectCursorAction, CollectItemAction, CursorDriverImpl, CursorError, CursorResult}; +use super::{CursorAction, CursorDriverImpl, CursorError, CursorItemAction, CursorResult}; use common::{serialize_to_js, stringify_js_error}; use mm2_err_handle::prelude::*; use serde_json::Value as Json; @@ -38,7 +38,7 @@ impl CursorDriverImpl for IdbSingleKeyCursor { Ok(Some(key_range)) } - fn on_iteration(&mut self, _key: JsValue) -> CursorResult<(CollectItemAction, CollectCursorAction)> { - Ok((CollectItemAction::Include, CollectCursorAction::Continue)) + fn on_iteration(&mut self, _key: JsValue) -> CursorResult<(CursorItemAction, CursorAction)> { + Ok((CursorItemAction::Include, CursorAction::Continue)) } } From d8bc96c1501e8343ffd0ad4fbcec0590ce1c876f Mon Sep 17 00:00:00 2001 From: Sergey Boyko Date: Thu, 2 Mar 2023 12:15:25 +0100 Subject: [PATCH 6/6] Fix PR issues --- .../wasm/indexeddb_block_header_storage.rs | 30 +++++++++---------- 1 file changed, 14 insertions(+), 16 deletions(-) diff --git a/mm2src/coins/utxo/utxo_block_header_storage/wasm/indexeddb_block_header_storage.rs b/mm2src/coins/utxo/utxo_block_header_storage/wasm/indexeddb_block_header_storage.rs index 5b0b4b0f6c..3e19792548 100644 --- a/mm2src/coins/utxo/utxo_block_header_storage/wasm/indexeddb_block_header_storage.rs +++ b/mm2src/coins/utxo/utxo_block_header_storage/wasm/indexeddb_block_header_storage.rs @@ -195,14 +195,13 @@ impl BlockHeaderStorageOps for IDBBlockHeadersStorage { .await .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))?; - if let Some((_item_id, item)) = maybe_item { - let height = item - .height - .to_u64() - .ok_or_else(|| BlockHeaderStorageError::get_err(&ticker, "height is too large".to_string()))?; - return Ok(Some(height)); - } - Ok(None) + maybe_item + .map(|(_, item)| { + item.height + .to_u64() + .ok_or_else(|| BlockHeaderStorageError::get_err(&ticker, "height is too large".to_string())) + }) + .transpose() } async fn get_last_block_header_with_non_max_bits( @@ -292,14 +291,13 @@ impl BlockHeaderStorageOps for IDBBlockHeadersStorage { .await .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))?; - if let Some((_item_id, header)) = maybe_item { - let height = header - .height - .to_i64() - .ok_or_else(|| BlockHeaderStorageError::get_err(&ticker, "height is too large".to_string()))?; - return Ok(Some(height)); - } - Ok(None) + maybe_item + .map(|(_, item)| { + item.height + .to_i64() + .ok_or_else(|| BlockHeaderStorageError::get_err(&ticker, "height is too large".to_string())) + }) + .transpose() } async fn remove_headers_up_to_height(&self, to_height: u64) -> Result<(), BlockHeaderStorageError> {