1
use std::sync::Arc;
2

            
3
use async_trait::async_trait;
4
use bonsaidb_core::connection::{
5
    self, AccessPolicy, AsyncConnection, AsyncLowLevelConnection, AsyncStorageConnection,
6
    Connection, HasSchema, HasSession, IdentityReference, LowLevelConnection, Range,
7
    SerializedQueryKey, Session, Sort, StorageConnection,
8
};
9
use bonsaidb_core::document::{DocumentId, Header, OwnedDocument};
10
use bonsaidb_core::keyvalue::{AsyncKeyValue, KeyOperation, KeyValue, Output};
11
use bonsaidb_core::permissions::Permissions;
12
use bonsaidb_core::pubsub::{self, AsyncPubSub, AsyncSubscriber, PubSub, Receiver};
13
use bonsaidb_core::schema::view::map::MappedSerializedValue;
14
use bonsaidb_core::schema::{
15
    self, CollectionName, Nameable, Schema, SchemaName, SchemaSummary, Schematic, ViewName,
16
};
17
use bonsaidb_core::transaction::{self, OperationResult, Transaction};
18

            
19
use crate::config::StorageConfiguration;
20
use crate::database::DatabaseNonBlocking;
21
use crate::storage::{AnyBackupLocation, StorageNonBlocking};
22
use crate::{Database, Error, Storage, Subscriber};
23

            
24
/// A file-based, multi-database, multi-user database engine. This type is
25
/// designed for use with [Tokio](https://tokio.rs). For blocking
26
/// (non-asynchronous) code, see the [`Storage`] type instead.
27
///
28
/// ## Converting between Blocking and Async Types
29
///
30
/// [`AsyncDatabase`] and [`Database`] can be converted to and from each other
31
/// using:
32
///
33
/// - [`AsyncStorage::into_blocking()`]
34
/// - [`AsyncStorage::to_blocking()`]
35
/// - [`AsyncStorage::as_blocking()`]
36
/// - [`Storage::into_async()`]
37
/// - [`Storage::to_async()`]
38
/// - [`Storage::into_async_with_runtime()`]
39
/// - [`Storage::to_async_with_runtime()`]
40
///
41
/// ## Converting from `AsyncDatabase::open` to `AsyncStorage::open`
42
///
43
/// [`AsyncDatabase::open`](AsyncDatabase::open) is a simple method that uses
44
/// `AsyncStorage` to create a database named `default` with the schema
45
/// provided. These two ways of opening the database are the same:
46
///
47
/// ```rust
48
/// // `bonsaidb_core` is re-exported to `bonsaidb::core` or `bonsaidb_local::core`.
49
/// use bonsaidb_core::connection::AsyncStorageConnection;
50
/// use bonsaidb_core::schema::Schema;
51
/// // `bonsaidb_local` is re-exported to `bonsaidb::local` if using the omnibus crate.
52
/// use bonsaidb_local::{
53
///     config::{Builder, StorageConfiguration},
54
///     AsyncDatabase, AsyncStorage,
55
/// };
56
/// # async fn open<MySchema: Schema>() -> anyhow::Result<()> {
57
/// // This creates a Storage instance, creates a database, and returns it.
58
/// let db = AsyncDatabase::open::<MySchema>(StorageConfiguration::new("my-db.bonsaidb")).await?;
59
///
60
/// // This is the equivalent code being executed:
61
/// let storage =
62
///     AsyncStorage::open(StorageConfiguration::new("my-db.bonsaidb").with_schema::<MySchema>()?)
63
///         .await?;
64
/// storage.create_database::<MySchema>("default", true).await?;
65
/// let db = storage.database::<MySchema>("default").await?;
66
/// #     Ok(())
67
/// # }
68
/// ```
69
///
70
/// ## Using multiple databases
71
///
72
/// This example shows how to use `AsyncStorage` to create and use multiple
73
/// databases with multiple schemas:
74
///
75
/// ```rust
76
/// use bonsaidb_core::connection::AsyncStorageConnection;
77
/// use bonsaidb_core::schema::{Collection, Schema};
78
/// use bonsaidb_local::config::{Builder, StorageConfiguration};
79
/// use bonsaidb_local::AsyncStorage;
80
/// use serde::{Deserialize, Serialize};
81
///
82
/// #[derive(Debug, Schema)]
83
/// #[schema(name = "my-schema", collections = [BlogPost, Author])]
84
/// # #[schema(core = bonsaidb_core)]
85
/// struct MySchema;
86
///
87
/// #[derive(Debug, Serialize, Deserialize, Collection)]
88
/// #[collection(name = "blog-posts")]
89
/// # #[collection(core = bonsaidb_core)]
90
/// struct BlogPost {
91
///     pub title: String,
92
///     pub contents: String,
93
///     pub author_id: u64,
94
/// }
95
///
96
/// #[derive(Debug, Serialize, Deserialize, Collection)]
97
/// #[collection(name = "blog-posts")]
98
/// # #[collection(core = bonsaidb_core)]
99
/// struct Author {
100
///     pub name: String,
101
/// }
102
///
103
/// # async fn test_fn() -> Result<(), bonsaidb_core::Error> {
104
/// let storage = AsyncStorage::open(
105
///     StorageConfiguration::new("my-db.bonsaidb")
106
///         .with_schema::<BlogPost>()?
107
///         .with_schema::<MySchema>()?,
108
/// )
109
/// .await?;
110
///
111
/// storage
112
///     .create_database::<BlogPost>("ectons-blog", true)
113
///     .await?;
114
/// let ectons_blog = storage.database::<BlogPost>("ectons-blog").await?;
115
/// storage
116
///     .create_database::<MySchema>("another-db", true)
117
///     .await?;
118
/// let another_db = storage.database::<MySchema>("another-db").await?;
119
///
120
/// #     Ok(())
121
/// # }
122
/// ```
123
194161
#[derive(Debug, Clone)]
124
#[must_use]
125
pub struct AsyncStorage {
126
    pub(crate) storage: Storage,
127
    pub(crate) runtime: Arc<tokio::runtime::Handle>,
128
}
129

            
130
impl AsyncStorage {
131
    /// Creates or opens a multi-database [`AsyncStorage`] with its data stored in `directory`.
132
3490
    pub async fn open(configuration: StorageConfiguration) -> Result<Self, Error> {
133
165
        tokio::task::spawn_blocking(move || Storage::open(configuration))
134
165
            .await?
135
165
            .map(Storage::into_async)
136
165
    }
137

            
138
    /// Restores all data from a previously stored backup `location`.
139
1
    pub async fn restore<L: AnyBackupLocation + 'static>(&self, location: L) -> Result<(), Error> {
140
1
        let task_self = self.clone();
141
1
        self.runtime
142
1
            .spawn_blocking(move || task_self.storage.restore(&location))
143
1
            .await?
144
1
    }
