Refactor take_while pattern

DeFiCh · Sep 12, 2024 · bbf4e59 · bbf4e59
1 parent ba4f838
commit bbf4e59
Show file tree

Hide file tree

Showing 7 changed files with 90 additions and 106 deletions.
diff --git a/lib/ain-ocean/src/api/address.rs b/lib/ain-ocean/src/api/address.rs
@@ -323,15 +323,12 @@ async fn list_transactions(
             SortOrder::Descending,
         )?
         .skip(usize::from(query.next.is_some()))
-        .take(query.size)
-        .take_while(|item| match item {
-            Ok((k, _)) => k.0 == hid,
-            _ => true,
-        })
-        .map(|item| {
-            let (_, v) = item?;
-            Ok(v.into())
+        .filter_map(|item| match item {
+            Ok((k, v)) if k.0 == hid => Some(Ok(v.into())),
+            Ok(_) => None,
+            Err(e) => Some(Err(e.into())),
         })
+        .take(query.size)
         .collect::<Result<Vec<_>>>()?;
 
     Ok(ApiPagedResponse::of(res, query.size, |item| {
@@ -422,16 +419,12 @@ async fn list_transaction_unspent(
             SortOrder::Ascending,
         )?
         .skip(usize::from(query.next.is_some()))
-        .take(query.size)
-        .take_while(|item| match item {
-            Ok((k, _)) => k.0 == hid.clone(),
-            _ => true,
-        })
-        .map(|item| {
-            let (_, v) = item?;
-            let res = v.into();
-            Ok(res)
+        .filter_map(|item| match item {
+            Ok((k, v)) if k.0 == hid => Some(Ok(v.into())),
+            Ok(_) => None,
+            Err(e) => Some(Err(e.into())),
         })
+        .take(query.size)
         .collect::<Result<Vec<_>>>()?;
 
     Ok(ApiPagedResponse::of(res, query.size, |item| {

diff --git a/lib/ain-ocean/src/api/block.rs b/lib/ain-ocean/src/api/block.rs
@@ -146,12 +146,15 @@ async fn get_transactions(
     let txs = repository
         .list(Some(next), SortOrder::Ascending)?
         .paginate(&query)
-        .take_while(|item| match item {
-            Ok(((h, _), _)) => h == &hash,
-            _ => true,
+        .filter_map(|item| match item {
+            Ok(v) if v.0 .0 == hash => Some(
+                repository
+                    .retrieve_primary_value(Ok(v))
+                    .map(TransactionResponse::from),
+            ),
+            Ok(_) => None,
+            Err(e) => Some(Err(e.into())),
         })
-        .map(|el| repository.retrieve_primary_value(el))
-        .map(|v| v.map(TransactionResponse::from))
         .collect::<Result<Vec<_>>>()?;
 
     Ok(ApiPagedResponse::of(txs, query.size, |tx| tx.order))

diff --git a/lib/ain-ocean/src/api/loan.rs b/lib/ain-ocean/src/api/loan.rs
@@ -525,25 +525,22 @@ async fn list_vault_auction_history(
             Some((vault_id, batch_index, next.0, next.1)),
             SortOrder::Descending,
         )?
-        .take(size)
-        .take_while(|item| match item {
-            Ok((k, _)) => k.0 == vault_id && k.1 == batch_index,
-            _ => true,
-        })
-        .map(|item| {
-            let (_, id) = item?;
-
-            let auction = ctx
-                .services
-                .auction
-                .by_id
-                .get(&id)?
-                .context(NotFoundSnafu {
-                    kind: NotFoundKind::Auction,
-                })?;
-
-            Ok(auction)
+        .filter_map(|item| match item {
+            Ok((k, id)) if k.0 == vault_id && k.1 == batch_index => {
+                match ctx.services.auction.by_id.get(&id) {
+                    Ok(Some(auction)) => Some(Ok(auction)),
+                    Ok(None) => Some(Err(NotFoundSnafu {
+                        kind: NotFoundKind::Auction,
+                    }
+                    .build()
+                    .into())),
+                    Err(e) => Some(Err(e.into())),
+                }
+            }
+            Ok(_) => None,
+            Err(e) => Some(Err(e.into())),
         })
+        .take(size)
         .collect::<Result<Vec<_>>>()?;
 
     Ok(ApiPagedResponse::of(auctions, query.size, |auction| {
@@ -617,22 +614,27 @@ async fn map_liquidation_batches(
             batch.index,
             Txid::from_byte_array([0xffu8; 32]),
         );
-        let bids = repo
+
+        let froms = repo
             .by_id
             .list(Some(id), SortOrder::Descending)?
-            .take_while(|item| match item {
-                Ok(((vid, bindex, _), _)) => vid.to_string() == vault_id && bindex == &batch.index,
-                _ => true,
+            .filter_map(|item| match item {
+                Ok(((vid, bindex, _), v))
+                    if vid.to_string() == vault_id && bindex == batch.index =>
+                {
+                    Some(Ok(v))
+                }
+                Ok(_) => None,
+                Err(e) => Some(Err(e.into())),
             })
-            .collect::<Vec<_>>();
-        let froms = bids
-            .into_iter()
-            .map(|bid| {
-                let (_, v) = bid?;
-                let from_addr = from_script(v.from, ctx.network.into())?;
-                Ok::<String, Error>(from_addr)
+            .map(|result| {
+                result.and_then(|v| {
+                    let from_addr = from_script(v.from, ctx.network.into())?;
+                    Ok(from_addr)
+                })
             })
-            .collect::<Result<Vec<_>>>()?;
+            .collect::<Result<Vec<String>>>()?;
+
         vec.push(VaultLiquidatedBatchResponse {
             index: batch.index,
             collaterals: map_token_amounts(ctx, batch.collaterals).await?,

diff --git a/lib/ain-ocean/src/api/pool_pair/mod.rs b/lib/ain-ocean/src/api/pool_pair/mod.rs
@@ -388,15 +388,12 @@ async fn list_pool_swaps(
         .pool
         .by_id
         .list(Some(next), SortOrder::Descending)?
-        .take(size)
-        .take_while(|item| match item {
-            Ok((k, _)) => k.0 == id,
-            _ => true,
-        })
-        .map(|item| {
-            let (_, swap) = item?;
-            Ok(PoolSwapResponse::from(swap))
+        .filter_map(|item| match item {
+            Ok((k, swap)) if k.0 == id => Some(Ok(PoolSwapResponse::from(swap))),
+            Ok(_) => None,
+            Err(e) => Some(Err(e.into())),
         })
+        .take(size)
         .collect::<Result<Vec<_>>>()?;
 
     Ok(ApiPagedResponse::of(swaps, query.size, |swap| {
@@ -427,11 +424,8 @@ async fn list_pool_swaps_verbose(
         .pool
         .by_id
         .list(Some(next), SortOrder::Descending)?
+        .filter(|item| matches!(item, Ok((k, _)) if k.0 == id))
         .take(size)
-        .take_while(|item| match item {
-            Ok((k, _)) => k.0 == id,
-            _ => true,
-        })
         .map(|item| async {
             let (_, swap) = item?;
             let from = find_swap_from(&ctx, swap.clone()).await?;
@@ -505,12 +499,14 @@ async fn list_pool_swap_aggregates(
     let aggregates = repository
         .by_key
         .list(Some((pool_id, interval, next)), SortOrder::Descending)?
-        .take(query.size)
-        .take_while(|item| match item {
-            Ok((k, _)) => k.0 == pool_id && k.1 == interval,
-            _ => true,
+        .filter_map(|item| match item {
+            Ok(k) if k.0 .0 == pool_id && k.0 .1 == interval => {
+                Some(repository.by_key.retrieve_primary_value(Ok(k)))
+            }
+            Ok(_) => None,
+            Err(e) => Some(Err(e.into())),
         })
-        .map(|e| repository.by_key.retrieve_primary_value(e))
+        .take(query.size)
         .collect::<Result<Vec<_>>>()?;
 
     let mut aggregated_usd = Vec::<PoolSwapAggregatedResponse>::new();

diff --git a/lib/ain-ocean/src/api/pool_pair/service.rs b/lib/ain-ocean/src/api/pool_pair/service.rs
@@ -345,12 +345,14 @@ async fn gather_amount(
     let swaps = repository
         .by_key
         .list(Some((pool_id, interval, i64::MAX)), SortOrder::Descending)?
-        .take(count)
-        .take_while(|item| match item {
-            Ok((k, _)) => k.0 == pool_id && k.1 == interval,
-            _ => true,
+        .filter_map(|item| match item {
+            Ok(v) if v.0 .0 == pool_id && v.0 .1 == interval => {
+                Some(repository.by_key.retrieve_primary_value(Ok(v)))
+            }
+            Ok(_) => None,
+            Err(e) => Some(Err(e.into())),
         })
-        .map(|e| repository.by_key.retrieve_primary_value(e))
+        .take(count)
         .collect::<Result<Vec<_>>>()?;
 
     let mut aggregated = HashMap::<String, Decimal>::new();

diff --git a/lib/ain-ocean/src/api/prices.rs b/lib/ain-ocean/src/api/prices.rs
@@ -24,7 +24,7 @@ use crate::{
     error::{ApiError, Error, OtherSnafu},
     model::{
         BlockContext, OracleIntervalSeconds, OraclePriceActive, OraclePriceActiveNextOracles,
-        OraclePriceAggregated, OraclePriceAggregatedInterval, OracleTokenCurrency, PriceTicker,
+        OraclePriceAggregated, OraclePriceAggregatedInterval, PriceTicker,
     },
     storage::{RepositoryOps, SortOrder},
     Result,
@@ -159,15 +159,12 @@ async fn get_feed(
     let oracle_aggregated = repo
         .by_id
         .list(Some(id), SortOrder::Descending)?
-        .take(query.size)
-        .take_while(|item| match item {
-            Ok((k, _)) => k.0 == token && k.1 == currency,
-            _ => true,
-        })
-        .map(|item| {
-            let (_, v) = item?;
-            Ok(v)
+        .filter_map(|item| match item {
+            Ok((k, v)) if k.0 == token && k.1 == currency => Some(Ok(v)),
+            Ok(_) => None,
+            Err(e) => Some(Err(e.into())),
         })
+        .take(query.size)
         .collect::<Result<Vec<_>>>()?;
 
     Ok(ApiPagedResponse::of(
@@ -270,16 +267,13 @@ async fn get_oracles(
         .oracle_token_currency
         .by_id
         .list(Some(id.clone()), SortOrder::Descending)?
-        .take(query.size)
-        .take_while(|item| match item {
-            Ok((k, _)) => k.0 == id.0 && k.1 == id.1,
-            _ => true,
+        .filter_map(|item| match item {
+            Ok((k, oracle)) if k.0 == id.0 && k.1 == id.1 => Some(Ok(oracle)),
+            Ok(_) => None,
+            Err(e) => Some(Err(e.into())),
         })
-        .flat_map(|item| {
-            let (_, oracle) = item?;
-            Ok::<OracleTokenCurrency, Error>(oracle)
-        })
-        .collect::<Vec<_>>();
+        .take(query.size)
+        .collect::<Result<Vec<_>>>()?;
 
     let mut prices = Vec::new();
     for oracle in oracles {

diff --git a/lib/ain-ocean/src/api/transactions.rs b/lib/ain-ocean/src/api/transactions.rs
@@ -47,13 +47,10 @@ async fn get_vins(
         .vin_by_id
         .list(Some(next), SortOrder::Descending)?
         .paginate(&query)
-        .take_while(|item| match item {
-            Ok((_, vin)) => vin.txid == id,
-            _ => true,
-        })
-        .map(|item| {
-            let (_, v) = item?;
-            Ok(v)
+        .filter_map(|item| match item {
+            Ok((_, vout)) if vout.txid == id => Some(Ok(vout)),
+            Ok(_) => None,
+            Err(e) => Some(Err(e.into())),
         })
         .collect::<Result<Vec<_>>>()?;
 
@@ -77,13 +74,10 @@ async fn get_vouts(
         .vout_by_id
         .list(Some((id, next)), SortOrder::Ascending)?
         .paginate(&query)
-        .take_while(|item| match item {
-            Ok((_, vout)) => vout.txid == id,
-            _ => true,
-        })
-        .map(|item| {
-            let (_, v) = item?;
-            Ok(v)
+        .filter_map(|item| match item {
+            Ok((_, vin)) if vin.txid == id => Some(Ok(vin)),
+            Ok(_) => None,
+            Err(e) => Some(Err(e.into())),
         })
         .collect::<Result<Vec<_>>>()?;