Grcov report - client.rs

1

#[cfg(feature = "test-util")]

2

use std::sync::atomic::AtomicBool;

3

use std::{

4

    any::TypeId,

5

    collections::HashMap,

6

    fmt::Debug,

7

    ops::Deref,

8

    sync::{

9

        atomic::{AtomicU32, Ordering},

10

        Arc,

11

},

12

};

13

14

use async_trait::async_trait;

15

#[cfg(feature = "password-hashing")]

16

use bonsaidb_core::connection::Authentication;

17

use bonsaidb_core::{

18

    admin::{Admin, ADMIN_DATABASE_NAME},

19

    api::{self, Api},

20

    arc_bytes::{serde::Bytes, OwnedBytes},

21

    connection::{AsyncStorageConnection, Database, HasSession, IdentityReference, Session},

22

    networking::{

23

        AlterUserPermissionGroupMembership, AlterUserRoleMembership, AssumeIdentity,

24

        CreateDatabase, CreateUser, DeleteDatabase, DeleteUser, ListAvailableSchemas,

25

        ListDatabases, LogOutSession, MessageReceived, Payload, UnregisterSubscriber,

26

        CURRENT_PROTOCOL_VERSION,

27

},

28

    permissions::Permissions,

29

    schema::{ApiName, Nameable, Schema, SchemaName, Schematic},

30

};

31

use bonsaidb_utils::fast_async_lock;

32

use flume::Sender;

33

use futures::{future::BoxFuture, Future, FutureExt};

34

use parking_lot::Mutex;

35

#[cfg(not(target_arch = "wasm32"))]

36

use tokio::{runtime::Handle, task::JoinHandle};

37

use url::Url;

38

39

pub use self::remote_database::{RemoteDatabase, RemoteSubscriber};

40

use crate::{error::Error, ApiError, Builder};

41

42

#[cfg(not(target_arch = "wasm32"))]

43

mod quic_worker;

44

mod remote_database;

45

#[cfg(not(target_arch = "wasm32"))]

46

mod sync;

47

#[cfg(all(feature = "websockets", not(target_arch = "wasm32")))]

48

mod tungstenite_worker;

49

#[cfg(all(feature = "websockets", target_arch = "wasm32"))]

50

mod wasm_websocket_worker;

51

52

6517

#[derive(Debug, Clone, Default)]

53

pub struct SubscriberMap(Arc<Mutex<HashMap<u64, flume::Sender<Message>>>>);

54

55

impl SubscriberMap {

56

2983

