engineioxide 0.17.6

Engine IO server implementation as a Tower Service.
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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167

use std::sync::Arc;

use bytes::Bytes;
use engineioxide::{
    Str,
    handler::EngineIoHandler,
    socket::{DisconnectReason, Socket},
};
use tokio::sync::mpsc;
use tokio_tungstenite::tungstenite::Message;

mod fixture;
use fixture::{create_polling_connection, create_server, create_ws_connection, send_req};

#[derive(Debug, Clone)]
struct VolatileHandler {
    socket_tx: mpsc::Sender<Arc<Socket<()>>>,
}
impl EngineIoHandler for VolatileHandler {
    type Data = ();
    fn on_connect(self: Arc<Self>, socket: Arc<Socket<()>>) {
        self.socket_tx.try_send(socket).ok();
    }
    fn on_disconnect(&self, _socket: Arc<Socket<()>>, _reason: DisconnectReason) {}
    fn on_message(self: &Arc<Self>, msg: Str, socket: Arc<Socket<()>>) {
        if msg == "trigger_volatile" {
            socket.emit_volatile("volatile_response");
        } else if msg == "echo" {
            socket.emit(msg).ok();
        }
    }
    fn on_binary(self: &Arc<Self>, data: Bytes, socket: Arc<Socket<()>>) {
        socket.emit_binary(data).ok();
    }
}

#[tokio::test]
async fn volatile_message_arrives_via_polling() {
    let (socket_tx, _socket_rx) = mpsc::channel(10);
    let mut svc = create_server(VolatileHandler { socket_tx }).await;
    let sid = create_polling_connection(&mut svc).await;

    send_req(
        &mut svc,
        format!("transport=polling&sid={sid}"),
        http::Method::POST,
        Some("4trigger_volatile".into()),
    )
    .await;

    let response = send_req(
        &mut svc,
        format!("transport=polling&sid={sid}"),
        http::Method::GET,
        None,
    )
    .await;
    // send_req skips the first character (packet type '4')
    assert_eq!(response, "volatile_response");
}

#[tokio::test]
async fn mixed_volatile_and_normal_via_polling() {
    let (socket_tx, mut socket_rx) = mpsc::channel(10);
    let mut svc = create_server(VolatileHandler { socket_tx }).await;
    let sid = create_polling_connection(&mut svc).await;

    let socket = socket_rx
        .recv()
        .await
        .expect("socket not received from on_connect");

    // Send a normal message AND a volatile
    socket.emit("normal_msg").ok();
    assert!(socket.emit_volatile("volatile_msg"));

    let response = send_req(
        &mut svc,
        format!("transport=polling&sid={sid}"),
        http::Method::GET,
        None,
    )
    .await;
    // Volatile should have priority and appear first, before normal.
    // send_req skips only the first character (the leading '4' of the volatile).
    assert_eq!(response, "normal_msg\x1e4volatile_msg");
}

#[tokio::test]
async fn volatile_overwrite_only_latest_survives() {
    let (socket_tx, mut socket_rx) = mpsc::channel(10);
    let mut svc = create_server(VolatileHandler { socket_tx }).await;
    let sid = create_polling_connection(&mut svc).await;

    let socket = socket_rx
        .recv()
        .await
        .expect("socket not received from on_connect");

    assert!(socket.emit_volatile("dropped"));
    assert!(socket.emit_volatile("kept"));
    assert!(socket.emit("normal").is_ok());

    let response = send_req(
        &mut svc,
        format!("transport=polling&sid={sid}"),
        http::Method::GET,
        None,
    )
    .await;
    // After send_req's skip(1): "kept" then \x1e separator then "4normal"
    assert_eq!(response, "normal\x1e4kept");
}

#[tokio::test]
async fn volatile_only_arrives_via_polling() {
    let (socket_tx, mut socket_rx) = mpsc::channel(10);
    let mut svc = create_server(VolatileHandler { socket_tx }).await;
    let sid = create_polling_connection(&mut svc).await;

    let socket = socket_rx
        .recv()
        .await
        .expect("socket not received from on_connect");

    assert!(socket.emit_volatile("volatile_only"));

    let response = send_req(
        &mut svc,
        format!("transport=polling&sid={sid}"),
        http::Method::GET,
        None,
    )
    .await;
    assert_eq!(response, "volatile_only");
}

#[tokio::test]
async fn volatile_only_arrives_via_ws() {
    use std::time::Duration;

    use futures_util::StreamExt;

    let (socket_tx, mut socket_rx) = mpsc::channel(10);
    let mut svc = create_server(VolatileHandler { socket_tx }).await;
    let mut stream = create_ws_connection(&mut svc).await;

    let socket = socket_rx
        .recv()
        .await
        .expect("socket not received from on_connect");

    let _open_packet = stream.next().await.unwrap().unwrap();

    let ping = stream.next().await.unwrap().unwrap();
    assert_eq!(ping, Message::Text("2".into()));

    assert!(socket.emit_volatile("volatile_only"));

    let volatile = tokio::time::timeout(Duration::from_millis(100), stream.next()).await;

    let msg = volatile
        .expect("timeout: volatile-only packet was never flushed over websocket")
        .unwrap()
        .unwrap();
    assert_eq!(msg, Message::Text("4volatile_only".into()));
}