From 9ddbb6404970d6c5b6c1c7d05ae088cad40343bf Mon Sep 17 00:00:00 2001 From: Matthias Beyer Date: Tue, 8 Nov 2022 15:22:37 +0100 Subject: cargo-fmt Signed-off-by: Matthias Beyer --- src/async_dag.rs | 170 ++++++++++++++++++++++++++++++----------------------- src/dag_backend.rs | 9 ++- src/lib.rs | 1 - src/node.rs | 1 - src/node_id.rs | 3 +- src/test_impl.rs | 1 - 6 files changed, 102 insertions(+), 83 deletions(-) diff --git a/src/async_dag.rs b/src/async_dag.rs index 57ecc16..67b8dc2 100644 --- a/src/async_dag.rs +++ b/src/async_dag.rs @@ -6,20 +6,21 @@ use std::pin::Pin; -use anyhow::Result; use anyhow::anyhow; +use anyhow::Result; use futures::stream::StreamExt; use futures::task::Poll; +use crate::DagBackend; use crate::Node; use crate::NodeId; -use crate::DagBackend; /// An async DAG, generic over Node, Node identifier and Backend implementation pub struct AsyncDag - where Id: NodeId + Send, - N: Node, - Backend: DagBackend +where + Id: NodeId + Send, + N: Node, + Backend: DagBackend, { head: Id, backend: Backend, @@ -27,24 +28,20 @@ pub struct AsyncDag } impl AsyncDag - where Id: NodeId + Send, - N: Node, - Backend: DagBackend +where + Id: NodeId + Send, + N: Node, + Backend: DagBackend, { /// Create a new DAG with a backend and a HEAD node /// /// The head node is `DagBackend::put` into the backend before the AsyncDag object is created. pub async fn new(mut backend: Backend, head: N) -> Result { - backend - .put(head) - .await - .map(|id| { - AsyncDag { - head: id, - backend: backend, - _node: std::marker::PhantomData, - } - }) + backend.put(head).await.map(|id| AsyncDag { + head: id, + backend: backend, + _node: std::marker::PhantomData, + }) } /// Load a AsyncDag object using `head` as HEAD node. @@ -53,14 +50,13 @@ impl AsyncDag /// /// This fails if backend.get(head) fails. pub async fn load(backend: Backend, head: Id) -> Result { - backend.get(head) + backend + .get(head) .await? - .map(|(id, _)| { - AsyncDag { - head: id, - backend: backend, - _node: std::marker::PhantomData, - } + .map(|(id, _)| AsyncDag { + head: id, + backend: backend, + _node: std::marker::PhantomData, }) .ok_or_else(|| anyhow!("Node not found")) } @@ -80,18 +76,14 @@ impl AsyncDag /// Check whether an `id` is in the DAG. pub async fn has_id(&self, id: &Id) -> Result { self.stream() - .map(|r| -> Result { - r.map(|(ref node_id, _)| node_id == id) - }) + .map(|r| -> Result { r.map(|(ref node_id, _)| node_id == id) }) .collect::>>() .await .into_iter() - .fold(Ok(false), |acc, e| { - match (acc, e) { - (Err(e), _) => Err(e), - (Ok(_), Err(e)) => Err(e), - (Ok(a), Ok(b)) => Ok(a || b), - } + .fold(Ok(false), |acc, e| match (acc, e) { + (Err(e), _) => Err(e), + (Ok(_), Err(e)) => Err(e), + (Ok(a), Ok(b)) => Ok(a || b), }) } @@ -102,14 +94,14 @@ impl AsyncDag /// # Warning /// /// The order of the nodes is not (yet) guaranteed. - pub fn stream(&self) -> Stream { + pub fn stream(&self) -> Stream { Stream { dag: self, backlog: { let mut v = Vec::with_capacity(2); v.push(self.backend.get(self.head.clone())); v - } + }, } } @@ -143,7 +135,8 @@ impl AsyncDag /// This function creates a new AsyncDag object with the same backend (thus the backend must be /// `Clone` in this case). pub fn branch(&self) -> AsyncDag - where Backend: Clone + where + Backend: Clone, { AsyncDag { head: self.head.clone(), @@ -157,7 +150,8 @@ impl AsyncDag /// Use the `merger` function to merge the two head IDs and generate a new Node instance for /// the new HEAD of `self`. pub async fn merge(&mut self, other: &AsyncDag, merger: M) -> Result - where M: Merger + where + M: Merger, { let node = merger.create_merge_node(&self.head, &other.head)?; let id = self.backend.put(node).await?; @@ -167,16 +161,18 @@ impl AsyncDag } pub trait Merger - where Id: NodeId, - N: Node +where + Id: NodeId, + N: Node, { fn create_merge_node(&self, left_id: &Id, right_id: &Id) -> Result; } impl Merger for F - where Id: NodeId, - N: Node, - F: Fn(&Id, &Id) -> Result, +where + Id: NodeId, + N: Node, + F: Fn(&Id, &Id) -> Result, { fn create_merge_node(&self, left_id: &Id, right_id: &Id) -> Result { (self)(left_id, right_id) @@ -185,22 +181,35 @@ impl Merger for F /// Stream adapter for streaming all nodes in a DAG pub struct Stream<'a, Id, N, Backend> - where Id: NodeId + Send, - N: Node, - Backend: DagBackend +where + Id: NodeId + Send, + N: Node, + Backend: DagBackend, { dag: &'a AsyncDag, - backlog: Vec>> + std::marker::Send + 'a)>>>, + backlog: Vec< + Pin< + Box< + (dyn futures::future::Future>> + + std::marker::Send + + 'a), + >, + >, + >, } impl<'a, Id, N, Backend> futures::stream::Stream for Stream<'a, Id, N, Backend> - where Id: NodeId + Send, - N: Node, - Backend: DagBackend +where + Id: NodeId + Send, + N: Node, + Backend: DagBackend, { type Item = Result<(Id, N)>; - fn poll_next(mut self: std::pin::Pin<&mut Self>, cx: &mut futures::task::Context<'_>) -> futures::task::Poll> { + fn poll_next( + mut self: std::pin::Pin<&mut Self>, + cx: &mut futures::task::Context<'_>, + ) -> futures::task::Poll> { if let Some(mut fut) = self.as_mut().backlog.pop() { match fut.as_mut().poll(cx) { Poll::Ready(Err(e)) => Poll::Ready(Some(Err(e))), @@ -210,14 +219,14 @@ impl<'a, Id, N, Backend> futures::stream::Stream for Stream<'a, Id, N, Backend> self.as_mut().backlog.push(fut); } Poll::Ready(Some(Ok((node_id, node)))) - }, + } Poll::Ready(Ok(None)) => { // backend.get() returned Ok(None), so the referenced node seems not to exist // // TODO: Decide whether we should return an error here. cx.waker().wake_by_ref(); Poll::Pending - }, + } Poll::Pending => { cx.waker().wake_by_ref(); Poll::Pending @@ -229,16 +238,15 @@ impl<'a, Id, N, Backend> futures::stream::Stream for Stream<'a, Id, N, Backend> } } - #[cfg(test)] mod tests { use anyhow::Result; use futures::StreamExt; - use crate::DagBackend; - use crate::AsyncDag; use crate::test_impl as test; + use crate::AsyncDag; + use crate::DagBackend; #[test] fn test_dag_two_nodes() { @@ -254,9 +262,7 @@ mod tests { data: 0, }) }, - { - Some(head.clone()) - }, + { Some(head.clone()) }, ]); { @@ -297,9 +303,7 @@ mod tests { data: 0, }) }, - { - Some(head.clone()) - }, + { Some(head.clone()) }, ]); let dag = tokio_test::block_on(AsyncDag::new(b, head)); @@ -341,11 +345,25 @@ mod tests { assert_eq!(dag.head, test::Id(1)); assert_eq!(dag.backend.0.read().unwrap()[0].as_ref().unwrap().data, 0); - assert!(dag.backend.0.read().unwrap()[0].as_ref().unwrap().parents.is_empty()); + assert!(dag.backend.0.read().unwrap()[0] + .as_ref() + .unwrap() + .parents + .is_empty()); assert_eq!(dag.backend.0.read().unwrap()[1].as_ref().unwrap().data, 1); - assert_eq!(dag.backend.0.read().unwrap()[1].as_ref().unwrap().parents.len(), 1); - assert_eq!(dag.backend.0.read().unwrap()[1].as_ref().unwrap().parents[0], test::Id(0)); + assert_eq!( + dag.backend.0.read().unwrap()[1] + .as_ref() + .unwrap() + .parents + .len(), + 1 + ); + assert_eq!( + dag.backend.0.read().unwrap()[1].as_ref().unwrap().parents[0], + test::Id(0) + ); } #[test] @@ -432,7 +450,11 @@ mod tests { struct M; impl super::Merger for M { - fn create_merge_node(&self, left_id: &test::Id, right_id: &test::Id) -> Result { + fn create_merge_node( + &self, + left_id: &test::Id, + right_id: &test::Id, + ) -> Result { Ok(test::Node { parents: vec![left_id.clone(), right_id.clone()], data: 3, @@ -495,12 +517,15 @@ mod tests { assert_eq!(branched.head, test::Id(2)); } - let merge = tokio_test::block_on(dag.merge(&branched, |left_id: &test::Id, right_id: &test::Id| { - Ok(test::Node { - parents: vec![left_id.clone(), right_id.clone()], - data: 3, - }) - })); + let merge = tokio_test::block_on(dag.merge( + &branched, + |left_id: &test::Id, right_id: &test::Id| { + Ok(test::Node { + parents: vec![left_id.clone(), right_id.clone()], + data: 3, + }) + }, + )); assert!(merge.is_ok()); let _ = merge.unwrap(); @@ -508,4 +533,3 @@ mod tests { assert_eq!(dag.head, test::Id(3)); } } - diff --git a/src/dag_backend.rs b/src/dag_backend.rs index b3077c4..2be9736 100644 --- a/src/dag_backend.rs +++ b/src/dag_backend.rs @@ -7,8 +7,8 @@ use anyhow::Result; use async_trait::async_trait; -use crate::NodeId; use crate::Node; +use crate::NodeId; /// An interface to a DAG backend storage /// @@ -16,10 +16,10 @@ use crate::Node; /// (`DagBackend::get`) nodes. #[async_trait] pub trait DagBackend - where N: Node, - Id: NodeId + Send +where + N: Node, + Id: NodeId + Send, { - /// Get a `Node` from the backend that is identified by `id` /// /// # Returns @@ -102,5 +102,4 @@ mod tests { assert!(node.is_none()); } } - } diff --git a/src/lib.rs b/src/lib.rs index 655cef1..bcb6ae0 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -18,4 +18,3 @@ pub use node_id::*; #[cfg(test)] mod test_impl; - diff --git a/src/node.rs b/src/node.rs index 794c4af..58515a8 100644 --- a/src/node.rs +++ b/src/node.rs @@ -11,4 +11,3 @@ pub trait Node { type Id: NodeId; fn parent_ids(&self) -> Vec; } - diff --git a/src/node_id.rs b/src/node_id.rs index 6211e19..df1e4d4 100644 --- a/src/node_id.rs +++ b/src/node_id.rs @@ -7,5 +7,4 @@ /// A unique identifier for a `Node` /// /// The `NodeId` should be cheap to clone (for example a UUID or some form of a hash value). -pub trait NodeId: Clone + Eq + PartialEq { -} +pub trait NodeId: Clone + Eq + PartialEq {} diff --git a/src/test_impl.rs b/src/test_impl.rs index 672c3fc..a95c4c6 100644 --- a/src/test_impl.rs +++ b/src/test_impl.rs @@ -64,4 +64,3 @@ impl crate::DagBackend for Backend { Ok(Id(idx)) } } - -- cgit v1.2.3