    pub fn clear(&self) {

57

2983

        let mut data = self.lock();

58

2983

        data.clear();

59

2983

60

61

62

impl Deref for SubscriberMap {

63

    type Target = Mutex<HashMap<u64, flume::Sender<Message>>>;

64

65

4484

    fn deref(&self) -> &Self::Target {

66

4484

        &self.0

67

4484

68

69

70

use bonsaidb_core::circulate::Message;

71

72

#[cfg(all(feature = "websockets", not(target_arch = "wasm32")))]

73

pub type WebSocketError = tokio_tungstenite::tungstenite::Error;

74

75

#[cfg(all(feature = "websockets", target_arch = "wasm32"))]

76

pub type WebSocketError = wasm_websocket_worker::WebSocketError;

77

78

/// Client for connecting to a BonsaiDb server.

79

///

80

///

81

///

82

/// ## Connecting via QUIC

83

///

84

/// The URL scheme to connect via QUIC is `bonsaidb`. If no port is specified,

85

/// port 5645 is assumed.

86

///

87

/// ### With a valid TLS certificate

88

///

89

/// ```rust

90

/// # use bonsaidb_client::{Client, fabruic::Certificate, url::Url};

91

/// # async fn test_fn() -> anyhow::Result<()> {

92

/// let client = Client::build(Url::parse("bonsaidb://my-server.com")?).finish()?;

93

/// # Ok(())

94

/// # }

95

/// ```

96

///

97

/// ### With a Self-Signed Pinned Certificate

98

///

99

/// When using `install_self_signed_certificate()`, clients will need the

100

/// contents of the `pinned-certificate.der` file within the database. It can be

101

/// specified when building the client:

102

///

103

/// ```rust

104

/// # use bonsaidb_client::{Client, fabruic::Certificate, url::Url};

105

/// # async fn test_fn() -> anyhow::Result<()> {

106

/// let certificate =

107

///     Certificate::from_der(std::fs::read("mydb.bonsaidb/pinned-certificate.der")?)?;

108

/// let client = Client::build(Url::parse("bonsaidb://localhost")?)

109

///     .with_certificate(certificate)

110

///     .finish()?;

111

/// # Ok(())

112

/// # }

113

/// ```

114

///

115

/// ## Connecting via WebSockets

116

///

117

/// WebSockets are built atop the HTTP protocol. There are two URL schemes for

118

/// WebSockets:

119

///

120

/// - `ws`: Insecure WebSockets. Port 80 is assumed if no port is specified.

121

/// - `wss`: Secure WebSockets. Port 443 is assumed if no port is specified.

122

///

123

/// ### Without TLS

124

///

125

/// ```rust

126

/// # use bonsaidb_client::{Client, fabruic::Certificate, url::Url};

127

/// # async fn test_fn() -> anyhow::Result<()> {

128

/// let client = Client::build(Url::parse("ws://localhost")?).finish()?;

129

/// # Ok(())

130

/// # }

131

/// ```

132

///

133

/// ### With TLS

134

///

135

/// ```rust

136

/// # use bonsaidb_client::{Client, fabruic::Certificate, url::Url};

137

/// # async fn test_fn() -> anyhow::Result<()> {

138

/// let client = Client::build(Url::parse("wss://my-server.com")?).finish()?;

139

/// # Ok(())

140

/// # }

141

/// ```

142

///

143

/// ## Using a `Api`

144

///

145

/// Our user guide has a [section on creating and

146

/// using](https://dev.bonsaidb.io/release/guide/about/access-models/custom-api-server.html)

147

/// an [`Api`](api::Api).

148

///

149

/// ```rust

150

/// # use bonsaidb_client::{Client, fabruic::Certificate, url::Url};

151

/// // `bonsaidb_core` is re-exported to `bonsaidb::core` or `bonsaidb_client::core`.

152

/// use bonsaidb_core::{

153

///     api::{Api, Infallible},

154

///     schema::{ApiName, Qualified},

155

/// };

156

/// use serde::{Deserialize, Serialize};

157

///

158

/// #[derive(Serialize, Deserialize, Debug)]

159

/// pub struct Ping;

160

///

161

/// #[derive(Serialize, Deserialize, Clone, Debug)]

162

/// pub struct Pong;

163

///

164

/// impl Api for Ping {

165

///     type Response = Pong;

166

///     type Error = Infallible;

167

///

168

///     fn name() -> ApiName {

169

///         ApiName::private("ping")

170

///     }

171

/// }

172

///

173

/// # async fn test_fn() -> anyhow::Result<()> {

174

/// let client = Client::build(Url::parse("bonsaidb://localhost")?).finish()?;

175

/// let Pong = client.send_api_request_async(&Ping).await?;

176

/// # Ok(())

177

/// # }

178

/// ```

179

///

180

/// ### Receiving out-of-band messages from the server

181

///

182

/// If the server sends a message that isn't in response to a request, the

183

/// client will invoke it's [api callback](Builder::with_api_callback):

184

///

185

/// ```rust

186

/// # use bonsaidb_client::{Client, ApiCallback, fabruic::Certificate, url::Url};

187

/// # // `bonsaidb_core` is re-exported to `bonsaidb::core` or `bonsaidb_client::core`.

188

/// # use bonsaidb_core::{api::{Api, Infallible}, schema::{ApiName, Qualified}};

189

/// # use serde::{Serialize, Deserialize};

190

/// # #[derive(Serialize, Deserialize, Debug)]

191

/// # pub struct Ping;

192

/// # #[derive(Serialize, Deserialize, Clone, Debug)]

193

/// # pub struct Pong;

194

/// # impl Api for Ping {

195

/// #     type Response = Pong;

196

/// #     type Error = Infallible;

197

/// #

198

/// #     fn name() -> ApiName {

199

/// #         ApiName::private("ping")

200

/// #     }

201

/// # }

202

/// # async fn test_fn() -> anyhow::Result<()> {

203

/// let client = Client::build(Url::parse("bonsaidb://localhost")?)

204

///     .with_api_callback(ApiCallback::<Ping>::new(|result: Pong| async move {

205

///         println!("Received out-of-band Pong");

206

///     }))

207

///     .finish()?;

208

/// # Ok(())

209

/// # }

210

/// ```

211

2471

#[derive(Debug, Clone)]

212

pub struct Client {

213

    pub(crate) data: Arc<Data>,

214

    session: Arc<Session>,

215

216

217

impl Drop for Client {

218

35188

    fn drop(&mut self) {

219

35188

        if Arc::strong_count(&self.session) == 1 {

220

2945

            if let Some(session_id) = self.session.id {

221

133

                // Final reference to an authenticated session

222

133

                drop(self.invoke_api_request(&LogOutSession(session_id)));

223

2812

224

32243

225

35188

226

227

228

impl PartialEq for Client {

229

    fn eq(&self, other: &Self) -> bool {

230

        Arc::ptr_eq(&self.data, &other.data)

231

232

233

234

#[derive(Debug)]

235

pub struct Data {

236

    request_sender: Sender<PendingRequest>,

237

    #[cfg(not(target_arch = "wasm32"))]

238

    _worker: CancellableHandle<Result<(), Error>>,

239

    effective_permissions: Mutex<Option<Permissions>>,

240

    schemas: Mutex<HashMap<TypeId, Arc<Schematic>>>,

241

    request_id: AtomicU32,

242

    subscribers: SubscriberMap,

243

    #[cfg(feature = "test-util")]

244

    background_task_running: Arc<AtomicBool>,

245

246

247

impl Client {

248

    /// Returns a builder for a new client connecting to `url`.

249

1634

    pub fn build(url: Url) -> Builder {

250

1634

        Builder::new(url)

251

1634

252

253

254

impl Client {

255

    /// Initialize a client connecting to `url`. This client can be shared by

256

    /// cloning it. All requests are done asynchronously over the same

257

    /// connection.

258

///

259

    /// If the client has an error connecting, the first request made will

260

    /// present that error. If the client disconnects while processing requests,

261

    /// all requests being processed will exit and return

262

    /// [`Error::Disconnected`]. The client will automatically try reconnecting.

263

///

264

    /// The goal of this design of this reconnection strategy is to make it

265

    /// easier to build resilliant apps. By allowing existing Client instances

266

    /// to recover and reconnect, each component of the apps built can adopt a

267

    /// "retry-to-recover" design, or "abort-and-fail" depending on how critical

268

    /// the database is to operation.

269

1197

    pub fn new(url: Url) -> Result<Self, Error> {

270

1197

        Self::new_from_parts(

271

1197

            url,

272

1197

            CURRENT_PROTOCOL_VERSION,

273

1197

            HashMap::default(),

274

1197

            #[cfg(not(target_arch = "wasm32"))]

275

1197

            None,

276

1197

            #[cfg(not(target_arch = "wasm32"))]

277

1197

            Handle::try_current().ok(),

278

1197

279

1197

280

281

    /// Initialize a client connecting to `url` with `certificate` being used to

282

    /// validate and encrypt the connection. This client can be shared by

283

    /// cloning it. All requests are done asynchronously over the same

284

    /// connection.

285

///

286

    /// If the client has an error connecting, the first request made will

287

    /// present that error. If the client disconnects while processing requests,

288

    /// all requests being processed will exit and return

289

    /// [`Error::Disconnected`]. The client will automatically try reconnecting.

290

///

291

    /// The goal of this design of this reconnection strategy is to make it

292

    /// easier to build resilliant apps. By allowing existing Client instances

293

    /// to recover and reconnect, each component of the apps built can adopt a

294

    /// "retry-to-recover" design, or "abort-and-fail" depending on how critical

295

    /// the database is to operation.

296

2831

    pub(crate) fn new_from_parts(

297

2831

        url: Url,

298

2831

        protocol_version: &'static str,

299

2831

        mut custom_apis: HashMap<ApiName, Option<Arc<dyn AnyApiCallback>>>,

300

2831

        #[cfg(not(target_arch = "wasm32"))] certificate: Option<fabruic::Certificate>,

301

2831

        #[cfg(not(target_arch = "wasm32"))] tokio: Option<Handle>,

302

2831

    ) -> Result<Self, Error> {

303

2831

        let subscribers = SubscriberMap::default();

304

2831

        let callback_subscribers = subscribers.clone();

305

2831

        custom_apis.insert(

306

2831

            MessageReceived::name(),

307

2831

            Some(Arc::new(ApiCallback::<MessageReceived>::new(

308

2831

                move |message: MessageReceived| {

309

855

                    let callback_subscribers = callback_subscribers.clone();

310

855

                    async move {

311

855

                        let mut subscribers = callback_subscribers.lock();

312

855

                        if let Some(sender) = subscribers.get(&message.subscriber_id) {

313

855

                            if sender

314

855

                                .send(bonsaidb_core::circulate::Message {

315

855

                                    topic: OwnedBytes::from(message.topic.into_vec()),

316

855

                                    payload: OwnedBytes::from(message.payload.into_vec()),

317

855

})

318

855

                                .is_err()

319

320

                                subscribers.remove(&message.subscriber_id);

321

855

322

323

855

324

2831

},

325

2831

            ))),

326

2831

);

327

2831

        match url.scheme() {

328

2831

            #[cfg(not(target_arch = "wasm32"))]

329

2831

            "bonsaidb" => Ok(Self::new_bonsai_client(

330

1178

                url,

331

1178

                protocol_version,

332

1178

                certificate,

333

1178

                custom_apis,

334

1178

                tokio,

335

1178

                subscribers,

336

1178

)),

337

            #[cfg(feature = "websockets")]

338

1653

            "wss" | "ws" => Ok(Self::new_websocket_client(

339

1653

                url,

340

1653

                protocol_version,

341

1653

                custom_apis,

342

1653

                #[cfg(not(target_arch = "wasm32"))]

343

1653

                tokio,

344

1653

                subscribers,

345

1653

)),

346

            other => {

347

                return Err(Error::InvalidUrl(format!("unsupported scheme {}", other)));

348

349

350

2831

351

352

    #[cfg(not(target_arch = "wasm32"))]

353

1178

    fn new_bonsai_client(

354

1178

        url: Url,

355

1178

        protocol_version: &'static str,

356

1178

        certificate: Option<fabruic::Certificate>,

357

1178

        custom_apis: HashMap<ApiName, Option<Arc<dyn AnyApiCallback>>>,

358

1178

        tokio: Option<Handle>,

359

1178

        subscribers: SubscriberMap,

360

1178

    ) -> Self {

361

1178

        let (request_sender, request_receiver) = flume::unbounded();

362

1178

363

1178

        let worker = sync::spawn_client(

364

1178

            quic_worker::reconnecting_client_loop(

365

1178

                url,

366

1178

                protocol_version,

367

1178

                certificate,

368

1178

                request_receiver,

369

1178

                Arc::new(custom_apis),

370

1178

                subscribers.clone(),

371

1178

),

372

1178

            tokio,

373

1178

);

374

1178

375

1178

        #[cfg(feature = "test-util")]

376

1178

        let background_task_running = Arc::new(AtomicBool::new(true));

377

1178

378

1178

        Self {

379

1178

            data: Arc::new(Data {

380

1178

                request_sender,

381

1178

                _worker: CancellableHandle {

382

1178

                    worker,

383

1178

                    #[cfg(feature = "test-util")]

384

1178

                    background_task_running: background_task_running.clone(),

385

1178

},

386

1178

                schemas: Mutex::default(),

387

1178

                request_id: AtomicU32::default(),

388

1178

                effective_permissions: Mutex::default(),

389

1178

                subscribers,

390

1178

                #[cfg(feature = "test-util")]

391

1178

                background_task_running,

392

1178

}),

393

1178

            session: Arc::new(Session::default()),

394

1178

395

1178

396

397

    #[cfg(all(feature = "websockets", not(target_arch = "wasm32")))]

398

1653

    fn new_websocket_client(

399

1653

        url: Url,

400

1653

        protocol_version: &'static str,

401

1653

        custom_apis: HashMap<ApiName, Option<Arc<dyn AnyApiCallback>>>,

402

1653

        tokio: Option<Handle>,

403

1653

        subscribers: SubscriberMap,

404

1653

    ) -> Self {

405

1653

        let (request_sender, request_receiver) = flume::unbounded();

406

1653

407

1653

        let worker = sync::spawn_client(

408

1653

            tungstenite_worker::reconnecting_client_loop(

409

1653

                url,

410

1653

                protocol_version,

411

1653

                request_receiver,

412

1653

                Arc::new(custom_apis),

413

1653

                subscribers.clone(),

414

1653

),

415

1653

            tokio,

416

1653

);

417

1653

418

1653

        #[cfg(feature = "test-util")]

419

1653

        let background_task_running = Arc::new(AtomicBool::new(true));

420

1653

421

1653

        Self {

422

1653

            data: Arc::new(Data {

423

1653

                request_sender,

424

1653

                #[cfg(not(target_arch = "wasm32"))]

425

1653

                _worker: CancellableHandle {

426

1653

                    worker,

427

1653

                    #[cfg(feature = "test-util")]

428

1653

                    background_task_running: background_task_running.clone(),

429

1653

},

430

1653

                schemas: Mutex::default(),

431

1653

                request_id: AtomicU32::default(),

432

1653

                effective_permissions: Mutex::default(),

433

1653

                subscribers,

434

1653

                #[cfg(feature = "test-util")]

435

1653

                background_task_running,

436

1653

}),

437

1653

            session: Arc::new(Session::default()),

438

1653

439

1653

440

441

    #[cfg(all(feature = "websockets", target_arch = "wasm32"))]

442

    fn new_websocket_client(

443

        url: Url,

444

        protocol_version: &'static str,

445

        custom_apis: HashMap<ApiName, Option<Arc<dyn AnyApiCallback>>>,

446

        subscribers: SubscriberMap,

447

    ) -> Self {

448

        let (request_sender, request_receiver) = flume::unbounded();

449

450

        wasm_websocket_worker::spawn_client(

451

            Arc::new(url),

452

            protocol_version,

453

            request_receiver,

454

            Arc::new(custom_apis),

455

            subscribers.clone(),

456

);

457

458

        #[cfg(feature = "test-util")]

459

        let background_task_running = Arc::new(AtomicBool::new(true));

460

461

        Self {

462

            data: Arc::new(Data {

463

                request_sender,

464

                #[cfg(not(target_arch = "wasm32"))]

465

                worker: CancellableHandle {

466

                    worker,

467

                    #[cfg(feature = "test-util")]

468

                    background_task_running: background_task_running.clone(),

469

},

470

                schemas: Mutex::default(),

471

                request_id: AtomicU32::default(),

472

                effective_permissions: Mutex::default(),

473

                subscribers,

474

                #[cfg(feature = "test-util")]

475

                background_task_running,

476

}),

477

            session: Arc::new(Session::default()),

478

479

480

481

1351774

    fn send_request_without_confirmation(

482

1351774

        &self,

483

1351774

        name: ApiName,

484

1351774

        bytes: Bytes,

485

1351774

    ) -> Result<flume::Receiver<Result<Bytes, Error>>, Error> {

486

1351774

        let (result_sender, result_receiver) = flume::bounded(1);

487

1351774

        let id = self.data.request_id.fetch_add(1, Ordering::SeqCst);

488

1351774

        self.data.request_sender.send(PendingRequest {

489

1351774

            request: Payload {

490

1351774

                session_id: self.session.id,

491

1351774

                id: Some(id),

492

1351774

                name,

493

1351774

                value: Ok(bytes),

494

1351774

},

495

1351774

            responder: result_sender,

496

1351774

        })?;

497

498

1351774

        Ok(result_receiver)

499

1351774

500

501

1136333

    async fn send_request_async(&self, name: ApiName, bytes: Bytes) -> Result<Bytes, Error> {

502

1136333

        let result_receiver = self.send_request_without_confirmation(name, bytes)?;

503

504

1502691

        result_receiver.recv_async().await?

505

1136333

506

507

215308

    fn send_request(&self, name: ApiName, bytes: Bytes) -> Result<Bytes, Error> {

508

215308

        let result_receiver = self.send_request_without_confirmation(name, bytes)?;

509

510

215308

        result_receiver.recv()?

511

215308

512

513

    /// Sends an api `request`.

514

1135667

    pub async fn send_api_request_async<Api: api::Api>(

515

1135667

        &self,

516

1135667

        request: &Api,

517

1135667

    ) -> Result<Api::Response, ApiError<Api::Error>> {

518

1135667

        let request = Bytes::from(pot::to_vec(request).map_err(Error::from)?);

519

1501827

        let response = self.send_request_async(Api::name(), request).await?;

520

1125422

        let response =

521

1125422

            pot::from_slice::<Result<Api::Response, Api::Error>>(&response).map_err(Error::from)?;

522

1125422

        response.map_err(ApiError::Api)

523

1135667

524

525

    /// Sends an api `request` without waiting for a result. The response from

526

    /// the server will be ignored.

527

133

    pub fn invoke_api_request<Api: api::Api>(&self, request: &Api) -> Result<(), Error> {

528

133

        let request = Bytes::from(pot::to_vec(request).map_err(Error::from)?);

529

133

        self.send_request_without_confirmation(Api::name(), request)

530

133

            .map(|_| ())

531

133

532

533

    /// Sends an api `request`.

534

215109

    pub fn send_api_request<Api: api::Api>(

535

215109

        &self,

536

215109

        request: &Api,

537

215109

    ) -> Result<Api::Response, ApiError<Api::Error>> {

538

215109

        let request = Bytes::from(pot::to_vec(request).map_err(Error::from)?);

539

215109

        let response = self.send_request(Api::name(), request)?;

540

541

214824

        let response =

542

214805

            pot::from_slice::<Result<Api::Response, Api::Error>>(&response).map_err(Error::from)?;

543

214824

        response.map_err(ApiError::Api)

544

215128

545

546

    /// Returns the current effective permissions for the client. Returns None

547

    /// if unauthenticated.

548

    pub async fn effective_permissions(&self) -> Option<Permissions> {

549

        let effective_permissions = self.data.effective_permissions.lock();

550

        effective_permissions.clone()

551

552

553

    #[cfg(feature = "test-util")]

554

    #[doc(hidden)]

555

    #[must_use]

556

    pub fn background_task_running(&self) -> Arc<AtomicBool> {

557

        self.data.background_task_running.clone()

558

559

560

456

    pub(crate) fn register_subscriber(&self, id: u64, sender: flume::Sender<Message>) {

561

456

        let mut subscribers = self.data.subscribers.lock();

562

456

        subscribers.insert(id, sender);

563

456

564

565

38

    pub(crate) async fn unregister_subscriber_async(&self, database: String, id: u64) {

566

38

        drop(

567

38

            self.send_api_request_async(&UnregisterSubscriber {

568

38

                database,

569

38

                subscriber_id: id,

570

38

})

571

38

            .await,

572

);

573

38

        let mut subscribers = self.data.subscribers.lock();

574

38

        subscribers.remove(&id);

575

38

576

577

    #[cfg(not(target_arch = "wasm32"))]

578

152

    pub(crate) fn unregister_subscriber(&self, database: String, id: u64) {

579

152

        drop(self.send_api_request(&UnregisterSubscriber {

580

152

            database,

581

152

            subscriber_id: id,

582

152

        }));

583

152

        let mut subscribers = self.data.subscribers.lock();

584

152

        subscribers.remove(&id);

585

152

586

587

1305

    fn database<DB: bonsaidb_core::schema::Schema>(

588

1305

        &self,

589

1305

        name: &str,

590

1305

    ) -> Result<RemoteDatabase, bonsaidb_core::Error> {

591

1305

        let mut schemas = self.data.schemas.lock();

592

1305

        let type_id = TypeId::of::<DB>();

593

1305

        let schematic = if let Some(schematic) = schemas.get(&type_id) {

594

1103

            schematic.clone()

595

        } else {

596

202

            let schematic = Arc::new(DB::schematic()?);

597

202

            schemas.insert(type_id, schematic.clone());

598

202

            schematic

599

};

600

1305

        Ok(RemoteDatabase::new(

601

1305

            self.clone(),

602

1305

            name.to_string(),

603

1305

            schematic,

604

1305

))

605

1305

606

607

608

impl HasSession for Client {

609

    fn session(&self) -> Option<&Session> {

610

        Some(&self.session)

611

612

613

614

#[async_trait]

615

impl AsyncStorageConnection for Client {

616

    type Database = RemoteDatabase;

617

    type Authenticated = Self;

618

619

    async fn admin(&self) -> Self::Database {

620

        self.database::<Admin>(ADMIN_DATABASE_NAME).unwrap()

621

622

623

10868

    async fn create_database_with_schema(

624

10868

        &self,

625

10868

        name: &str,

626

10868

        schema: SchemaName,

627

10868

        only_if_needed: bool,

628

10868

    ) -> Result<(), bonsaidb_core::Error> {

629

10868

        self.send_api_request_async(&CreateDatabase {

630

10868

            database: Database {

631

10868

                name: name.to_string(),

632

10868

                schema,

633

10868

},

634

10868

            only_if_needed,

635

39786

})

636

39786

        .await?;

637

10754

        Ok(())

638

21736

639

640

1186

    async fn database<DB: Schema>(

641

1186

        &self,

642

1186

        name: &str,

643

1186

    ) -> Result<Self::Database, bonsaidb_core::Error> {

644

1186

        self.database::<DB>(name)

645

1186

646

647

9576

    async fn delete_database(&self, name: &str) -> Result<(), bonsaidb_core::Error> {

648

9576

        self.send_api_request_async(&DeleteDatabase {

649

9576

            name: name.to_string(),

650

36518

})

651

36518

        .await?;

652

9538

        Ok(())

653

19152

654

655

38

    async fn list_databases(&self) -> Result<Vec<Database>, bonsaidb_core::Error> {

656

38

        Ok(self.send_api_request_async(&ListDatabases).await?)

657

76

658

659

38

    async fn list_available_schemas(&self) -> Result<Vec<SchemaName>, bonsaidb_core::Error> {

660

38

        Ok(self.send_api_request_async(&ListAvailableSchemas).await?)

661

76

662

663

76

    async fn create_user(&self, username: &str) -> Result<u64, bonsaidb_core::Error> {

664

76

        Ok(self

665

76

            .send_api_request_async(&CreateUser {

666

76

                username: username.to_string(),

667

76

})

668

76

            .await?)

669

152

670

671

2

    async fn delete_user<'user, U: Nameable<'user, u64> + Send + Sync>(

672

2

        &self,

673

2

        user: U,

674

2

    ) -> Result<(), bonsaidb_core::Error> {

675

        Ok(self

676

            .send_api_request_async(&DeleteUser {

677

2

                user: user.name()?.into_owned(),

678

2

})

679

2

            .await?)

680

4

681

682

    #[cfg(feature = "password-hashing")]

683

    async fn set_user_password<'user, U: Nameable<'user, u64> + Send + Sync>(

684

        &self,

685

        user: U,

686

        password: bonsaidb_core::connection::SensitiveString,

687

    ) -> Result<(), bonsaidb_core::Error> {

688

        Ok(self

689

            .send_api_request_async(&bonsaidb_core::networking::SetUserPassword {

690

                user: user.name()?.into_owned(),

691

                password,

692

})

693

            .await?)

694

695

696

    #[cfg(feature = "password-hashing")]

697

5

    async fn authenticate<'user, U: Nameable<'user, u64> + Send + Sync>(

698

5

        &self,

699

5

        user: U,

700

5

        authentication: Authentication,

701

5

    ) -> Result<Self::Authenticated, bonsaidb_core::Error> {

702

5

        let session = self

703

            .send_api_request_async(&bonsaidb_core::networking::Authenticate {

704

5

                user: user.name()?.into_owned(),

705

5

                authentication,

706

10

})

707

10

            .await?;

708

5

        Ok(Self {

709

5

            data: self.data.clone(),

710

5

            session: Arc::new(session),

711

5

})

712

10

713

714

38

    async fn assume_identity(

715

38

        &self,

716

38

        identity: IdentityReference<'_>,

717

38

    ) -> Result<Self::Authenticated, bonsaidb_core::Error> {

718

38

        let session = self

719

38

            .send_api_request_async(&AssumeIdentity(identity.into_owned()))

720

38

            .await?;

721

38

        Ok(Self {

722

38

            data: self.data.clone(),

723

38

            session: Arc::new(session),

724

38

})

725

76

726

727

4

    async fn add_permission_group_to_user<

728

4

        'user,

729

4

        'group,

730

4

        U: Nameable<'user, u64> + Send + Sync,

731

4

        G: Nameable<'group, u64> + Send + Sync,

732

4

>(

733

4

        &self,

734

4

        user: U,

735

4

        permission_group: G,

736

4

    ) -> Result<(), bonsaidb_core::Error> {

737

        self.send_api_request_async(&AlterUserPermissionGroupMembership {

738

4

            user: user.name()?.into_owned(),

739

4

            group: permission_group.name()?.into_owned(),

740

            should_be_member: true,

741

4

})

742

4

        .await?;

743

4

        Ok(())

744

8

745

746

4

    async fn remove_permission_group_from_user<

747

4

        'user,

748

4

        'group,

749

4

        U: Nameable<'user, u64> + Send + Sync,

750

4

        G: Nameable<'group, u64> + Send + Sync,

751

4

>(

752

4

        &self,

753

4

        user: U,

754

4

        permission_group: G,

755

4

    ) -> Result<(), bonsaidb_core::Error> {

756

        self.send_api_request_async(&AlterUserPermissionGroupMembership {

757

4

            user: user.name()?.into_owned(),

758

4

            group: permission_group.name()?.into_owned(),

759

            should_be_member: false,

760

4

})

761

4

        .await?;

762

4

        Ok(())

763

8

764

765

4

    async fn add_role_to_user<

766

4

        'user,

767

4

        'group,

768

4

        U: Nameable<'user, u64> + Send + Sync,

769

4

        G: Nameable<'group, u64> + Send + Sync,

770

4

>(

771

4

        &self,

772

4

        user: U,

773

4

        role: G,

774

4

    ) -> Result<(), bonsaidb_core::Error> {

775

        self.send_api_request_async(&AlterUserRoleMembership {

776

4

            user: user.name()?.into_owned(),

777

4

            role: role.name()?.into_owned(),

778

            should_be_member: true,

779

4

})

780

4

        .await?;

781

4

        Ok(())

782

8

783

784

4

    async fn remove_role_from_user<

785

4

        'user,

786

4

        'group,

787

4

        U: Nameable<'user, u64> + Send + Sync,

788

4

        G: Nameable<'group, u64> + Send + Sync,

789

4

>(

790

4

        &self,

791

4

        user: U,

792

4

        role: G,

793

4

    ) -> Result<(), bonsaidb_core::Error> {

794

        self.send_api_request_async(&AlterUserRoleMembership {

795

4

            user: user.name()?.into_owned(),

796

4

            role: role.name()?.into_owned(),

797

            should_be_member: false,

798

4

})

799

4

        .await?;

800

4

        Ok(())

801

8

802

803

804

type OutstandingRequestMap = HashMap<u32, PendingRequest>;

805

type OutstandingRequestMapHandle = Arc<async_lock::Mutex<OutstandingRequestMap>>;

806

type PendingRequestResponder = Sender<Result<Bytes, Error>>;

807

808

#[derive(Debug)]

809

pub struct PendingRequest {

810

    request: Payload,

811

    responder: PendingRequestResponder,

812

813

814

#[cfg(not(target_arch = "wasm32"))]

815

#[derive(Debug)]

816

struct CancellableHandle<T> {

817

    worker: JoinHandle<T>,

818

    #[cfg(feature = "test-util")]

819

    background_task_running: Arc<AtomicBool>,

820

821

822

#[cfg(not(target_arch = "wasm32"))]

823

impl<T> Drop for CancellableHandle<T> {

824

2812

    fn drop(&mut self) {

825

2812

        self.worker.abort();

826

2812

        #[cfg(feature = "test-util")]

827

2812

        self.background_task_running.store(false, Ordering::Release);

828

2812

829

830

831

1352477

async fn process_response_payload(

832

1352477

    payload: Payload,

833

1352477

    outstanding_requests: &OutstandingRequestMapHandle,

834

1352477

    custom_apis: &HashMap<ApiName, Option<Arc<dyn AnyApiCallback>>>,

835

1352477

) {

836

1352477

    if let Some(payload_id) = payload.id {

837

1351622

        if let Some(outstanding_request) = {

838

1351622

            let mut outstanding_requests = fast_async_lock!(outstanding_requests);

839

1351622

            outstanding_requests.remove(&payload_id)

840

1351622

} {

841

1351622

            drop(

842

1351622

                outstanding_request

843

1351622

                    .responder

844

1351622

                    .send(payload.value.map_err(Error::from)),

845

1351622

);

846

1351622

847

855

    } else if let (Some(custom_api_callback), Ok(value)) = (

848

855

        custom_apis.get(&payload.name).and_then(Option::as_ref),

849

855

        payload.value,

850

855

) {

851

855

        custom_api_callback.response_received(value).await;

852

    } else {

853

        log::warn!("unexpected api response received ({})", payload.name);

854

855

1352477

856

857

trait ApiWrapper<Response>: Send + Sync {

858

    fn invoke(&self, response: Response) -> BoxFuture<'static, ()>;

859

860

861

/// A callback that is invoked when an [`Api::Response`](Api::Response)

862

/// value is received out-of-band (not in reply to a request).

863

pub struct ApiCallback<Api: api::Api> {

864

    generator: Box<dyn ApiWrapper<Api::Response>>,

865

866

867

/// The trait bounds required for the function wrapped in a [`ApiCallback`].

868

pub trait ApiCallbackFn<Request, F>: Fn(Request) -> F + Send + Sync + 'static {}

869

870

impl<T, Request, F> ApiCallbackFn<Request, F> for T where T: Fn(Request) -> F + Send + Sync + 'static

871

{}

872

873

struct ApiFutureBoxer<Response: Send + Sync, F: Future<Output = ()> + Send + Sync>(

874

    Box<dyn ApiCallbackFn<Response, F>>,

875

);

876

877

impl<Response: Send + Sync, F: Future<Output = ()> + Send + Sync + 'static> ApiWrapper<Response>

878

    for ApiFutureBoxer<Response, F>

879

880

855

    fn invoke(&self, response: Response) -> BoxFuture<'static, ()> {

881

855

        (&self.0)(response).boxed()

882

855

883

884

885

impl<Api: api::Api> ApiCallback<Api> {

886

    /// Returns a new instance wrapping the provided function.

887

2831

    pub fn new<

888

2831

        F: ApiCallbackFn<Api::Response, Fut>,

889

2831

        Fut: Future<Output = ()> + Send + Sync + 'static,

890

2831

>(

891

2831

        callback: F,

892

2831

    ) -> Self {

893

2831

        Self {

894

2831

            generator: Box::new(ApiFutureBoxer::<Api::Response, Fut>(Box::new(callback))),

895

2831

896

2831

897

    /// Returns a new instance wrapping the provided function, passing a clone

898

    /// of `context` as the second parameter. This is just a convenience wrapper

899

    /// around `new()` that produces more readable code when needing to access

900

    /// external information inside of the callback.

901

    pub fn new_with_context<

902

        Context: Send + Sync + Clone + 'static,

903

        F: Fn(Api::Response, Context) -> Fut + Send + Sync + 'static,

904

        Fut: Future<Output = ()> + Send + Sync + 'static,

905

>(

906

        context: Context,

907

        callback: F,

908

    ) -> Self {

909

        Self {

910

            generator: Box::new(ApiFutureBoxer::<Api::Response, Fut>(Box::new(

911

                move |request| {

912

                    let context = context.clone();

913

                    callback(request, context)

914

},

915

            ))),

916

917

918

919

920

#[async_trait]

921

pub trait AnyApiCallback: Send + Sync + 'static {

922

    /// An out-of-band `response` was received. This happens when the server

923

    /// sends a response that isn't in response to a request.

924

    async fn response_received(&self, response: Bytes);

925

926

927

#[async_trait]

928

impl<Api: api::Api> AnyApiCallback for ApiCallback<Api> {

929

855

    async fn response_received(&self, response: Bytes) {

930

855

        match pot::from_slice::<Result<Api::Response, Api::Error>>(&response) {

931

855

            Ok(response) => self.generator.invoke(response.unwrap()).await,

932

            Err(err) => {

933

                log::error!("error deserializing api: {err}");

934

935

936

855

937