meio-connect 0.97.0

Set of connectors for MEIO framework
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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182

//! Contains basic routing capabilities.

use anyhow::Error;
use hyper::{Body, Request, Response};
use meio::handlers::Interact;
use meio::prelude::{ActionHandler, Actor, Address, Interaction};
use serde::{de::DeserializeOwned, Deserialize};
use slab::Slab;
use std::future::Future;
use std::net::SocketAddr;
use std::ops::Deref;
use std::pin::Pin;
use std::sync::Arc;
use thiserror::Error;
use tokio::sync::RwLock;

/// A boxed route alias to hold them in the `RoutingTable`.
pub type BoxedRoute = Box<dyn Route>;

#[derive(Clone, Default)]
pub(super) struct RoutingTable {
    routes: Arc<RwLock<Slab<BoxedRoute>>>,
}

impl RoutingTable {
    pub async fn insert_route(&mut self, route: BoxedRoute) {
        let mut routes = self.routes.write().await;
        routes.insert(route);
    }

    pub async fn routes(&self) -> impl Deref<Target = Slab<BoxedRoute>> + '_ {
        self.routes.read().await
    }
}

/// Empty parameters.
#[derive(Default, Deserialize)]
pub struct NoParameters {}

/// Error when route failed (matched, but can't be processed).
#[derive(Debug, Error)]
#[error("route error [path = {path}, query = {query}]: {reason}")]
pub struct RouteError {
    /// Path of a request.
    pub path: String,
    /// Query of a request.
    pub query: String,
    /// The reason of failing.
    pub reason: String,
}

impl RouteError {
    /// Creates a new error instance.
    /// Expects parameters of a request.
    pub fn new(path: impl ToString, query: impl ToString, reason: impl ToString) -> Self {
        Self {
            path: path.to_string(),
            query: query.to_string(),
            reason: reason.to_string(),
        }
    }
}

/// The special handler for plain routes.
pub trait DirectPath: Sized + Send + Sync + 'static {
    /// Extracted value from the request.
    type Output: DeserializeOwned + Send;
    /// Extra type parameter.
    /// Used for web sockets.
    type Parameter;
    /// Slice of plain paths for matching a request.
    fn paths() -> &'static [&'static str];
}

impl<T> FromRequest for T
where
    T: DirectPath,
{
    type Output = <T as DirectPath>::Output;

    fn from_request(&self, request: &Request<Body>) -> Option<Result<Self::Output, Error>> {
        let uri = request.uri();
        let path = uri.path();
        if Self::paths().iter().any(|p| p == &path) {
            // `Body` should be parsed in the handler, becuase it doesn't participate in routing.
            // And `Body` extracting requires an ownership.
            let query = uri.query().unwrap_or("");
            let output =
                serde_qs::from_str(query).map_err(|err| RouteError::new(path, query, err).into());
            Some(output)
        } else {
            None
        }
    }
}

/// Gets the data from a request.
pub trait FromRequest: Sized + Send + Sync + 'static {
    /// The expected result of data extraction.
    type Output: Send;

    /// Matches and tries to extract data for the corresponding route.
    fn from_request(&self, request: &Request<Body>) -> Option<Result<Self::Output, Error>>;
}

/// The request wrapper.
pub struct Req<T: FromRequest> {
    /// Address of a client.
    pub addr: SocketAddr,
    /// Parsed request value.
    pub data: T::Output,
    /// Body of the request.
    pub body: Body,
}

impl<T: FromRequest> Interaction for Req<T> {
    type Output = Response<Body>;
}

/// The result of `Route`'s matching.
///
/// The future for handing the request of instant error response.
pub type RouteResult =
    Result<Pin<Box<dyn Future<Output = Result<Response<Body>, Error>> + Send>>, Request<Body>>;

/// The abstract route.
pub trait Route: Send + Sync + 'static {
    /// Tries to process a request and produce a response for it.
    ///
    /// Returns a `Future` tha tproduces a response.
    fn try_route(&self, addr: &SocketAddr, request: Request<Body>) -> RouteResult;
}

/// The route matcher for ordinary HTTP request.
pub struct WebRoute<E, A>
where
    A: Actor,
{
    extractor: E,
    address: Address<A>,
}

impl<E, A> WebRoute<E, A>
where
    A: Actor,
{
    /// Creates a new `Route` checker for plain requests.
    ///
    /// Expects an `extractor` that sends a matched request
    /// to an `address`.
    pub fn new(extractor: E, address: Address<A>) -> Self {
        Self { extractor, address }
    }
}

/// Here used `ActionHandler` instead of `InteractionHandler`
/// to make it possible to use both types of handlers.
/// And `ActionHandler` is useful to handle long running requests.
impl<E, A> Route for WebRoute<E, A>
where
    E: FromRequest,
    A: Actor + ActionHandler<Interact<Req<E>>>,
{
    fn try_route(&self, addr: &SocketAddr, request: Request<Body>) -> RouteResult {
        match self.extractor.from_request(&request) {
            Some(Ok(data)) => {
                let msg = Req {
                    addr: *addr,
                    data,
                    body: request.into_body(),
                };
                let fut = self.address.interact(msg).recv();
                Ok(Box::pin(fut))
            }
            None => Err(request),
            Some(Err(err)) => {
                let fut = async move { Err(err) };
                Ok(Box::pin(fut))
            }
        }
    }
}