WIP

Cargo.toml

@@ -28,6 +28,7 @@ members = [
 # nebari = { path = "../nebari/nebari", version = "0.3" }
 # nebari = { git = "https://github.com/khonsulabs/nebari.git", branch = "main" }
 # arc-bytes = { path = "../shared-buffer" }
+circulate = { path = "../circulate" }
 
 # [patch."https://github.com/khonsulabs/custodian.git"]
 # custodian-password = { path = "../custodian/password" }

crates/bonsaidb-client/src/client.rs

@@ -17,6 +17,7 @@ use async_trait::async_trait;
 #[cfg(feature = "password-hashing")]
 use bonsaidb_core::connection::{Authenticated, Authentication};
 use bonsaidb_core::{
+    arc_bytes::OwnedBytes,
     connection::{Database, StorageConnection},
     custom_api::{CustomApi, CustomApiResult},
     networking::{
@@ -828,7 +829,7 @@ async fn process_response_payload<A: CustomApi>(
                     if sender
                         .send(std::sync::Arc::new(bonsaidb_core::circulate::Message {
                             topic,
-                            payload: payload.into_vec(),
+                            payload: OwnedBytes::from(payload.0),
                         }))
                         .is_err()
                     {

crates/bonsaidb-client/src/client/remote_database/pubsub.rs

@@ -2,7 +2,7 @@ use std::sync::Arc;
 
 use async_trait::async_trait;
 use bonsaidb_core::{
-    arc_bytes::serde::Bytes,
+    arc_bytes::OwnedBytes,
     circulate::Message,
     custom_api::CustomApi,
     networking::{DatabaseRequest, DatabaseResponse, Request, Response},
@@ -51,13 +51,31 @@ where
         payload: &P,
     ) -> Result<(), bonsaidb_core::Error> {
         let payload = pot::to_vec(&payload)?;
+        self.publish_raw(topic, payload).await
+    }
+
+    async fn publish_to_all<P: Serialize + Sync>(
+        &self,
+        topics: Vec<String>,
+        payload: &P,
+    ) -> Result<(), bonsaidb_core::Error> {
+        let payload = pot::to_vec(&payload)?;
+        self.publish_raw_to_all(topics, payload).await
+    }
+
+    async fn publish_raw<S: Into<String> + Send, P: Into<OwnedBytes> + Send>(
+        &self,
+        topic: S,
+        payload: P,
+    ) -> Result<(), bonsaidb_core::Error> {
+        let payload = payload.into();
         match self
             .client
             .send_request(Request::Database {
                 database: self.name.to_string(),
                 request: DatabaseRequest::Publish {
                     topic: topic.into(),
-                    payload: Bytes::from(payload),
+                    payload,
                 },
             })
             .await?
@@ -70,20 +88,17 @@ where
         }
     }
 
-    async fn publish_to_all<P: Serialize + Sync>(
+    async fn publish_raw_to_all<P: Into<OwnedBytes> + Send>(
         &self,
         topics: Vec<String>,
-        payload: &P,
+        payload: P,
     ) -> Result<(), bonsaidb_core::Error> {
-        let payload = pot::to_vec(&payload)?;
+        let payload = payload.into();
         match self
             .client
             .send_request(Request::Database {
                 database: self.name.to_string(),
-                request: DatabaseRequest::PublishToAll {
-                    topics,
-                    payload: Bytes::from(payload),
-                },
+                request: DatabaseRequest::PublishToAll { topics, payload },
             })
             .await?
         {

crates/bonsaidb-core/src/networking.rs

@@ -1,4 +1,4 @@
-use arc_bytes::serde::Bytes;
+use arc_bytes::{serde::Bytes, OwnedBytes};
 use derive_where::derive_where;
 use schema::SchemaName;
 use serde::{Deserialize, Serialize};
@@ -230,15 +230,15 @@ pub enum DatabaseRequest {
         /// The topics to publish to.
         topic: String,
         /// The payload to publish.
-        payload: Bytes,
+        payload: OwnedBytes,
     },
     /// Publishes `payload` to all subscribers of all `topics`.
     #[cfg_attr(feature = "actionable-traits", actionable(protection = "custom"))]
     PublishToAll {
         /// The topics to publish to.
         topics: Vec<String>,
         /// The payload to publish.
-        payload: Bytes,
+        payload: OwnedBytes,
     },
     /// Subscribes `subscriber_id` to messages for `topic`.
     #[cfg_attr(feature = "actionable-traits", actionable(protection = "simple"))]

crates/bonsaidb-core/src/pubsub.rs

@@ -1,5 +1,6 @@
 use std::sync::Arc;
 
+use arc_bytes::OwnedBytes;
 use async_trait::async_trait;
 use circulate::{flume, Message, Relay};
 use serde::Serialize;
@@ -22,12 +23,26 @@ pub trait PubSub {
         payload: &P,
     ) -> Result<(), Error>;
 
+    /// Publishes a `payload` without any extra processing to all subscribers of `topic`.
+    async fn publish_raw<S: Into<String> + Send, P: Into<OwnedBytes> + Send>(
+        &self,
+        topic: S,
+        payload: P,
+    ) -> Result<(), Error>;
+
     /// Publishes a `payload` to all subscribers of all `topics`.
     async fn publish_to_all<P: Serialize + Sync>(
         &self,
         topics: Vec<String>,
         payload: &P,
     ) -> Result<(), Error>;
+
+    /// Publishes a `payload` without any extra processing to all subscribers of `topic`.
+    async fn publish_raw_to_all<P: Into<OwnedBytes> + Send>(
+        &self,
+        topics: Vec<String>,
+        payload: P,
+    ) -> Result<(), Error>;
 }
 
 /// A subscriber to one or more topics.
@@ -69,6 +84,24 @@ impl PubSub for Relay {
         self.publish_to_all(topics, payload).await?;
         Ok(())
     }
+
+    async fn publish_raw<S: Into<String> + Send, P: Into<OwnedBytes> + Send>(
+        &self,
+        topic: S,
+        payload: P,
+    ) -> Result<(), Error> {
+        self.publish_raw(topic, payload).await;
+        Ok(())
+    }
+
+    async fn publish_raw_to_all<P: Into<OwnedBytes> + Send>(
+        &self,
+        topics: Vec<String>,
+        payload: P,
+    ) -> Result<(), Error> {
+        self.publish_raw_to_all(topics, payload).await;
+        Ok(())
+    }
 }
 
 #[async_trait]

crates/bonsaidb-jobs/Cargo.toml

@@ -15,6 +15,9 @@ rust-version = "1.58"
 [dependencies]
 bonsaidb-core = { version = "0.2.0", path = "../bonsaidb-core" }
 serde = { version = "1", features = ["derive"] }
+thiserror = "1"
+tokio = { version = "=1.16.1", default-features = false, features = ["sync"] }
+flume = "0.10"
 
 [dev-dependencies]
 tokio = { version = "=1.16.1", features = ["full"] }