bonsaidb 0.5.0

A programmable, ACID-compliant, document-database inspired by CouchDB.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

//! Tests breaking a connection and detecting it on the Client.

use std::sync::Arc;
use std::time::Duration;

use bonsaidb::client::url::Url;
use bonsaidb::client::AsyncClient;
use bonsaidb::core::test_util::{Basic, TestDirectory};
use bonsaidb::local::config::Builder;
use bonsaidb::server::{DefaultPermissions, Server, ServerConfiguration};
use bonsaidb_core::connection::{AsyncStorageConnection, HasSession, SensitiveString};
use bonsaidb_core::schema::SerializedCollection;
use bonsaidb_server::BonsaiListenConfig;
use futures::Future;

#[tokio::test]
#[cfg(feature = "websockets")]
async fn websockets() -> anyhow::Result<()> {
    test_sessions(
        "sessions-ws.bonsaidb",
        "ws://localhost:12345",
        |server| async move {
            server
                .listen_for_websockets_on("0.0.0.0:12345", false)
                .await
                .unwrap();
        },
    )
    .await
}

#[tokio::test]
async fn quic() -> anyhow::Result<()> {
    test_sessions(
        "sessions-quic.bonsaidb",
        "bonsaidb://localhost:12346",
        |server| async move {
            server
                .listen_on(BonsaiListenConfig::from(12346).reuse_address(true))
                .await
                .unwrap();
        },
    )
    .await
}

async fn test_sessions<F, Fut>(dir_name: &str, connect_addr: &str, listen: F) -> anyhow::Result<()>
where
    F: Fn(Server) -> Fut + Send + Sync + 'static,
    Fut: Future<Output = ()> + Send + 'static,
{
    drop(env_logger::try_init());
    println!("here");
    let dir = TestDirectory::new(dir_name);
    let server = Server::open(
        ServerConfiguration::new(&dir)
            .default_permissions(DefaultPermissions::AllowAll)
            .with_schema::<Basic>()?,
    )
    .await?;
    println!("Installing cert");
    server.install_self_signed_certificate(false).await?;

    let user_id = server.create_user("ecton").await?;
    server
        .set_user_password(user_id, SensitiveString::from("hunter2"))
        .await?;

    let certificate = server
        .certificate_chain()
        .await?
        .into_end_entity_certificate();

    let reboot = Arc::new(tokio::sync::Notify::new());
    let rebooted = Arc::new(tokio::sync::Notify::new());

    tokio::spawn({
        let reboot = reboot.clone();
        let rebooted = rebooted.clone();
        async move {
            // Start listening
            println!("Listening");

            tokio::spawn(listen(server.clone()));

            // Wait for the client to signal that we can disconnect.
            reboot.notified().await;
            println!("Server Shutting down.");
            // Completely shut the server down and restart it.
            server.shutdown(None).await.unwrap();
            drop(server);
            println!("Server shut down.");
            // Give time for the endpoint to completley close.
            // TODO this is stupid
            tokio::time::sleep(Duration::from_millis(500)).await;
            let server = Server::open(
                ServerConfiguration::new(&dir)
                    .default_permissions(DefaultPermissions::AllowAll)
                    .with_schema::<Basic>()
                    .unwrap(),
            )
            .await
            .unwrap();

            println!("Server listening again.");
            rebooted.notify_one();
            listen(server).await;
        }
    });

    let client = AsyncClient::build(Url::parse(connect_addr)?)
        .with_certificate(certificate.clone())
        .build()?;

    println!("Authenticating.");
    let authenticated = client
        .authenticate_with_password("ecton", SensitiveString::from("hunter2"))
        .await?;
    println!("Creating db");
    let db = client.create_database::<Basic>("basic", true).await?;
    // Verify we have a session at this point.
    authenticated
        .session()
        .unwrap()
        .id
        .expect("session should be present");

    reboot.notify_one();
    rebooted.notified().await;
    // Give the listener a moment to become established.
    tokio::time::sleep(Duration::from_millis(500)).await;
    println!("Continuing client.");

    // Get a disconnection error
    assert!(Basic::get_async(&0, &db).await.is_err());
    println!("Reconnecting");
    // Reconnect
    assert!(Basic::get_async(&0, &db).await.unwrap().is_none());
    println!("Checking session");
    // Verify the client recognizes it was de-authenticated
    assert!(client.session().unwrap().id.is_none());

    println!("Done");
    Ok(())
}