Grcov report - client.rs

1

use std::any::TypeId;

2

use std::collections::HashMap;

3

use std::fmt::Debug;

4

use std::ops::Deref;

5

use std::sync::atomic::{AtomicU32, Ordering};

6

use std::sync::Arc;

7

use std::time::Duration;

8

9

use async_trait::async_trait;

10

use bonsaidb_core::admin::{Admin, ADMIN_DATABASE_NAME};

11

use bonsaidb_core::api::{self, Api, ApiName};

12

use bonsaidb_core::arc_bytes::serde::Bytes;

13

use bonsaidb_core::arc_bytes::OwnedBytes;

14

use bonsaidb_core::connection::{

15

    AsyncStorageConnection, Database, HasSession, IdentityReference, Session,

16

};

17

use bonsaidb_core::networking::{

18

    AlterUserPermissionGroupMembership, AlterUserRoleMembership, AssumeIdentity, CreateDatabase,

19

    CreateUser, DeleteDatabase, DeleteUser, ListAvailableSchemas, ListDatabases, LogOutSession,

20

    MessageReceived, Payload, UnregisterSubscriber, CURRENT_PROTOCOL_VERSION,

21

};

22

use bonsaidb_core::permissions::Permissions;

23

use bonsaidb_core::schema::{Nameable, Schema, SchemaName, SchemaSummary, Schematic};

24

use bonsaidb_utils::fast_async_lock;

25

use flume::Sender;

26

use futures::future::BoxFuture;

27

use futures::{Future, FutureExt};

28

use parking_lot::Mutex;

29

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

30

use tokio::runtime::Handle;

31

use url::Url;

32

33

pub use self::remote_database::{AsyncRemoteDatabase, AsyncRemoteSubscriber};

34

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

35

pub use self::sync::{BlockingClient, BlockingRemoteDatabase, BlockingRemoteSubscriber};

36

use crate::builder::Async;

37

use crate::error::Error;

38

use crate::{ApiError, Builder};

39

40

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

41

mod quic_worker;

42

mod remote_database;

43

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

44

mod sync;

45

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

46

mod tungstenite_worker;

47

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

48

mod wasm_websocket_worker;

49

50

7038

#[derive(Debug, Clone, Default)]

51

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

52

53

impl SubscriberMap {

54

4194

