rifts 0.2.0

Rift Realtime Protocol / 1.0 — server-side implementation
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

//! Ntex WebSocket adapter.
//!
//! Wraps ntex WebSocket (`ntex::ws::WsSink` + message stream) as a
//! Rift `TransportConnection` via a channel bridge.

use std::net::SocketAddr;

use crate::transport::TransportConnection;
use crate::transport::bridge::spawn_bridge_local;

/// Wrap an ntex WebSocket pair into a `TransportConnection`.
///
/// `stream` is the message stream from `ntex::web::ws::start()`.
/// `sink` is the `ntex::ws::WsSink` for outgoing messages.
///
/// ```ignore
/// use ntex::web;
/// use ntex::ws;
///
/// async fn handler(req: web::HttpRequest, stream: web::Payload)
///     -> Result<web::HttpResponse, web::Error>
/// {
///     ws::start(req, stream, |msg_stream, sink| async move {
///         let conn = rift::transport::ntex::into_connection(sink, msg_stream, None);
///         tokio::spawn(async move {
///             rift_server.accept_and_spawn(conn);
///         });
///     })
/// }
/// ```
pub fn into_connection<S, E>(
    sink: ntex::ws::WsSink,
    mut stream: S,
    peer: Option<SocketAddr>,
) -> Box<dyn TransportConnection>
where
    S: futures_util::Stream<Item = Result<ntex::ws::Message, E>> + Unpin + 'static,
    E: std::fmt::Debug,
{
    spawn_bridge_local(
        peer,
        256,
        // Reader: pull from ntex message stream → tokio channel.
        move |tx| {
            ntex::rt::spawn(async move {
                use futures_util::StreamExt;
                while let Some(msg) = stream.next().await {
                    let raw = match msg {
                        Ok(ntex::ws::Message::Binary(bin)) => {
                            let mut v = Vec::with_capacity(1 + bin.len());
                            v.push(b'B');
                            v.extend_from_slice(&bin);
                            v
                        }
                        Ok(ntex::ws::Message::Text(text)) => {
                            let mut v = Vec::with_capacity(1 + text.len());
                            v.push(b'T');
                            v.extend_from_slice(text.as_bytes());
                            v
                        }
                        Ok(ntex::ws::Message::Close(_)) => vec![b'C'],
                        Ok(_) => continue,
                        Err(_) => break,
                    };
                    if tx.send(raw).await.is_err() {
                        break;
                    }
                }
            });
        },
        // Writer: receive from tokio channel → ntex WsSink.
        move |mut rx| {
            ntex::rt::spawn(async move {
                while let Some(raw) = rx.recv().await {
                    if raw.first() == Some(&b'C') {
                        let _ = sink.send(ntex::ws::Message::Close(None)).await;
                        break;
                    }
                    if raw.len() > 1 {
                        let _ = sink
                            .send(ntex::ws::Message::Binary(raw[1..].to_vec().into()))
                            .await;
                    }
                }
            });
        },
    )
}