Merge pull request #1667 from tursodatabase/libsql-wal-rpc-server

libsql-wal final integration

Author: MarinPostma

Cargo.lock

@@ -14,6 +14,12 @@ pub struct LibsqlInjector {
     injector: Injector<StdIO>,
 }
 
+impl LibsqlInjector {
+    pub fn new(injector: Injector<StdIO>) -> Self {
+        Self { injector }
+    }
+}
+
 impl super::Injector for LibsqlInjector {
     async fn inject_frame(&mut self, frame: RpcFrame) -> Result<Option<FrameNo>> {
         // this is a bit annoying be we want to read the frame, and it has to be aligned, so we
@@ -24,6 +30,7 @@ impl super::Injector for LibsqlInjector {
             todo!("invalid frame");
         }
         wal_frame.as_bytes_mut().copy_from_slice(&frame.data[..]);
+
         Ok(self
             .injector
             .insert_frame(wal_frame)

libsql-replication/src/injector/libsql_injector.rs

@@ -23,9 +23,14 @@ pub mod proxy {
 
 pub mod replication {
     #![allow(clippy::all)]
+    use std::pin::Pin;
 
+    use tokio_stream::Stream;
     use uuid::Uuid;
 
+    pub type BoxStream<'a, T> = Pin<Box<dyn Stream<Item = T> + Send + 'a>>;
+
+    use self::replication_log_server::ReplicationLog;
     include!("generated/wal_log.rs");
 
     pub const NO_HELLO_ERROR_MSG: &str = "NO_HELLO";
@@ -53,6 +58,47 @@ pub mod replication {
             }
         }
     }
+
+    pub type BoxReplicationService = Box<
+        dyn ReplicationLog<
+            LogEntriesStream = BoxStream<'static, Result<Frame, tonic::Status>>,
+            SnapshotStream = BoxStream<'static, Result<Frame, tonic::Status>>,
+        >,
+    >;
+
+    #[tonic::async_trait]
+    impl ReplicationLog for BoxReplicationService {
+        type LogEntriesStream = BoxStream<'static, Result<Frame, tonic::Status>>;
+        type SnapshotStream = BoxStream<'static, Result<Frame, tonic::Status>>;
+
+        async fn log_entries(
+            &self,
+            req: tonic::Request<LogOffset>,
+        ) -> Result<tonic::Response<Self::LogEntriesStream>, tonic::Status> {
+            self.as_ref().log_entries(req).await
+        }
+
+        async fn batch_log_entries(
+            &self,
+            req: tonic::Request<LogOffset>,
+        ) -> Result<tonic::Response<Frames>, tonic::Status> {
+            self.as_ref().batch_log_entries(req).await
+        }
+
+        async fn hello(
+            &self,
+            req: tonic::Request<HelloRequest>,
+        ) -> Result<tonic::Response<HelloResponse>, tonic::Status> {
+            self.as_ref().hello(req).await
+        }
+
+        async fn snapshot(
+            &self,
+            req: tonic::Request<LogOffset>,
+        ) -> Result<tonic::Response<Self::SnapshotStream>, tonic::Status> {
+            self.as_ref().snapshot(req).await
+        }
+    }
 }
 
 pub mod metadata {

libsql-replication/src/rpc.rs

@@ -37,7 +37,7 @@ hyper-tungstenite = "0.11"
 itertools = "0.10.5"
 jsonwebtoken = "9"
 libsql = { path = "../libsql/", optional = true }
-libsql_replication = { path = "../libsql-replication" }
+libsql_replication = { path = "../libsql-replication", features = ["libsql_wal"] }
 libsql-wal = { path = "../libsql-wal/" }
 libsql-storage = { path = "../libsql-storage", optional = true }
 metrics = "0.21.1"
@@ -92,11 +92,12 @@ async-recursion = "1"
 mimalloc = "0.1.42"
 rheaper = { git = "https://github.com/MarinPostma/rheaper.git", tag = "v0.2.0", default-features = false, features = ["allocator"] }
 tar = "0.4.41"
+aws-config = "1"
+aws-sdk-s3 = "1"
+aws-smithy-runtime = "1.6.2"
 
 [dev-dependencies]
 arbitrary = { version = "1.3.0", features = ["derive_arbitrary"] }
-aws-config = "0.55"
-aws-sdk-s3 = "0.28"
 env_logger = "0.10"
 hyper = { workspace = true, features = ["client"] }
 insta = { version = "1.26.0", features = ["json"] }