145

            
146
    /// Stores a copy of all data in this instance to `location`.
147
1
    pub async fn backup<L: AnyBackupLocation + 'static>(&self, location: L) -> Result<(), Error> {
148
1
        let task_self = self.clone();
149
1
        self.runtime
150
1
            .spawn_blocking(move || task_self.storage.backup(&location))
151
1
            .await?
152
1
    }
153

            
154
    /// Restricts an unauthenticated instance to having `effective_permissions`.
155
    /// Returns `None` if a session has already been established.
156
    #[must_use]
157
    pub fn with_effective_permissions(&self, effective_permissions: Permissions) -> Option<Self> {
158
        self.storage
159
            .with_effective_permissions(effective_permissions)
160
            .map(|storage| Self {
161
                storage,
162
                runtime: self.runtime.clone(),
163
            })
164
    }
165

            
166
    #[cfg(feature = "internal-apis")]
167
    #[doc(hidden)]
168
2548296
    pub async fn database_without_schema(&self, name: &str) -> Result<AsyncDatabase, Error> {
169
70786
        let name = name.to_owned();
170
70786
        let task_self = self.clone();
171
70786
        self.runtime
172
70786
            .spawn_blocking(move || {
173
70786
                task_self
174
70786
                    .storage
175
70786
                    .database_without_schema(&name)
176
70786
                    .map(Database::into_async)
177
70786
            })
178
66141
            .await?
179
70786
    }
180

            
181
    /// Converts this instance into its blocking version, which is able to be
182
    /// used without async.
183
    pub fn into_blocking(self) -> Storage {
184
        self.storage
185
    }
186

            
187
    /// Converts this instance into its blocking version, which is able to be
188
    /// used without async.
189
    pub fn to_blocking(&self) -> Storage {
190
        self.storage.clone()
191
    }
192

            
193
    /// Returns a reference to this instance's blocking version, which is able
194
    /// to be used without async.
195
    pub fn as_blocking(&self) -> &Storage {
196
        &self.storage
197
    }
