Add Hypberbee::add_stream

src/blocks.rs

@@ -1,6 +1,7 @@
 use std::{collections::BTreeMap, fmt::Debug, sync::Arc};
 
 use derive_builder::Builder;
+use futures_lite::{AsyncRead, AsyncWrite};
 use hypercore::{AppendOutcome, CoreMethods, SharedCore};
 use prost::{bytes::Buf, DecodeError, Message};
 use replicator::{Replicate, Replicator};
@@ -37,6 +38,14 @@ impl BlocksBuilder {
 }
 
 impl Blocks {
+    pub async fn add_stream<S: AsyncRead + AsyncWrite + Send + Sync + Unpin + 'static>(
+        &self,
+        stream: S,
+        is_initiator: bool,
+    ) -> Result<(), HyperbeeError> {
+        Ok(self.replicator.add_stream(stream, is_initiator).await?)
+    }
+
     /// Get a BlockEntry for the given `seq`
     /// # Errors
     /// when the provided `seq` is not in the Hypercore

src/error.rs

@@ -1,3 +1,4 @@
+use replicator::ReplicatorError;
 use std::{num::TryFromIntError, string::FromUtf8Error};
 use thiserror::Error;
 
@@ -48,4 +49,6 @@ pub enum HyperbeeError {
     NoRootError,
     #[error("The tree already has a header")]
     HeaderAlreadyExists,
+    #[error("Replication error: {0}")]
+    ReplicationError(#[from] ReplicatorError),
 }

src/hb.rs

@@ -1,7 +1,7 @@
 use std::{fmt::Debug, path::Path, sync::Arc};
 
 use derive_builder::Builder;
-use futures_lite::Stream;
+use futures_lite::{AsyncRead, AsyncWrite, Stream};
 use hypercore::{AppendOutcome, Hypercore};
 use tokio::sync::RwLock;
 
@@ -27,6 +27,20 @@ pub struct Hyperbee {
 }
 
 impl Hyperbee {
+    /// add replication stream
+    pub async fn add_stream<S: AsyncRead + AsyncWrite + Send + Sync + Unpin + 'static>(
+        &self,
+        stream: S,
+        is_initiator: bool,
+    ) -> Result<(), HyperbeeError> {
+        Ok(self
+            .tree
+            .read()
+            .await
+            .add_stream(stream, is_initiator)
+            .await?)
+    }
+
     /// The number of blocks in the hypercore.
     /// The first block is always the header block so:
     /// `version` would be the `seq` of the next block

src/tree.rs

@@ -1,8 +1,7 @@
 use derive_builder::Builder;
-use futures_lite::{Stream, StreamExt};
+use futures_lite::{AsyncRead, AsyncWrite, Stream, StreamExt};
 use hypercore::{AppendOutcome, Hypercore, HypercoreBuilder, SharedCore, Storage};
 use prost::Message;
-use replicator::Replicate;
 
 use crate::{
     blocks::{Blocks, BlocksBuilder},
@@ -29,6 +28,20 @@ pub struct Tree {
 }
 
 impl Tree {
+    /// add replication stream
+    pub async fn add_stream<S: AsyncRead + AsyncWrite + Send + Sync + Unpin + 'static>(
+        &self,
+        stream: S,
+        is_initiator: bool,
+    ) -> Result<(), HyperbeeError> {
+        Ok(self
+            .blocks
+            .read()
+            .await
+            .add_stream(stream, is_initiator)
+            .await?)
+    }
+
     /// The number of blocks in the hypercore.
     /// The first block is always the header block so:
     /// `version` would be the `seq` of the next block