pipey 0.1.1

A lightweight HTTP-to-WebSocket event delivery 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

use crate::models::{ActixWsSession, AppState, ConnectRequestQuery};
use actix_web::{Error, HttpRequest, HttpResponse, get, rt, web};
use actix_ws::{Message, handle};
use futures_util::StreamExt;
use std::time::Duration;
use tokio::{select, time};

#[get("/ws/pipey")]
pub async fn open_pipe(
    req: HttpRequest,
    body: web::Payload,
    state: web::Data<AppState>,
    query: web::Query<ConnectRequestQuery>,
) -> Result<HttpResponse, Error> {
    let (resp, session, mut msg_stream) = handle(&req, body)?;

    let user_channel = format!("User::{}", query.user_id);
    let channels = vec![user_channel];

    let arifa = state.arifa.clone();
    let user_id = format!("{}", query.user_id);

    rt::spawn(async move {
        let mut session = session;

        let session_id = arifa
            .subscribe(&channels, ActixWsSession::new(session.clone()), user_id)
            .await;

        println!("Subscribed session={session_id}");

        let mut interval = time::interval(Duration::from_secs(30));
        let mut heartbeat_failures: u8 = 0;

        'connection: loop {
            select! {
                _ = interval.tick() => {
                    heartbeat_failures += 1;

                    if heartbeat_failures >= 5 {
                        println!("Heartbeat timeout: {session_id}");
                        break 'connection;
                    }

                    if session.ping(b"").await.is_err() {
                        println!("Failed to send ping");
                        break 'connection;
                    }
                }

                Some(frame) = msg_stream.next() => {
                    let msg = match frame {
                        Ok(msg) => msg,
                        Err(err) => {
                            eprintln!("WebSocket error: {err}");
                            break 'connection;
                        }
                    };

                    match msg {
                        Message::Ping(bytes) => {
                            let _ = session.pong(&bytes).await;
                        }

                        Message::Pong(_) => {
                            heartbeat_failures = 0;
                        }

                        Message::Close(reason) => {
                            println!("Client disconnected: {reason:?}");
                            break 'connection;
                        }

                        Message::Text(_) => {}

                        Message::Binary(_) => {}

                        Message::Continuation(_) => {}

                        Message::Nop => {}
                    }
                }

                else => break 'connection,
            }
        }

        let _ = arifa.remove_online_user(&session_id).await;
        arifa.unsubscribe(&session_id, &channels);

        println!("Unsubscribed session={session_id}");

        let _ = session.close(None).await;
    });

    Ok(resp)
}