kevy-client-async 1.0.11

Async client for kevy — runtime-agnostic core with feature-gated tokio / smol / async-std transports; mirrors the blocking kevy-client surface.
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

//! Integration test (T4.22): wire round-trip under the `async-std`
//! runtime feature. Same scenarios as `tokio_basic.rs` adapted for
//! async-std's `block_on` + `task::spawn`. Only compiled when
//! `async-std` is enabled.

#![cfg(feature = "async-std")]

use std::io;

use async_std::io::prelude::{ReadExt, WriteExt};
use async_std::net::{TcpListener, TcpStream};
use async_std::task;
use kevy_client_async::AsyncConnection;
use kevy_resp::Reply;

async fn spawn_replier_seq(steps: Vec<(Vec<u8>, Vec<u8>)>) -> io::Result<u16> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let port = listener.local_addr()?.port();
    task::spawn(async move {
        let (mut sock, _) = listener.accept().await.expect("accept");
        for (incoming, outgoing) in steps {
            let mut buf = vec![0u8; incoming.len()];
            sock.read_exact(&mut buf).await.expect("read");
            assert_eq!(buf, incoming, "client wire mismatch");
            sock.write_all(&outgoing).await.expect("write");
        }
        sock.shutdown(std::net::Shutdown::Both).ok();
    });
    Ok(port)
}

async fn spawn_replier(
    incoming_expected: Vec<u8>,
    outgoing: Vec<u8>,
) -> io::Result<u16> {
    spawn_replier_seq(vec![(incoming_expected, outgoing)]).await
}

#[test]
fn ping_round_trip() {
    task::block_on(async {
        let port = spawn_replier(b"*1\r\n$4\r\nPING\r\n".to_vec(), b"+PONG\r\n".to_vec())
            .await
            .unwrap();
        let url = format!("tcp://127.0.0.1:{port}");
        let mut conn = AsyncConnection::open(&url).await.unwrap();
        conn.ping().await.unwrap();
    });
}

#[test]
fn set_then_get() {
    task::block_on(async {
        let port = spawn_replier_seq(vec![
            (
                b"*3\r\n$3\r\nSET\r\n$1\r\nk\r\n$1\r\nv\r\n".to_vec(),
                b"+OK\r\n".to_vec(),
            ),
            (
                b"*2\r\n$3\r\nGET\r\n$1\r\nk\r\n".to_vec(),
                b"$1\r\nv\r\n".to_vec(),
            ),
        ])
        .await
        .unwrap();
        let url = format!("tcp://127.0.0.1:{port}");
        let mut conn = AsyncConnection::open(&url).await.unwrap();
        conn.set(b"k", b"v").await.unwrap();
        let v = conn.get(b"k").await.unwrap();
        assert_eq!(v.as_deref(), Some(&b"v"[..]));
    });
}

#[test]
fn pipeline_one_round_trip() {
    task::block_on(async {
        let port = spawn_replier(
            b"*3\r\n$3\r\nSET\r\n$1\r\nk\r\n$1\r\nv\r\n\
              *2\r\n$3\r\nGET\r\n$1\r\nk\r\n\
              *2\r\n$4\r\nINCR\r\n$3\r\ncnt\r\n"
                .to_vec(),
            b"+OK\r\n$1\r\nv\r\n:1\r\n".to_vec(),
        )
        .await
        .unwrap();
        let url = format!("tcp://127.0.0.1:{port}");
        let mut conn = AsyncConnection::open(&url).await.unwrap();
        let replies = conn
            .pipeline()
            .set(b"k", b"v")
            .get(b"k")
            .incr(b"cnt")
            .run(&mut conn)
            .await
            .unwrap();
        assert_eq!(replies.len(), 3);
        assert!(matches!(replies[0], Reply::Simple(ref s) if s == b"OK"));
        assert!(matches!(replies[1], Reply::Bulk(ref v) if v == b"v"));
        assert!(matches!(replies[2], Reply::Int(1)));
    });
}

#[test]
fn from_transport_accepts_async_std_tcpstream() {
    task::block_on(async {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let port = listener.local_addr().unwrap().port();
        task::spawn(async move {
            let (mut sock, _) = listener.accept().await.unwrap();
            let mut buf = [0u8; 32];
            let _ = sock.read(&mut buf).await;
            sock.write_all(b"+PONG\r\n").await.unwrap();
        });
        let s = TcpStream::connect(("127.0.0.1", port)).await.unwrap();
        let mut conn = AsyncConnection::from_transport(s);
        conn.ping().await.unwrap();
    });
}