    pub fn clear(&self) {

55

4194

        let mut data = self.lock();

56

4194

        data.clear();

57

4194

58

59

60

impl Deref for SubscriberMap {

61

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

62

63

5616

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

64

5616

        &self.0

65

5616

66

67

68

use bonsaidb_core::circulate::Message;

69

70

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

71

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

72

73

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

74

pub type WebSocketError = wasm_websocket_worker::WebSocketError;

75

76

/// Client for connecting to a BonsaiDb server.

77

///

78

/// ## How this type automatically reconnects

79

///

80

/// This type is designed to automatically reconnect if the underlying network

81

/// connection has been lost. When a disconnect happens, the error that caused

82

/// the disconnection will be returned to at least one requestor. If multiple

83

/// pending requests are outstanding, all remaining pending requests will have

84

/// an [`Error::Disconnected`](bonsaidb_core::networking::Error::Disconnected)

85

/// returned. This allows the application to detect when a networking issue has

86

/// arisen.

87

///

88

/// If the disconnect happens while the client is completely idle, the next

89

/// request will report the disconnection error. The subsequent request will

90

/// cause the client to begin reconnecting again.

91

///

92

/// When unauthenticated, this reconnection behavior is mostly transparent --

93

/// disconnection errors can be shown to the user, and service will be restored

94

/// automatically. However, when dealing with authentication, the client does

95

/// not store credentials to be able to send them again when reconnecting. This

96

/// means that the existing client handles will lose their authentication when

97

/// the network connection is broken. The current authentication status can be

98

/// checked using [`HasSession::session()`].

99

///

100

/// ## Connecting via QUIC

101

///

102

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

103

/// port 5645 is assumed.

104

///

105

/// ### With a valid TLS certificate

106

///

107

/// ```rust

108

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

109

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

110

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

111

/// # Ok(())

112

/// # }

113

/// ```

114

///

115

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

116

///

117

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

118

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

119

/// specified when building the client:

120

///

121

/// ```rust

122

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

123

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

124

/// let certificate =

125

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

126

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

127

///     .with_certificate(certificate)

128

///     .build()?;

129

/// # Ok(())

130

/// # }

131

/// ```

132

///

133

/// ## Connecting via WebSockets

134

///

135

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

136

/// WebSockets:

137

///

138

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

139

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

140

///

141

/// ### Without TLS

142

///

143

/// ```rust

144

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

145

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

146

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

147

/// # Ok(())

148

/// # }

149

/// ```

150

///

151

/// ### With TLS

152

///

153

/// ```rust

154

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

155

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

156

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

157

/// # Ok(())

158

/// # }

159

/// ```

160

///

161

/// ## Using a `Api`

162

///

163

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

164

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

165

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

166

///

167

/// ```rust

168

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

169

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

170

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

171

/// use bonsaidb_core::schema::Qualified;

172

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

173

///

174

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

175

/// pub struct Ping;

176

///

177

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

178

/// pub struct Pong;

179

///

180

/// impl Api for Ping {

181

///     type Error = Infallible;

182

///     type Response = Pong;

183

///

184

///     fn name() -> ApiName {

185

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

186

///     }

187

/// }

188

///

189

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

190

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

191

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

192

/// # Ok(())

193

/// # }

194

/// ```

195

///

196

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

197

///

198

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

199

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

200

///

201

/// ```rust

202

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

203

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

204

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

205

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

206

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

207

/// # pub struct Ping;

208

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

209

/// # pub struct Pong;

210

/// # impl Api for Ping {

211

/// #     type Response = Pong;

212

/// #     type Error = Infallible;

213

/// #

214

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

215

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

216

/// #     }

217

/// # }

218

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

219

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

220

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

221

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

222

///     }))

223

///     .build()?;

224

/// # Ok(())

225

/// # }

226

/// ```

227

2523

#[derive(Debug, Clone)]

228

pub struct AsyncClient {

229

    pub(crate) data: Arc<Data>,

230

    session: ClientSession,

231

    request_timeout: Duration,

232

233

234

impl Drop for AsyncClient {

235

34812

    fn drop(&mut self) {

236

34812

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

237

3438

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

238

342

                // Final reference to an authenticated session

239

342

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

240

3096

241

31374

242

34812

243

244

245

impl PartialEq for AsyncClient {

246

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

247

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

248

249

250

251

#[derive(Debug)]

252

pub struct Data {

253

    request_sender: Sender<PendingRequest>,

254

    effective_permissions: Mutex<Option<Permissions>>,

255

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

256

    connection_counter: Arc<AtomicU32>,

257

    request_id: AtomicU32,

258

    subscribers: SubscriberMap,

259

260

261

impl AsyncClient {

262

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

263

1746

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

264

1746

        Builder::new(url)

265

1746

266

267

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

268

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

269

    /// connection.

270

///

271

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

272

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

273

    /// all requests being processed will exit and return

274

    /// [`Error::Disconnected`](bonsaidb_core::networking::Error::Disconnected).

275

    /// The client will automatically try reconnecting.

276

///

277

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

278

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

279

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

280

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

281

    /// the database is to operation.

282

648

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

283

648

        Self::new_from_parts(

284

648

            url,

285

648

            CURRENT_PROTOCOL_VERSION,

286

648

            HashMap::default(),

287

648

            None,

288

648

            None,

289

648

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

290

648

            None,

291

648

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

292

648

            Handle::try_current().ok(),

293

648

294

648

295

296

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

297

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

298

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

299

    /// connection.

300

///

301

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

302

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

303

    /// all requests being processed will exit and return

304

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

305

///

306

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

307

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

308

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

309

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

310

    /// the database is to operation.

311

3114

    pub(crate) fn new_from_parts(

312

3114

        url: Url,

313

3114

        protocol_version: &'static str,

314

3114

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

315

3114

        connect_timeout: Option<Duration>,

316

3114

        request_timeout: Option<Duration>,

317

3114

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

318

3114

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

319

3114

    ) -> Result<Self, Error> {

320

3114

        let subscribers = SubscriberMap::default();

321

3114

        let callback_subscribers = subscribers.clone();

322

3114

        custom_apis.insert(

323

3114

            MessageReceived::name(),

324

3114

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

325

3114

                move |message: MessageReceived| {

326

810

                    let callback_subscribers = callback_subscribers.clone();

327

810

                    async move {

328

810

                        let mut subscribers = callback_subscribers.lock();

329

810

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

330

810

                            if sender

331

810

                                .send(bonsaidb_core::circulate::Message {

332

810

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

333

810

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

334

810

})

335

810

                                .is_err()

336

337

                                subscribers.remove(&message.subscriber_id);

338

810

339

340

810

341

3114

},

342

3114

            ))),

343

3114

);

344

3114

        // Default timeouts to 1 minute.

345

3114

        let connection = ConnectionInfo {

346

3114

            url,

347

3114

            subscribers,

348

3114

            connect_timeout: connect_timeout.unwrap_or(Duration::from_secs(60)),

349

3114

            request_timeout: request_timeout.unwrap_or(Duration::from_secs(60)),

350

3114

};

351

3114

        match connection.url.scheme() {

352

3114

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

353

3114

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

354

1296

                connection,

355

1296

                protocol_version,

356

1296

                certificate,

357

1296

                custom_apis,

358

1296

                tokio,

359

1296

)),

360

            #[cfg(feature = "websockets")]

361

1818

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

362

1818

                connection,

363

1818

                protocol_version,

364

1818

                custom_apis,

365

1818

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

366

1818

                tokio,

367

1818

)),

368

            other => Err(Error::InvalidUrl(format!("unsupported scheme {other}"))),

369

370

3114

371

372

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

373

1296

    fn new_bonsai_client(

374

1296

        server: ConnectionInfo,

375

1296

        protocol_version: &'static str,

376

1296

        certificate: Option<fabruic::Certificate>,

377

1296

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

378

1296

        tokio: Option<Handle>,

379

1296

    ) -> Self {

380

1296

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

381

1296

        let connection_counter = Arc::new(AtomicU32::default());

382

1296

        let request_timeout = server.request_timeout;

383

1296

        let subscribers = server.subscribers.clone();

384

1296

385

1296

        sync::spawn_client(

386

1296

            quic_worker::reconnecting_client_loop(

387

1296

                server,

388

1296

                protocol_version,

389

1296

                certificate,

390

1296

                request_receiver,

391

1296

                Arc::new(custom_apis),

392

1296

                connection_counter.clone(),

393

1296

),

394

1296

            tokio,

395

1296

);

396

1296

397

1296

        Self {

398

1296

            data: Arc::new(Data {

399

1296

                request_sender,

400

1296

                schemas: Mutex::default(),

401

1296

                connection_counter,

402

1296

                request_id: AtomicU32::default(),

403

1296

                effective_permissions: Mutex::default(),

404

1296

                subscribers,

405

1296

}),

406

1296

            session: ClientSession::default(),

407

1296

            request_timeout,

408

1296

409

1296

410

411

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

412

1818

    fn new_websocket_client(

413

1818

        server: ConnectionInfo,

414

1818

        protocol_version: &'static str,

415

1818

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

416

1818

        tokio: Option<Handle>,

417

1818

    ) -> Self {

418

1818

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

419

1818

        let connection_counter = Arc::new(AtomicU32::default());

420

1818

        let request_timeout = server.request_timeout;

421

1818

        let subscribers = server.subscribers.clone();

422

1818

423

1818

        sync::spawn_client(

424

1818

            tungstenite_worker::reconnecting_client_loop(

425

1818

                server,

426

1818

                protocol_version,

427

1818

                request_receiver,

428

1818

                Arc::new(custom_apis),

429

1818

                connection_counter.clone(),

430

1818

),

431

1818

            tokio,

432

1818

);

433

1818

434

1818

        Self {

435

1818

            data: Arc::new(Data {

436

1818

                request_sender,

437

1818

                schemas: Mutex::default(),

438

1818

                request_id: AtomicU32::default(),

439

1818

                connection_counter,

440

1818

                effective_permissions: Mutex::default(),

441

1818

                subscribers,

442

1818

}),

443

1818

            session: ClientSession::default(),

444

1818

            request_timeout,

445

1818

446

1818

447

448

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

449

    fn new_websocket_client(

450

        server: ConnectionInfo,

451

        protocol_version: &'static str,

452

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

453

    ) -> Self {

454

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

455

        let connection_counter = Arc::new(AtomicU32::default());

456

457

        wasm_websocket_worker::spawn_client(

458

            Arc::new(server.url),

459

            protocol_version,

460

            request_receiver,

461

            Arc::new(custom_apis),

462

            server.subscribers.clone(),

463

            connection_counter.clone(),

464

            None,

465

            server.connect_timeout,

466

);

467

468

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

469

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

470

471

        Self {

472

            data: Arc::new(Data {

473

                request_sender,

474

                schemas: Mutex::default(),

475

                request_id: AtomicU32::default(),

476

                connection_counter,

477

                effective_permissions: Mutex::default(),

478

                subscribers: server.subscribers,

479

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

480

                background_task_running,

481

}),

482

            session: ClientSession::default(),

483

            request_timeout: server.request_timeout,

484

485

486

487

1303578

    fn send_request_without_confirmation(

488

1303578

        &self,

489

1303578

        name: ApiName,

490

1303578

        bytes: Bytes,

491

1303578

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

492

1303578

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

493

1303578

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

494

1303578

        self.data.request_sender.send(PendingRequest {

495

1303578

            request: Payload {

496

1303578

                session_id: self.session.session.id,

497

1303578

                id: Some(id),

498

1303578

                name,

499

1303578

                value: Ok(bytes),

500

1303578

},

501

1303578

            responder: result_sender,

502

1303578

        })?;

503

504

1303578

        Ok(result_receiver)

505

1303578

506

507

1098432

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

508

1098432

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

509

510

        #[cfg(target_arch = "wasm32")]

511

        let result = {

512

            use wasm_bindgen::JsCast;

513

            let (timeout_sender, mut timeout_receiver) = futures::channel::oneshot::channel();

514

            // Install the timeout.

515

516

                if let Some(window) = web_sys::window() {

517

                    let timeout = wasm_bindgen::closure::Closure::once_into_js(move || {

518

                        let _result = timeout_sender.send(());

519

});

520

                    let _: Result<_, _> = window

521

                        .set_timeout_with_callback_and_timeout_and_arguments_0(

522

                            timeout.as_ref().unchecked_ref(),

523

                            self.request_timeout

524

                                .as_millis()

525

                                .try_into()

526

                                .unwrap_or(i32::MAX),

527

);

528

529

530

            futures::select! {

531

                result = result_receiver.recv_async() => Ok(result),

532

                _ = timeout_receiver => Err(Error::Network(bonsaidb_core::networking::Error::RequestTimeout)),

533

534

};

535

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

536

1868454

        let result = tokio::time::timeout(self.request_timeout, result_receiver.recv_async()).await;

537

538

1098432

        match result {

539

1098396

            Ok(response) => response?,

540

36

            Err(_) => Err(Error::request_timeout()),

541

542

1098432

543

544

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

545

204804

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

546

204804

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

547

548

204804

        result_receiver.recv_timeout(self.request_timeout)?

549

204804

550

551

    /// Sends an api `request`.

552

1097854

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

553

1097854

        &self,

554

1097854

        request: &Api,

555

1097854

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

556

1097854

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

557

1867825

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

558

1088002

        let response =

559

1088002

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

560

1088002

        response.map_err(ApiError::Api)

561

1097854

562

563

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

564

204600

    fn send_blocking_api_request<Api: api::Api>(

565

204600

        &self,

566

204600

        request: &Api,

567

204600

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

568

204600

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

569

204600

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

570

571

204238

        let response =

572

204238

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

573

204238

        response.map_err(ApiError::Api)

574

204600

575

576

342

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

577

342

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

578

342

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

579

342

            .map(|_| ())

580

342

581

582

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

583

    /// if unauthenticated.

584

    #[must_use]

585

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

586

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

587

        effective_permissions.clone()

588

589

590

432

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

591

432

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

592

432

        subscribers.insert(id, sender);

593

432

594

595

36

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

596

36

        drop(

597

36

            self.send_api_request(&UnregisterSubscriber {

598

36

                database,

599

36

                subscriber_id: id,

600

36

})

601

36

            .await,

602

);

603

36

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

604

36

        subscribers.remove(&id);

605

36

606

607

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

608

144

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

609

144

        drop(self.send_blocking_api_request(&UnregisterSubscriber {

610

144

            database,

611

144

            subscriber_id: id,

612

144

        }));

613

144

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

614

144

        subscribers.remove(&id);

615

144

616

617

1384

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

618

1384

        &self,

619

1384

        name: &str,

620

1384

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

621

1384

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

622

1384

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

623

1384

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

624

1116

            schematic.clone()

625

        } else {

626

268

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

627

268

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

628

268

            schematic

629

};

630

1384

        Ok(AsyncRemoteDatabase::new(

631

1384

            self.clone(),

632

1384

            name.to_string(),

633

1384

            schematic,

634

1384

))

635

1384

636

637

35100

    fn session_is_current(&self) -> bool {

638

35100

        self.session.session.id.is_none()

639

702

            || self.data.connection_counter.load(Ordering::SeqCst) == self.session.connection_id

640

35100

641

642

    /// Sets this instance's request timeout.

643

///

644

    /// Each client has its own timeout. When cloning a client, this timeout

645

    /// setting will be copied to the clone.

646

    pub fn set_request_timeout(&mut self, timeout: impl Into<Duration>) {

647

        self.request_timeout = timeout.into();

648

649

650

651

impl HasSession for AsyncClient {

652

288

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

653

288

        self.session_is_current().then_some(&self.session.session)

654

288

655

656

657

#[async_trait]

658

impl AsyncStorageConnection for AsyncClient {

659

    type Authenticated = Self;

660

    type Database = AsyncRemoteDatabase;

661

662

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

663

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

664

665

666

10476

    async fn create_database_with_schema(

667

10476

        &self,

668

10476

        name: &str,

669

10476

        schema: SchemaName,

670

10476

        only_if_needed: bool,

671

10476

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

672

10476

        self.send_api_request(&CreateDatabase {

673

10476

            database: Database {

674

10476

                name: name.to_string(),

675

10476

                schema,

676

10476

},

677

10476

            only_if_needed,

678

10476

})

679

66096

        .await?;

680

10368

        Ok(())

681

20952

682

683

1206

    async fn database<DB: Schema>(

684

1206

        &self,

685

1206

        name: &str,

686

1206

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

687

1206

        self.remote_database::<DB>(name)

688

1206

689

690

9072

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

691

9072

        self.send_api_request(&DeleteDatabase {

692

9072

            name: name.to_string(),

693

9072

})

694

61236

        .await?;

695

9036

        Ok(())

696

18144

697

698

72

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

699

72

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

700

144

701

702

36

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

703

36

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

704

72

705

706

126

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

707

126

        Ok(self

708

126

            .send_api_request(&CreateUser {

709

126

                username: username.to_string(),

710

126

})

711

126

            .await?)

712

252

713

714

2

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

715

2

        &self,

716

2

        user: U,

717

2

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

718

2

        Ok(self

719

2

            .send_api_request(&DeleteUser {

720

2

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

721

})

722

2

            .await?)

723

4

724

725

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

726

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

727

        &self,

728

        user: U,

729

        password: bonsaidb_core::connection::SensitiveString,

730

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

731

        Ok(self

732

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

733

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

734

                password,

735

})

736

            .await?)

737

738

739

    #[cfg(any(feature = "token-authentication", feature = "password-hashing"))]

740

270

    async fn authenticate(

741

270

        &self,

742

270

        authentication: bonsaidb_core::connection::Authentication,

743

270

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

744

270

        let session = self

745

270

            .send_api_request(&bonsaidb_core::networking::Authenticate { authentication })

746

378

            .await?;

747

270

        Ok(Self {

748

270

            data: self.data.clone(),

749

270

            session: ClientSession {

750

270

                session: Arc::new(session),

751

270

                connection_id: self.data.connection_counter.load(Ordering::SeqCst),

752

270

},

753

270

            request_timeout: self.request_timeout,

754

270

})

755

540

756

757

36

    async fn assume_identity(

758

36

        &self,

759

36

        identity: IdentityReference<'_>,

760

36

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

761

36

        let session = self

762

36

            .send_api_request(&AssumeIdentity(identity.into_owned()))

763

36

            .await?;

764

36

        Ok(Self {

765

36

            data: self.data.clone(),

766

36

            session: ClientSession {

767

36

                session: Arc::new(session),

768

36

                connection_id: self.data.connection_counter.load(Ordering::SeqCst),

769

36

},

770

36

            request_timeout: self.request_timeout,

771

36

})

772

72

773

774

4

    async fn add_permission_group_to_user<

775

4

        'user,

776

4

        'group,

777

4

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

778

4

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

779

4

>(

780

4

        &self,

781

4

        user: U,

782

4

        permission_group: G,

783

4

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

784

4

        self.send_api_request(&AlterUserPermissionGroupMembership {

785

4

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

786

4

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

787

            should_be_member: true,

788

})

789

4

        .await?;

790

4

        Ok(())

791

8

792

793

4

    async fn remove_permission_group_from_user<

794

4

        'user,

795

4

        'group,

796

4

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

797

4

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

798

4

>(

799

4

        &self,

800

4

        user: U,

801

4

        permission_group: G,

802

4

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

803

4

        self.send_api_request(&AlterUserPermissionGroupMembership {

804

4

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

805

4

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

806

            should_be_member: false,

807

})

808

4

        .await?;

809

4

        Ok(())

810

8

811

812

4

    async fn add_role_to_user<

813

4

        'user,

814

4

        'group,

815

4

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

816

4

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

817

4

>(

818

4

        &self,

819

4

        user: U,

820

4

        role: G,

821

4

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

822

4

        self.send_api_request(&AlterUserRoleMembership {

823

4

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

824

4

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

825

            should_be_member: true,

826

})

827

4

        .await?;

828

4

        Ok(())

829

8

830

831

4

    async fn remove_role_from_user<

832

4

        'user,

833

4

        'group,

834

4

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

835

4

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

836

4

>(

837

4

        &self,

838

4

        user: U,

839

4

        role: G,

840

4

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

841

4

        self.send_api_request(&AlterUserRoleMembership {

842

4

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

843

4

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

844

            should_be_member: false,

845

})

846

4

        .await?;

847

4

        Ok(())

848

8

849

850

851

type OutstandingRequestMap = HashMap<u32, PendingRequest>;

852

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

853

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

854

855

#[derive(Debug)]

856

pub struct PendingRequest {

857

    request: Payload,

858

    responder: PendingRequestResponder,

859

860

861

1303884

async fn process_response_payload(

862

1303884

    payload: Payload,

863

1303884

    outstanding_requests: &OutstandingRequestMapHandle,

864

1303884

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

865

1303884

) {

866

1303884

    if let Some(payload_id) = payload.id {

867

1303074

        if let Some(outstanding_request) = {

868

1303074

            let mut outstanding_requests = fast_async_lock!(outstanding_requests);

869

1303074

            outstanding_requests.remove(&payload_id)

870

1303074

} {

871

1303074

            drop(

872

1303074

                outstanding_request

873

1303074

                    .responder

874

1303074

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

875

1303074

);

876

1303074

877

810

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

878

810

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

879

810

        payload.value,

880

810

) {

881

810

        custom_api_callback.response_received(value).await;

882

    } else {

883

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

884

885

1303884

886

887

trait ApiWrapper<Response>: Send + Sync {

888

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

889

890

891

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

892

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

893

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

894

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

895

896

897

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

898

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

899

900

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

901

{}

902

903

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

904

    Box<dyn ApiCallbackFn<Response, F>>,

905

);

906

907

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

908

    for ApiFutureBoxer<Response, F>

909

910

810

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

911

810

        self.0(response).boxed()

912

810

913

914

915

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

916

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

917

3114

    pub fn new<

918

3114

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

919

3114

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

920

3114

>(

921

3114

        callback: F,

922

3114

    ) -> Self {

923

3114

        Self {

924

3114

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

925

3114

926

3114

927

928

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

929

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

930

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

931

    /// external information inside of the callback.

932

    pub fn new_with_context<

933

        Context: Send + Sync + Clone + 'static,

934

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

935

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

936

>(

937

        context: Context,

938

        callback: F,

939

    ) -> Self {

940

        Self {

941

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

942

                move |request| {

943

                    let context = context.clone();

944

                    callback(request, context)

945

},

946

            ))),

947

948

949

950

951

#[async_trait]

952

pub trait AnyApiCallback: Send + Sync + 'static {

953

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

954

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

955

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

956

957

958

#[async_trait]

959

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

960

810

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

961

810

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

962

810

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

963

            Err(err) => {

964

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

965

966

967

810

968

969

970

4499

#[derive(Debug, Clone, Default)]

971

pub struct ClientSession {

972

    session: Arc<Session>,

973

    connection_id: u32,

974

975

976

1458

async fn disconnect_pending_requests(

977

1458

    outstanding_requests: &OutstandingRequestMapHandle,

978

1458

    pending_error: &mut Option<Error>,

979

1458

) {

980

1458

    let mut outstanding_requests = fast_async_lock!(outstanding_requests);

981

1458

    for (_, pending) in outstanding_requests.drain() {

982

198

        drop(

983

198

            pending

984

198

                .responder

985

198

                .send(Err(pending_error.take().unwrap_or(Error::disconnected()))),

986

198

);

987

198

988

1458

989

990

struct ConnectionInfo {

991

    pub url: Url,

992

    pub subscribers: SubscriberMap,

993

    #[cfg_attr(target_arch = "wasm32", allow(dead_code))]

994

    pub connect_timeout: Duration,

995

    pub request_timeout: Duration,

996