pipey 0.2.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

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 tokio::select;
use tracing::{info, warn};

#[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 heartbeats = state.heartbeats.clone();
    let user_id = query.user_id.to_string();

    rt::spawn(async move {
        let mut session = session;
        let session_id = arifa
            .subscribe(&channels, ActixWsSession::new(session.clone()), user_id)
            .await;

        info!(%session_id, "subscribed");
        heartbeats.register(session_id.clone(), session.clone());

        'connection: loop {
            select! {
                Some(frame) = msg_stream.next() => {
                    let msg = match frame {
                        Ok(msg) => msg,
                        Err(err) => {
                            warn!(%session_id, %err, "websocket error");
                            break 'connection;
                        }
                    };
                    match msg {
                        Message::Ping(bytes) => {
                            let _ = session.pong(&bytes).await;
                        }
                        Message::Pong(_) => {
                            heartbeats.touch(&session_id);
                        }
                        Message::Close(reason) => {
                            info!(%session_id, ?reason, "client disconnected");
                            break 'connection;
                        }
                        Message::Text(_) => {}
                        Message::Binary(_) => {}
                        Message::Continuation(_) => {}
                        Message::Nop => {}
                    }
                }
                else => break 'connection,
            }
        }

        heartbeats.remove(&session_id);
        let _ = arifa.remove_online_user(&session_id).await;
        arifa.unsubscribe(&session_id, &channels);
        info!(%session_id, "unsubscribed");
        let _ = session.close(None).await;
    });

    Ok(resp)
}