Add CarWriter sink

src/utils/db/car_stream.rs

@@ -6,7 +6,9 @@ use cid::{
     multihash::{Code, MultihashDigest},
     Cid,
 };
-use futures::{Stream, StreamExt};
+use flume::r#async::RecvStream;
+use futures::{sink::Sink, Stream, StreamExt};
+use fvm_ipld_car::CarHeader as FvmCarHeader;
 use integer_encoding::VarInt;
 use pin_project_lite::pin_project;
 use serde::{Deserialize, Serialize};
@@ -15,6 +17,7 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 use tokio::io::{AsyncBufRead, AsyncRead, AsyncSeek, AsyncSeekExt};
 use tokio_util::codec::FramedRead;
+use tokio_util::compat::TokioAsyncReadCompatExt;
 use tokio_util::either::Either;
 
 use crate::utils::encoding::{from_slice_with_fallback, uvibytes::UviBytes};
@@ -137,6 +140,65 @@ impl<ReaderT: AsyncBufRead> Stream for CarStream<ReaderT> {
     }
 }
 
+pin_project! {
+    pub struct CarWriter {
+        #[pin]
+        sender: flume::Sender<(Cid, Vec<u8>)>,
+        pub inner: std::pin::Pin<Box<dyn std::future::Future<Output = std::io::Result<()>>>>,
+    }
+
+}
+
+impl CarWriter {
+    pub fn new_carv1(roots: Vec<Cid>, dst: tokio::fs::File) -> Self {
+        let (sender, receiver): (
+            flume::Sender<(Cid, Vec<u8>)>,
+            flume::Receiver<(Cid, Vec<u8>)>,
+        ) = flume::bounded(5);
+        let car_writer = FvmCarHeader::from(roots);
+
+        Self {
+            sender,
+            inner: Box::pin(async move {
+                let mut file = dst.compat();
+                let mut stream: RecvStream<'_, (Cid, Vec<u8>)> = receiver.stream();
+                car_writer
+                    .write_stream_async(&mut file, &mut stream)
+                    .await
+                    .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))
+            }),
+        }
+    }
+}
+
+impl Sink<Block> for CarWriter {
+    type Error = io::Error;
+
+    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        if !self.sender.is_full() {
+            return Poll::Ready(Ok(()));
+        } else {
+            if self.sender.is_disconnected() {
+                return Poll::Ready(Err(io::Error::new(io::ErrorKind::Other, "channel dropped")));
+            } else {
+                return Poll::Pending;
+            }
+        }
+    }
+    fn start_send(self: Pin<&mut Self>, item: Block) -> Result<(), Self::Error> {
+        self.sender
+            .try_send((item.cid, item.data))
+            .map_err(|e| io::Error::new(io::ErrorKind::Other, e))
+    }
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        //self.project().writer.poll_flush(self, cx)
+        todo!()
+    }
+    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        todo!()
+    }
+}
+
 async fn read_header<ReaderT: AsyncRead + Unpin>(reader: &mut ReaderT) -> Option<CarHeader> {
     let mut framed_reader = FramedRead::new(reader, UviBytes::default());
     let header = from_slice_with_fallback::<CarHeader>(&framed_reader.next().await?.ok()?).ok()?;