lune-std-net 0.3.5

Lune standard library - Net
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

use std::{cell::Cell, net::SocketAddr, rc::Rc};

use async_net::TcpListener;
use futures_lite::pin;
use hyper::server::conn::http1::Builder as Http1Builder;

use mlua::prelude::*;
use mlua_luau_scheduler::LuaSpawnExt;

use crate::{
    server::{config::ServeConfig, handle::ServeHandle, service::Service},
    shared::{
        futures::{Either, either},
        hyper::{HyperIo, HyperTimer},
    },
};

pub mod config;
pub mod handle;
pub mod service;
pub mod upgrade;

/**
    Starts an HTTP server using the given port and configuration.

    Returns a `ServeHandle` that can be used to gracefully stop the server.
*/
pub async fn serve(lua: Lua, port: u16, config: ServeConfig) -> LuaResult<ServeHandle> {
    let address = SocketAddr::from((config.address, port));
    let service = Service {
        lua: lua.clone(),
        address,
        config,
    };

    let listener = TcpListener::bind(address).await?;
    let (handle, shutdown_rx) = ServeHandle::new(address);

    lua.spawn_local({
        let lua = lua.clone();
        async move {
            let handle_dropped = Rc::new(Cell::new(false));
            loop {
                // 1. Keep accepting new connections until we should shutdown
                let (conn, addr) = if handle_dropped.get() {
                    // 1a. Handle has been dropped, and we don't need to listen for shutdown
                    match listener.accept().await {
                        Ok(acc) => acc,
                        Err(_err) => {
                            // TODO: Propagate error somehow
                            continue;
                        }
                    }
                } else {
                    // 1b. Handle is possibly active, we must listen for shutdown
                    match either(shutdown_rx.recv(), listener.accept()).await {
                        Either::Left(Ok(())) => break,
                        Either::Left(Err(_)) => {
                            // NOTE #1: We will only get a RecvError if the serve handle is dropped,
                            // this means lua has garbage collected it and the user does not want
                            // to manually stop the server using the serve handle. Run forever.
                            handle_dropped.set(true);
                            continue;
                        }
                        Either::Right(Ok(acc)) => acc,
                        Either::Right(Err(_err)) => {
                            // TODO: Propagate error somehow
                            continue;
                        }
                    }
                };

                // 2. For each connection, spawn a new task to handle it
                lua.spawn_local({
                    let rx = shutdown_rx.clone();
                    let io = HyperIo::from(conn);

                    let mut svc = service.clone();
                    svc.address = addr;

                    let handle_dropped = Rc::clone(&handle_dropped);
                    async move {
                        let conn = Http1Builder::new()
                            .writev(false)
                            .timer(HyperTimer)
                            .keep_alive(true)
                            .serve_connection(io, svc)
                            .with_upgrades();
                        if handle_dropped.get() {
                            if let Err(_err) = conn.await {
                                // TODO: Propagate error somehow
                            }
                        } else {
                            // NOTE #2: Because we use keep_alive for websockets above, we need to
                            // also manually poll this future and handle the graceful shutdown,
                            // otherwise the already accepted connection will linger and run
                            // even if the stop method has been called on the serve handle
                            pin!(conn);
                            match either(rx.recv(), conn.as_mut()).await {
                                Either::Left(Ok(())) => conn.as_mut().graceful_shutdown(),
                                Either::Left(Err(_)) => {
                                    // Same as note #1
                                    handle_dropped.set(true);
                                    if let Err(_err) = conn.await {
                                        // TODO: Propagate error somehow
                                    }
                                }
                                Either::Right(Ok(())) => {}
                                Either::Right(Err(_err)) => {
                                    // TODO: Propagate error somehow
                                }
                            }
                        }
                    }
                });
            }
        }
    });

    Ok(handle)
}