mongreldb-server 0.52.3

HTTP daemon for MongrelDB — serves SQL, native queries, and typed Kit API over HTTP for multi-process access.
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

use futures::StreamExt;
use mongreldb_core::{ColumnDef, ColumnFlags, Database, Schema, TypeId, Value};
use mongreldb_server::build_app;
use std::sync::Arc;
use tempfile::tempdir;
use tower::ServiceExt;

fn schema() -> Schema {
    Schema {
        schema_id: 0,
        columns: vec![ColumnDef {
            id: 1,
            name: "id".into(),
            ty: TypeId::Int64,
            flags: ColumnFlags::empty().with(ColumnFlags::PRIMARY_KEY),
            default_value: None,
        }],
        indexes: Vec::new(),
        colocation: Vec::new(),
        constraints: Default::default(),
        clustered: false,
    }
}

fn put(db: &Database, id: i64) {
    db.transaction(|transaction| {
        transaction.put("items", vec![(1, Value::Int64(id))])?;
        Ok(())
    })
    .unwrap();
}

#[tokio::test]
async fn events_is_resumable_sse() {
    let dir = tempdir().unwrap();
    let db = Arc::new(Database::create(dir.path()).unwrap());
    db.create_table("items", schema()).unwrap();
    put(&db, 1);
    let first_id = db
        .change_events_since(None)
        .unwrap()
        .events
        .last()
        .unwrap()
        .id
        .clone()
        .unwrap();
    put(&db, 2);
    let app = build_app(Arc::clone(&db));

    let response = app
        .oneshot(
            axum::http::Request::builder()
                .uri("/events")
                .header("last-event-id", &first_id)
                .body(axum::body::Body::empty())
                .unwrap(),
        )
        .await
        .unwrap();
    assert_eq!(response.status(), axum::http::StatusCode::OK);
    assert_eq!(
        response
            .headers()
            .get(axum::http::header::CONTENT_TYPE)
            .unwrap(),
        "text/event-stream"
    );
    let mut body = response.into_body().into_data_stream();
    let chunk = tokio::time::timeout(std::time::Duration::from_secs(1), body.next())
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    let text = String::from_utf8(chunk.to_vec()).unwrap();
    assert!(text.contains("event: change"));
    assert!(text.contains("\"op\":\"put\""));
    assert!(!text.contains(&format!("id: {first_id}\n")));
}

#[tokio::test]
async fn events_rejects_malformed_resume_id() {
    let dir = tempdir().unwrap();
    let db = Arc::new(Database::create(dir.path()).unwrap());
    let response = build_app(db)
        .oneshot(
            axum::http::Request::builder()
                .uri("/events")
                .header("last-event-id", "broken")
                .body(axum::body::Body::empty())
                .unwrap(),
        )
        .await
        .unwrap();
    assert_eq!(response.status(), axum::http::StatusCode::BAD_REQUEST);
}