relay-knowledge 1.1.17

Graph-database-based knowledge graph project.
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

use std::time::Duration;

use tokio::io::AsyncWriteExt;

use super::*;
use crate::net::qos::{QosPolicy, QosRuntime};

#[tokio::test]
async fn send_request_with_qos_holds_permit_until_body_is_consumed() {
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
        .await
        .expect("listener should bind");
    let addr = listener.local_addr().expect("local addr should load");
    let (release_body, wait_for_release) = tokio::sync::oneshot::channel();
    let server = tokio::spawn(async move {
        let (mut stream, _) = listener.accept().await.expect("client should connect");
        stream
            .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\n")
            .await
            .expect("headers should write");
        let _ = wait_for_release.await;
        stream.write_all(b"ok").await.expect("body should write");
    });
    let qos = QosRuntime::default();
    let policy = QosPolicy::new(8, 1, 8).expect("policy should build");
    let client = reqwest::Client::new();

    let response = send_request_with_qos(&qos, &policy, client.get(format!("http://{addr}/slow")))
        .await
        .expect("response headers should arrive");

    assert_eq!(qos.diagnostics_snapshot().usage.in_flight_requests, 1);
    release_body.send(()).expect("body release should send");
    assert_eq!(
        response.text().await.expect("body should read"),
        "ok".to_owned()
    );
    assert_eq!(qos.diagnostics_snapshot().usage.in_flight_requests, 0);
    server.await.expect("server task should finish");
}

#[tokio::test]
async fn qos_response_records_timeout_while_reading_body() {
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
        .await
        .expect("listener should bind");
    let addr = listener.local_addr().expect("local addr should load");
    let server = tokio::spawn(async move {
        let (mut stream, _) = listener.accept().await.expect("client should connect");
        stream
            .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\n")
            .await
            .expect("headers should write");
        tokio::time::sleep(Duration::from_millis(200)).await;
    });
    let qos = QosRuntime::default();
    let policy = QosPolicy::new(8, 1, 8).expect("policy should build");
    let client = reqwest::Client::new();

    let response = send_request_with_qos(
        &qos,
        &policy,
        client
            .get(format!("http://{addr}/slow"))
            .timeout(Duration::from_millis(50)),
    )
    .await
    .expect("response headers should arrive");
    let error = response
        .text()
        .await
        .expect_err("body read should time out");

    assert!(error.is_timeout());
    assert_eq!(qos.diagnostics_snapshot().timed_out_total, 1);
    assert_eq!(qos.diagnostics_snapshot().usage.in_flight_requests, 0);
    server.await.expect("server task should finish");
}

#[tokio::test]
async fn cancelled_send_records_cancellation_and_releases_permit() {
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
        .await
        .expect("listener should bind");
    let addr = listener.local_addr().expect("local addr should load");
    let (accepted_sender, accepted_receiver) = tokio::sync::oneshot::channel();
    let server = tokio::spawn(async move {
        let (_stream, _) = listener.accept().await.expect("client should connect");
        let _ = accepted_sender.send(());
        std::future::pending::<()>().await;
    });
    let qos = QosRuntime::default();
    let policy = QosPolicy::new(8, 1, 8).expect("policy should build");
    let request_qos = qos.clone();
    let request = tokio::spawn(async move {
        send_request_with_qos(
            &request_qos,
            &policy,
            reqwest::Client::new().get(format!("http://{addr}/cancel")),
        )
        .await
    });

    accepted_receiver
        .await
        .expect("server should observe admitted request");
    request.abort();
    let _ = request.await;
    tokio::task::yield_now().await;

    let diagnostics = qos.diagnostics_snapshot();
    assert_eq!(diagnostics.cancelled_total, 1);
    assert_eq!(diagnostics.usage.in_flight_requests, 0);
    server.abort();
}