tube 0.0.2

This crate was renamed to 'tubez'. Please use that crate instead.
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

use std::collections::VecDeque;
use std::net::SocketAddr;
use std::sync::Arc;
use std::sync::Mutex;

use super::hyper_tubez_service::TubezMakeSvc;
use super::server_context::ServerContext;
use super::server_error::ServerError;
use super::server_event::ServerEvent;

pub struct Server {
    server_ctx: Arc<Mutex<ServerContext>>,
}
impl Server {
    pub async fn new(addr: &SocketAddr) -> Self {
        let server_ctx = Arc::new(Mutex::new(ServerContext {
            is_complete: false,
            pending_events: VecDeque::new(),
            waker: None,
        }));

        let hyper_server = 
            hyper::Server::bind(&addr)
                .http2_only(true)
                .serve(TubezMakeSvc::new(server_ctx.clone()));

        let tubez_server = Server {
            server_ctx: server_ctx.clone(),
        };

        tokio::spawn(async move {
            if let Err(e) = hyper_server.await {
                let mut server_ctx = server_ctx.lock().unwrap();
                log::error!("Http server error: {}", e);
                server_ctx.pending_events.push_back(Err(ServerError::Err(format!("{:?}", e))));
                // TODO: Need to iterate all tubes and error them here as well.
                if let Some(waker) = server_ctx.waker.take() {
                    waker.wake();
                };
            } else {
                // TODO: Indicate that the http request has EOM'd? Not sure...
                // 
                //         "completes when the server has been shutdown"
                //         https://docs.rs/hyper/latest/hyper/server/struct.Server.html
            }
        });

        tubez_server
    }

    pub async fn new_tube() /*TODO: -> Tube*/ {
        // TODO: This is just a boilerplate mitigator...
        //       Make a channel internal to Server{} and basically hide that 
        //       channel so that users only have to care about Server emitting 
        //       ServerEvent::NewTube() events...
    }
}
impl futures::stream::Stream for Server {
    type Item = Result<ServerEvent, ServerError>;

    fn poll_next(
        self: core::pin::Pin<&mut Self>,
        cx: &mut futures::task::Context,
    ) -> futures::task::Poll<Option<Self::Item>> {
        let mut server_ctx = self.server_ctx.lock().unwrap();
        server_ctx.waker = Some(cx.waker().clone());
        if let Some(event) = server_ctx.pending_events.pop_front() {
            futures::task::Poll::Ready(Some(event))
        } else {
            if server_ctx.is_complete {
                futures::task::Poll::Ready(None)
            } else {
                futures::task::Poll::Pending
            }
        }
    }
}

#[cfg(test)]
mod server_tests {
    // TODO
}