Lines
81.29 %
Functions
47.59 %
Branches
100 %
use std::{ops::Deref, sync::Arc};
use async_trait::async_trait;
use bonsaidb_core::{
connection::{AccessPolicy, Connection, QueryKey, Range, Sort},
custom_api::CustomApi,
document::OwnedDocument,
networking::{DatabaseRequest, DatabaseResponse, Request, Response},
schema::{
view::{
self,
map::{self, MappedDocuments},
SerializedView,
},
Collection, Key, Map, MappedValue, Schematic,
transaction::{Executed, OperationResult, Transaction},
};
use derive_where::derive_where;
use crate::Client;
mod pubsub;
pub use pubsub::*;
mod keyvalue;
/// A database on a remote server.
#[derive(Debug)]
#[derive_where(Clone)]
pub struct RemoteDatabase<A: CustomApi = ()> {
client: Client<A>,
name: Arc<String>,
schema: Arc<Schematic>,
}
impl<A: CustomApi> RemoteDatabase<A> {
/// Returns the name of the database.
#[must_use]
pub fn name(&self) -> &str {
self.name.as_ref()
impl<A: CustomApi> Deref for RemoteDatabase<A> {
type Target = Client<A>;
fn deref(&self) -> &Self::Target {
&self.client
pub(crate) fn new(client: Client<A>, name: String, schema: Arc<Schematic>) -> Self {
Self {
client,
name: Arc::new(name),
schema,
#[async_trait]
impl<A: CustomApi> Connection for RemoteDatabase<A> {
async fn get<C: Collection>(
&self,
id: u64,
) -> Result<Option<OwnedDocument>, bonsaidb_core::Error> {
match self
.client
.send_request(Request::Database {
database: self.name.to_string(),
request: DatabaseRequest::Get {
collection: C::collection_name(),
id,
})
.await?
{
Response::Database(DatabaseResponse::Documents(documents)) => {
Ok(documents.into_iter().next())
Response::Error(bonsaidb_core::Error::DocumentNotFound(_, _)) => Ok(None),
Response::Error(err) => Err(err),
other => Err(bonsaidb_core::Error::Networking(
bonsaidb_core::networking::Error::UnexpectedResponse(format!("{:?}", other)),
)),
async fn get_multiple<C: Collection>(
ids: &[u64],
) -> Result<Vec<OwnedDocument>, bonsaidb_core::Error> {
request: DatabaseRequest::GetMultiple {
ids: ids.to_vec(),
Response::Database(DatabaseResponse::Documents(documents)) => Ok(documents),
async fn list<C: Collection, R: Into<Range<u64>> + Send>(
ids: R,
order: Sort,
limit: Option<usize>,
request: DatabaseRequest::List {
ids: ids.into(),
order,
limit,
async fn query<V: SerializedView>(
key: Option<QueryKey<V::Key>>,
access_policy: AccessPolicy,
) -> Result<Vec<Map<V::Key, V::Value>>, bonsaidb_core::Error>
where
Self: Sized,
request: DatabaseRequest::Query {
view: self
.schema
.view::<V>()
.ok_or(bonsaidb_core::Error::CollectionNotFound)?
.view_name(),
key: key.map(|key| key.serialized()).transpose()?,
access_policy,
with_docs: false,
Response::Database(DatabaseResponse::ViewMappings(mappings)) => Ok(mappings
.iter()
.map(map::Serialized::deserialized::<V>)
.collect::<Result<Vec<_>, _>>()
.map_err(|err| bonsaidb_core::Error::Database(err.to_string()))?),
async fn query_with_docs<V: SerializedView>(
) -> Result<MappedDocuments<OwnedDocument, V>, bonsaidb_core::Error>
with_docs: true,
Response::Database(DatabaseResponse::ViewMappingsWithDocs(mappings)) => Ok(mappings
.deserialized::<V>()
async fn reduce<V: SerializedView>(
) -> Result<V::Value, bonsaidb_core::Error>
request: DatabaseRequest::Reduce {
grouped: false,
Response::Database(DatabaseResponse::ViewReduction(value)) => {
let value = V::deserialize(&value)?;
Ok(value)
async fn reduce_grouped<V: SerializedView>(
) -> Result<Vec<MappedValue<V::Key, V::Value>>, bonsaidb_core::Error>
grouped: true,
Response::Database(DatabaseResponse::ViewGroupedReduction(values)) => values
.into_iter()
.map(|map| {
Ok(MappedValue::new(
V::Key::from_big_endian_bytes(&map.key).map_err(|err| {
bonsaidb_core::Error::Database(
view::Error::key_serialization(err).to_string(),
)
})?,
V::deserialize(&map.value)?,
))
.collect::<Result<Vec<_>, bonsaidb_core::Error>>(),
async fn delete_docs<V: SerializedView>(
) -> Result<u64, bonsaidb_core::Error>
request: DatabaseRequest::DeleteDocs {
Response::Database(DatabaseResponse::DocumentsDeleted(count)) => Ok(count),
async fn apply_transaction(
transaction: Transaction,
) -> Result<Vec<OperationResult>, bonsaidb_core::Error> {
request: DatabaseRequest::ApplyTransaction { transaction },
Response::Database(DatabaseResponse::TransactionResults(results)) => Ok(results),
async fn list_executed_transactions(
starting_id: Option<u64>,
result_limit: Option<usize>,
) -> Result<Vec<Executed>, bonsaidb_core::Error> {
request: DatabaseRequest::ListExecutedTransactions {
starting_id,
result_limit,
Response::Database(DatabaseResponse::ExecutedTransactions(results)) => Ok(results),
async fn last_transaction_id(&self) -> Result<Option<u64>, bonsaidb_core::Error> {
request: DatabaseRequest::LastTransactionId,
Response::Database(DatabaseResponse::LastTransactionId(result)) => Ok(result),
async fn compact_collection<C: Collection>(&self) -> Result<(), bonsaidb_core::Error> {
request: DatabaseRequest::CompactCollection {
name: C::collection_name(),
Response::Ok => Ok(()),
async fn compact(&self) -> Result<(), bonsaidb_core::Error> {
request: DatabaseRequest::Compact,
async fn compact_key_value_store(&self) -> Result<(), bonsaidb_core::Error> {
request: DatabaseRequest::CompactKeyValueStore,