198
}
199

            
200
impl<'a> From<&'a AsyncStorage> for Storage {
201
    fn from(storage: &'a AsyncStorage) -> Self {
202
        storage.to_blocking()
203
    }
204
}
205

            
206
impl From<AsyncStorage> for Storage {
207
    fn from(storage: AsyncStorage) -> Self {
208
        storage.into_blocking()
209
    }
210
}
211

            
212
impl StorageNonBlocking for AsyncStorage {
213
6048
    fn path(&self) -> &std::path::Path {
214
6048
        self.storage.path()
215
6048
    }
216

            
217
2593908
    fn assume_session(&self, session: Session) -> Result<Self, bonsaidb_core::Error> {
218
2593908
        self.storage.assume_session(session).map(|storage| Self {
219
2593764
            storage,
220
2593764
            runtime: self.runtime.clone(),
221
2593908
        })
222
2593908
    }
223
}
224

            
225
/// A database stored in BonsaiDb. This type is designed for use with
226
/// [Tokio](https://tokio.rs). For blocking (non-asynchronous) code, see the
227
/// [`Database`] type instead.
228
///
229
/// ## Converting between Async and Blocking Types
230
///
231
/// [`AsyncDatabase`] and [`Database`] can be converted to and from each other
232
/// using:
233
///
234
/// - [`AsyncDatabase::into_blocking()`]
235
/// - [`AsyncDatabase::to_blocking()`]
236
/// - [`AsyncDatabase::as_blocking()`]
237
/// - [`Database::into_async()`]
238
/// - [`Database::to_async()`]
239
/// - [`Database::into_async_with_runtime()`]
240
/// - [`Database::to_async_with_runtime()`]
241
///
242
/// ## Using `Database` to create a single database
243
///
244
/// `Database`provides an easy mechanism to open and access a single database:
245
///
246
/// ```rust
247
/// // `bonsaidb_core` is re-exported to `bonsaidb::core` or `bonsaidb_local::core`.
248
/// use bonsaidb_core::schema::Collection;
249
/// // `bonsaidb_local` is re-exported to `bonsaidb::local` if using the omnibus crate.
250
/// use bonsaidb_local::{
251
///     config::{Builder, StorageConfiguration},
252
///     AsyncDatabase,
253
/// };
254
/// use serde::{Deserialize, Serialize};
255
///
256
/// #[derive(Debug, Serialize, Deserialize, Collection)]
257
/// #[collection(name = "blog-posts")]
258
/// # #[collection(core = bonsaidb_core)]
259
/// struct BlogPost {
260
///     pub title: String,
261
///     pub contents: String,
262
/// }
263
///
264
/// # async fn test_fn() -> Result<(), bonsaidb_core::Error> {
265
/// let db = AsyncDatabase::open::<BlogPost>(StorageConfiguration::new("my-db.bonsaidb")).await?;
266
/// #     Ok(())
267
/// # }
268
/// ```
269
///
270
/// Under the hood, this initializes a [`AsyncStorage`] instance pointing at
271
/// "./my-db.bonsaidb". It then returns (or creates) a database named "default"
272
/// with the schema `BlogPost`.
273
///
274
/// In this example, `BlogPost` implements the
275
/// [`Collection`](schema::Collection) trait, and all collections can be used as
276
/// a [`Schema`].
277
4057166
#[derive(Debug, Clone)]
278
pub struct AsyncDatabase {
279
    pub(crate) database: Database,
280
    pub(crate) runtime: Arc<tokio::runtime::Handle>,
281
}
282

            
283
impl AsyncDatabase {
284
    /// Creates a `Storage` with a single-database named "default" with its data stored at `path`.
285
12
    pub async fn open<DB: Schema>(configuration: StorageConfiguration) -> Result<Self, Error> {
286
12
        tokio::task::spawn_blocking(move || {
287
12
            Database::open::<DB>(configuration).map(Database::into_async)
288
12
        })
289
12
        .await?
290
12
    }
291

            
292
    /// Restricts an unauthenticated instance to having `effective_permissions`.
293
    /// Returns `None` if a session has already been established.
294
    #[must_use]
295
    pub fn with_effective_permissions(&self, effective_permissions: Permissions) -> Option<Self> {
296
        self.database
297
            .with_effective_permissions(effective_permissions)
298
            .map(|database| Self {
299
                database,
300
                runtime: self.runtime.clone(),
301
            })
302
    }
303

            
304
    /// Converts this instance into its blocking version, which is able to be
305
    /// used without async.
306
    #[must_use]
307
    pub fn into_blocking(self) -> Database {
308
        self.database
309
    }
310

            
311
    /// Converts this instance into its blocking version, which is able to be
312
    /// used without async.
313
    #[must_use]
314
    pub fn to_blocking(&self) -> Database {
315
        self.database.clone()
316
    }
317

            
318
    /// Returns a reference to this instance's blocking version, which is able
319
    /// to be used without async.
320
    #[must_use]
321
    pub fn as_blocking(&self) -> &Database {
322
        &self.database
323
    }
324
}
325

            
326
impl From<AsyncDatabase> for Database {
327
    fn from(database: AsyncDatabase) -> Self {
328
        database.into_blocking()
329
    }
330
}
331

            
332
impl<'a> From<&'a AsyncDatabase> for Database {
333
    fn from(database: &'a AsyncDatabase) -> Self {
334
        database.to_blocking()
335
    }
336
}
337

            
338
impl DatabaseNonBlocking for AsyncDatabase {
339
    fn name(&self) -> &str {
340
        self.database.name()
341
    }
342
}
343

            
344
impl HasSession for AsyncStorage {
345
3932
    fn session(&self) -> Option<&Session> {
346
3932
        self.storage.session()
347
3932
    }
348
}
349

            
350
#[async_trait]
351
impl AsyncStorageConnection for AsyncStorage {
352
    type Authenticated = Self;
353
    type Database = AsyncDatabase;
354

            
355
72
    async fn admin(&self) -> Self::Database {
356
72
        let task_self = self.clone();
357
72

            
358
72
        self.runtime
359
72
            .spawn_blocking(move || task_self.storage.admin())
360
72
            .await
361
72
            .unwrap()
362
72
            .into_async()
363
216
    }
364

            
365
29024
    async fn create_database_with_schema(
366
29024
        &self,
367
29024
        name: &str,
368
29024
        schema: SchemaName,
369
29024
        only_if_needed: bool,
370
29024
    ) -> Result<(), bonsaidb_core::Error> {
371
29024
        let task_self = self.clone();
372
29024
        let name = name.to_owned();
373
29024
        self.runtime
374
29024
            .spawn_blocking(move || {
375
29024
                StorageConnection::create_database_with_schema(
376
29024
                    &task_self.storage,
377
29024
                    &name,
378
29024
                    schema,
379
29024
                    only_if_needed,
380
29024
                )
381
29024
            })
382
28988
            .await
383
29024
            .map_err(Error::from)?
384
87072
    }
385

            
386
673
    async fn database<DB: Schema>(
387
673
        &self,
388
673
        name: &str,
389
673
    ) -> Result<Self::Database, bonsaidb_core::Error> {
390
673
        let task_self = self.clone();
391
673
        let name = name.to_owned();
392
673
        self.runtime
393
673
            .spawn_blocking(move || {
394
673
                task_self
395
673
                    .storage
396
673
                    .database::<DB>(&name)
397
673
                    .map(Database::into_async)
398
673
            })
399
673
            .await
400
673
            .map_err(Error::from)?
401
2019
    }
402

            
403
18292
    async fn delete_database(&self, name: &str) -> Result<(), bonsaidb_core::Error> {
404
18292
        let task_self = self.clone();
405
18292
        let name = name.to_owned();
406
18292
        self.runtime
407
18292
            .spawn_blocking(move || task_self.storage.delete_database(&name))
408
18292
            .await
409
18292
            .map_err(Error::from)?
410
54876
    }
411

            
412
146
    async fn list_databases(&self) -> Result<Vec<connection::Database>, bonsaidb_core::Error> {
413
146
        let task_self = self.clone();
414
146
        self.runtime
415
146
            .spawn_blocking(move || task_self.storage.list_databases())
416
146
            .await
417
146
            .map_err(Error::from)?
418
438
    }
419

            
420
146
    async fn list_available_schemas(&self) -> Result<Vec<SchemaSummary>, bonsaidb_core::Error> {
421
146
        let task_self = self.clone();
422
146
        self.runtime
423
146
            .spawn_blocking(move || task_self.storage.list_available_schemas())
424
146
            .await
425
146
            .map_err(Error::from)?
426
438
    }
427

            
428
616
    async fn create_user(&self, username: &str) -> Result<u64, bonsaidb_core::Error> {
429
616
        let task_self = self.clone();
430
616
        let username = username.to_owned();
431
616
        self.runtime
432
616
            .spawn_blocking(move || task_self.storage.create_user(&username))
433
580
            .await
434
616
            .map_err(Error::from)?
435
1848
    }
436

            
437
6
    async fn delete_user<'user, U: Nameable<'user, u64> + Send + Sync>(
438
6
        &self,
439
6
        user: U,
440
6
    ) -> Result<(), bonsaidb_core::Error> {
441
6
        let task_self = self.clone();
442
6
        let user = user.name()?.into_owned();
443
6
        self.runtime
444
6
            .spawn_blocking(move || task_self.storage.delete_user(user))
445
6
            .await
446
6
            .map_err(Error::from)?
447
18
    }
448

            
449
    #[cfg(feature = "password-hashing")]
450
5
    async fn set_user_password<'user, U: Nameable<'user, u64> + Send + Sync>(
451
5
        &self,
452
5
        user: U,
453
5
        password: bonsaidb_core::connection::SensitiveString,
454
5
    ) -> Result<(), bonsaidb_core::Error> {
455
5
        let task_self = self.clone();
456
5
        let user = user.name()?.into_owned();
457
5
        self.runtime
458
5
            .spawn_blocking(move || task_self.storage.set_user_password(user, password))
459
5
            .await
460
5
            .map_err(Error::from)?
461
15
    }
462

            
463
    #[cfg(any(feature = "token-authentication", feature = "password-hashing"))]
464
836
    async fn authenticate(
465
836
        &self,
466
836
        authentication: bonsaidb_core::connection::Authentication,
467
836
    ) -> Result<Self, bonsaidb_core::Error> {
468
836
        let task_self = self.clone();
469
836
        self.runtime
470
836
            .spawn_blocking(move || {
471
836
                task_self
472
836
                    .storage
473
836
                    .authenticate(authentication)
474
836
                    .map(Storage::into_async)
475
836
            })
476
728
            .await
477
836
            .map_err(Error::from)?
478
2508
    }
479

            
480
72
    async fn assume_identity(
481
72
        &self,
482
72
        identity: IdentityReference<'_>,
483
72
    ) -> Result<Self::Authenticated, bonsaidb_core::Error> {
484
72
        let task_self = self.clone();
485
72
        let identity = identity.into_owned();
486
72
        self.runtime
487
72
            .spawn_blocking(move || {
488
72
                task_self
489
72
                    .storage
490
72
                    .assume_identity(identity)
491
72
                    .map(Storage::into_async)
492
72
            })
493
72
            .await
494
72
            .map_err(Error::from)?
495
216
    }
496

            
497
16
    async fn add_permission_group_to_user<
498
16
        'user,
499
16
        'group,
500
16
        U: Nameable<'user, u64> + Send + Sync,
501
16
        G: Nameable<'group, u64> + Send + Sync,
502
16
    >(
503
16
        &self,
504
16
        user: U,
505
16
        permission_group: G,
506
16
    ) -> Result<(), bonsaidb_core::Error> {
507
16
        let task_self = self.clone();
508
16
        let user = user.name()?.into_owned();
509
16
        let group = permission_group.name()?.into_owned();
510
16
        self.runtime
511
16
            .spawn_blocking(move || task_self.storage.add_permission_group_to_user(user, group))
512
16
            .await
513
16
            .map_err(Error::from)?
514
48
    }
515

            
516
12
    async fn remove_permission_group_from_user<
517
12
        'user,
518
12
        'group,
519
12
        U: Nameable<'user, u64> + Send + Sync,
520
12
        G: Nameable<'group, u64> + Send + Sync,
521
12
    >(
522
12
        &self,
523
12
        user: U,
524
12
        permission_group: G,
525
12
    ) -> Result<(), bonsaidb_core::Error> {
526
12
        let task_self = self.clone();
527
12
        let user = user.name()?.into_owned();
528
12
        let group = permission_group.name()?.into_owned();
529
12
        self.runtime
530
12
            .spawn_blocking(move || {
531
12
                task_self
532
12
                    .storage
533
12
                    .remove_permission_group_from_user(user, group)
534
12
            })
535
11
            .await
536
12
            .map_err(Error::from)?
537
36
    }
538

            
539
12
    async fn add_role_to_user<
540
12
        'user,
541
12
        'group,
542
12
        U: Nameable<'user, u64> + Send + Sync,
543
12
        G: Nameable<'group, u64> + Send + Sync,
544
12
    >(
545
12
        &self,
546
12
        user: U,
547
12
        role: G,
548
12
    ) -> Result<(), bonsaidb_core::Error> {
549
12
        let task_self = self.clone();
550
12
        let user = user.name()?.into_owned();
551
12
        let role = role.name()?.into_owned();
552
12
        self.runtime
553
12
            .spawn_blocking(move || task_self.storage.add_role_to_user(user, role))
554
12
            .await
555
12
            .map_err(Error::from)?
556
36
    }
557

            
558
12
    async fn remove_role_from_user<
559
12
        'user,
560
12
        'group,
561
12
        U: Nameable<'user, u64> + Send + Sync,
562
12
        G: Nameable<'group, u64> + Send + Sync,
563
12
    >(
564
12
        &self,
565
12
        user: U,
566
12
        role: G,
567
12
    ) -> Result<(), bonsaidb_core::Error> {
568
12
        let task_self = self.clone();
569
12
        let user = user.name()?.into_owned();
570
12
        let role = role.name()?.into_owned();
571
12
        self.runtime
572
12
            .spawn_blocking(move || task_self.storage.remove_role_from_user(user, role))
573
12
            .await
574
12
            .map_err(Error::from)?
575
36
    }
576
}
577

            
578
impl HasSession for AsyncDatabase {
579
    fn session(&self) -> Option<&Session> {
580
        self.database.session()
581
    }
582
}
583

            
584
#[async_trait]
585
impl AsyncConnection for AsyncDatabase {
586
    type Storage = AsyncStorage;
587

            
588
    fn storage(&self) -> Self::Storage {
589
        AsyncStorage {
590
            storage: self.database.storage(),
591
            runtime: self.runtime.clone(),
592
        }
593
    }
594

            
595
38700
    async fn list_executed_transactions(
596
38700
        &self,
597
38700
        starting_id: Option<u64>,
598
38700
        result_limit: Option<u32>,
599
38700
    ) -> Result<Vec<transaction::Executed>, bonsaidb_core::Error> {
600
38700
        let task_self = self.clone();
601
38700
        self.runtime
602
38700
            .spawn_blocking(move || {
603
38700
                task_self
604
38700
                    .database
605
38700
                    .list_executed_transactions(starting_id, result_limit)
606
38700
            })
607
35604
            .await
608
38700
            .map_err(Error::from)?
609
116100
    }
610

            
611
146
    async fn last_transaction_id(&self) -> Result<Option<u64>, bonsaidb_core::Error> {
612
146
        Ok(self
613
146
            .database
614
146
            .roots()
615
146
            .transactions()
616
146
            .current_transaction_id())
617
292
    }
618

            
619
146
    async fn compact(&self) -> Result<(), bonsaidb_core::Error> {
620
146
        let task_self = self.clone();
621
146
        self.runtime
622
146
            .spawn_blocking(move || Connection::compact(&task_self.database))
623
146
            .await
624
146
            .map_err(Error::from)?
625
438
    }
626

            
627
3
    async fn compact_collection<C: schema::Collection>(&self) -> Result<(), bonsaidb_core::Error> {
628
3
        let task_self = self.clone();
629
3
        self.runtime
630
3
            .spawn_blocking(move || Connection::compact_collection::<C>(&task_self.database))
631
3
            .await
632
3
            .map_err(Error::from)?
633
9
    }
634

            
635
146
    async fn compact_key_value_store(&self) -> Result<(), bonsaidb_core::Error> {
636
146
        let task_self = self.clone();
637
146
        self.runtime
638
146
            .spawn_blocking(move || Connection::compact_key_value_store(&task_self.database))
639
146
            .await
640
146
            .map_err(Error::from)?
641
438
    }
642
}
643

            
644
#[async_trait]
645
impl AsyncKeyValue for AsyncDatabase {
646
1476218
    async fn execute_key_operation(
647
1476218
        &self,
648
1476218
        op: KeyOperation,
649
1476218
    ) -> Result<Output, bonsaidb_core::Error> {
650
1476218
        let task_self = self.clone();
651
1476218
        self.runtime
652
1476218
            .spawn_blocking(move || KeyValue::execute_key_operation(&task_self.database, op))
653
1265470
            .await
654
1476218
            .map_err(Error::from)?
655
4428654
    }
656
}
657

            
658
#[async_trait]
659
impl AsyncPubSub for AsyncDatabase {
660
    type Subscriber = Subscriber;
661

            
662
1240
    async fn create_subscriber(&self) -> Result<Self::Subscriber, bonsaidb_core::Error> {
663
1240
        PubSub::create_subscriber(&self.database)
664
2480
    }
665

            
666
1820
    async fn publish_bytes(
667
1820
        &self,
668
1820
        topic: Vec<u8>,
669
1820
        payload: Vec<u8>,
670
1820
    ) -> Result<(), bonsaidb_core::Error> {
671
1820
        PubSub::publish_bytes(&self.database, topic, payload)
672
3640
    }
673

            
674
6
    async fn publish_bytes_to_all(
675
6
        &self,
676
6
        topics: impl IntoIterator<Item = Vec<u8>> + Send + 'async_trait,
677
6
        payload: Vec<u8>,
678
6
    ) -> Result<(), bonsaidb_core::Error> {
679
6
        PubSub::publish_bytes_to_all(&self.database, topics, payload)
680
12
    }
681
}
682

            
683
#[async_trait]
684
impl AsyncSubscriber for Subscriber {
685
566
    async fn subscribe_to_bytes(&self, topic: Vec<u8>) -> Result<(), bonsaidb_core::Error> {
686
566
        pubsub::Subscriber::subscribe_to_bytes(self, topic)
687
1132
    }
688

            
689
38
    async fn unsubscribe_from_bytes(&self, topic: &[u8]) -> Result<(), bonsaidb_core::Error> {
690
38
        pubsub::Subscriber::unsubscribe_from_bytes(self, topic)
691
76
    }
692

            
693
1044
    fn receiver(&self) -> &Receiver {
694
1044
        pubsub::Subscriber::receiver(self)
695
1044
    }
696
}
697

            
698
impl HasSchema for AsyncDatabase {
699
512304
    fn schematic(&self) -> &Schematic {
700
512304
        self.database.schematic()
701
512304
    }
702
}
703

            
704
#[async_trait]
705
impl AsyncLowLevelConnection for AsyncDatabase {
706
610140
    async fn apply_transaction(
707
610140
        &self,
708
610140
        transaction: Transaction,
709
610140
    ) -> Result<Vec<OperationResult>, bonsaidb_core::Error> {
710
610140
        let task_self = self.clone();
711
610140
        self.runtime
712
610140
            .spawn_blocking(move || task_self.database.apply_transaction(transaction))
713
609528
            .await
714
610140
            .map_err(Error::from)?
715
1830420
    }
716

            
717
575640
    async fn get_from_collection(
718
575640
        &self,
719
575640
        id: DocumentId,
720
575640
        collection: &CollectionName,
721
575640
    ) -> Result<Option<OwnedDocument>, bonsaidb_core::Error> {
722
575640
        let task_self = self.clone();
723
575640
        let collection = collection.clone();
724
575640
        self.runtime
725
575640
            .spawn_blocking(move || task_self.database.get_from_collection(id, &collection))
726
544320
            .await
727
575640
            .map_err(Error::from)?
728
1726920
    }
729

            
730
584
    async fn list_from_collection(
731
584
        &self,
732
584
        ids: Range<DocumentId>,
733
584
        order: Sort,
734
584
        limit: Option<u32>,
735
584
        collection: &CollectionName,
736
584
    ) -> Result<Vec<OwnedDocument>, bonsaidb_core::Error> {
737
584
        let task_self = self.clone();
738
584
        let collection = collection.clone();
739
584
        self.runtime
740
584
            .spawn_blocking(move || {
741
584
                task_self
742
584
                    .database
743
584
                    .list_from_collection(ids, order, limit, &collection)
744
584
            })
745
584
            .await
746
584
            .map_err(Error::from)?
747
1752
    }
748

            
749
146
    async fn list_headers_from_collection(
750
146
        &self,
751
146
        ids: Range<DocumentId>,
752
146
        order: Sort,
753
146
        limit: Option<u32>,
754
146
        collection: &CollectionName,
755
146
    ) -> Result<Vec<Header>, bonsaidb_core::Error> {
756
146
        let task_self = self.clone();
757
146
        let collection = collection.clone();
758
146
        self.runtime
759
146
            .spawn_blocking(move || {
760
146
                task_self
761
146
                    .database
762
146
                    .list_headers_from_collection(ids, order, limit, &collection)
763
146
            })
764
146
            .await
765
146
            .map_err(Error::from)?
766
438
    }
767

            
768
292
    async fn count_from_collection(
769
292
        &self,
770
292
        ids: Range<DocumentId>,
771
292
        collection: &CollectionName,
772
292
    ) -> Result<u64, bonsaidb_core::Error> {
773
292
        let task_self = self.clone();
774
292
        let collection = collection.clone();
775
292
        self.runtime
776
292
            .spawn_blocking(move || task_self.database.count_from_collection(ids, &collection))
777
292
            .await
778
292
            .map_err(Error::from)?
779
876
    }
780

            
781
340360
    async fn get_multiple_from_collection(
782
340360
        &self,
783
340360
        ids: &[DocumentId],
784
340360
        collection: &CollectionName,
785
340360
    ) -> Result<Vec<OwnedDocument>, bonsaidb_core::Error> {
786
340360
        let task_self = self.clone();
787
340360
        // TODO avoid the allocation here, switch to IntoIterator.
788
340360
        let ids = ids.to_vec();
789
340360
        let collection = collection.clone();
790
340360
        self.runtime
791
340360
            .spawn_blocking(move || {
792
340360
                task_self
793
340360
                    .database
794
340360
                    .get_multiple_from_collection(&ids, &collection)
795
340360
            })
796
325204
            .await
797
340360
            .map_err(Error::from)?
798
1021080
    }
799

            
800
108
    async fn compact_collection_by_name(
801
108
        &self,
802
108
        collection: CollectionName,
803
108
    ) -> Result<(), bonsaidb_core::Error> {
804
108
        let task_self = self.clone();
805
108
        self.runtime
806
108
            .spawn_blocking(move || task_self.database.compact_collection_by_name(collection))
807
108
            .await
808
108
            .map_err(Error::from)?
809
324
    }
810

            
811
350320
    async fn query_by_name(
812
350320
        &self,
813
350320
        view: &ViewName,
814
350320
        key: Option<SerializedQueryKey>,
815
350320
        order: Sort,
816
350320
        limit: Option<u32>,
817
350320
        access_policy: AccessPolicy,
818
350320
    ) -> Result<Vec<schema::view::map::Serialized>, bonsaidb_core::Error> {
819
350320
        let task_self = self.clone();
820
350320
        let view = view.clone();
821
350320
        self.runtime
822
350320
            .spawn_blocking(move || {
823
350320
                task_self
824
350320
                    .database
825
350320
                    .query_by_name(&view, key, order, limit, access_policy)
826
350320
            })
827
337972
            .await
828
350320
            .map_err(Error::from)?
829
1050960
    }
830

            
831
    async fn query_by_name_with_docs(
832
        &self,
833
        view: &ViewName,
834
        key: Option<SerializedQueryKey>,
835
        order: Sort,
836
        limit: Option<u32>,
837
        access_policy: AccessPolicy,
838
    ) -> Result<schema::view::map::MappedSerializedDocuments, bonsaidb_core::Error> {
839
        let task_self = self.clone();
840
        let view = view.clone();
841
        self.runtime
842
            .spawn_blocking(move || {
843
                task_self
844
                    .database
845
                    .query_by_name_with_docs(&view, key, order, limit, access_policy)
846
            })
847
            .await
848
            .map_err(Error::from)?
849
    }
850

            
851
663168
    async fn reduce_by_name(
852
663168
        &self,
853
663168
        view: &ViewName,
854
663168
        key: Option<SerializedQueryKey>,
855
663168
        access_policy: AccessPolicy,
856
663168
    ) -> Result<Vec<u8>, bonsaidb_core::Error> {
857
663168
        let task_self = self.clone();
858
663168
        let view = view.clone();
859
663168
        self.runtime
860
663168
            .spawn_blocking(move || task_self.database.reduce_by_name(&view, key, access_policy))
861
631632
            .await
862
663168
            .map_err(Error::from)?
863
1989504
    }
864

            
865
364
    async fn reduce_grouped_by_name(
866
364
        &self,
867
364
        view: &ViewName,
868
364
        key: Option<SerializedQueryKey>,
869
364
        access_policy: AccessPolicy,
870
364
    ) -> Result<Vec<MappedSerializedValue>, bonsaidb_core::Error> {
871
364
        let task_self = self.clone();
872
364
        let view = view.clone();
873
364
        self.runtime
874
364
            .spawn_blocking(move || {
875
364
                task_self
876
364
                    .database
877
364
                    .reduce_grouped_by_name(&view, key, access_policy)
878
364
            })
879
364
            .await
880
364
            .map_err(Error::from)?
881
1092
    }
882

            
883
400
    async fn delete_docs_by_name(
884
400
        &self,
885
400
        view: &ViewName,
886
400
        key: Option<SerializedQueryKey>,
887
400
        access_policy: AccessPolicy,
888
400
    ) -> Result<u64, bonsaidb_core::Error> {
889
400
        let task_self = self.clone();
890
400
        let view = view.clone();
891
400
        self.runtime
892
400
            .spawn_blocking(move || {
893
400
                task_self
894
400
                    .database
895
400
                    .delete_docs_by_name(&view, key, access_policy)
896
400
            })
897
400
            .await
898
400
            .map_err(Error::from)?
899
1200
    }
900
}