cal_jambonz_actix/
lib.rs

1mod handler;
2
3use actix_web::dev::Server;
4use actix_web::http::header::{HeaderName, HeaderValue};
5use actix_web::web::{Data, Payload};
6use actix_web::{rt, web, App, Error, HttpRequest, HttpResponse, HttpServer};
7use actix_ws::Session;
8use cal_jambonz::ws::{JambonzRequest, WebsocketRequest};
9use std::pin::Pin;
10use std::sync::Arc;
11use uuid::Uuid;
12
13pub type HandlerFn<T> =
14    Arc<dyn Fn(HandlerContext<T>) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync>;
15
16type WsHandler<T> = Arc<
17    dyn Fn(
18            HttpRequest,
19            Payload,
20            Data<T>,
21        ) -> Pin<Box<dyn Future<Output = Result<HttpResponse, Error>> + Send>>
22        + Send
23        + Sync,
24>;
25
26pub fn register_handler<T, F, Fut>(handler: F) -> HandlerFn<T>
27where
28    T: Clone + Send + Sync + 'static,
29    F: Fn(HandlerContext<T>) -> Fut + Send + Sync + 'static,
30    Fut: Future<Output = ()> + Send + 'static,
31{
32    Arc::new(move |ctx: HandlerContext<T>| Box::pin(handler(ctx)))
33}
34
35#[derive(Clone)]
36pub struct JambonzRoute<T> {
37    pub path: String,
38    pub ws_type: JambonzRouteType,
39    pub handler: Arc<dyn Fn(HandlerContext<T>) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync>,
40}
41
42#[derive(Clone)]
43pub enum JambonzRouteType {
44    Hook,
45    Recording,
46}
47
48async fn ws_response<T: Clone + 'static>(
49    req: &HttpRequest,
50    stream: Payload,
51    state: Data<T>,
52    route: JambonzRoute<T>,
53) -> Result<HttpResponse, Error> {
54    match actix_ws::handle(req, stream) {
55        Ok((mut res, session, msg_stream)) => {
56            let protocol = match route.ws_type {
57                JambonzRouteType::Hook => "ws.jambonz.org",
58                JambonzRouteType::Recording => "audio.jambonz.org"
59            };
60            res.headers_mut().insert(
61                HeaderName::from_static("sec-websocket-protocol"),
62                HeaderValue::from_str(protocol).expect("valid header value"),
63            );
64            rt::spawn(handler::handler(session, msg_stream, state, route));
65            Ok(res)
66        }
67        Err(e) => {
68            println!("WebSocket error: {:?}", e);
69            Ok(HttpResponse::InternalServerError().finish())
70        }
71    }
72}
73
74pub fn start_server<T>(server: JambonzWebServer<T>) -> Server
75where
76    T: Clone + Send + Sync + 'static,
77{
78    // Wrap routes in an Arc to share it safely across threads
79    let routes_arc = Arc::new(server.routes.clone());
80
81    HttpServer::new(move || {
82        let mut app = App::new().app_data(Data::new(server.app_state.clone()));
83
84        // Clone the Arc for each worker
85        let routes_arc = Arc::clone(&routes_arc);
86
87        // For each route, create a handler
88        for route in routes_arc.iter() { // Use iter() to iterate over the Arc's content
89            let path = route.path.clone();
90
91            // Clone the entire route for use in the closure
92            let route_clone = route.clone();
93
94            // Create handler function with captured route clone
95            let handler_fn = move |req, stream, state| {
96                let route_clone = route_clone.clone(); // Clone again for the inner closure
97                async move {
98                    ws_response(&req, stream, state, route_clone).await
99                }
100            };
101
102            // Register route
103            app = app.route(&path, web::get().to(handler_fn));
104        }
105
106        app
107    })
108        .bind((server.bind_ip.clone(), server.bind_port))
109        .expect("Can not bind to port")
110        .run()
111}
112
113
114
115pub struct JambonzWebServer<T> {
116    pub bind_ip: String,
117    pub bind_port: u16,
118    pub app_state: T,
119    pub routes: Vec<JambonzRoute<T>>,
120}
121
122impl<T: Clone + Send + 'static + Sync> JambonzWebServer<T> {
123    pub fn new(app_state: T) -> Self
124    where
125        T: Clone + Send + 'static,
126    {
127        JambonzWebServer {
128            app_state,
129            bind_ip: "0.0.0.0".to_string(),
130            bind_port: 8080,
131            routes: vec![],
132        }
133    }
134
135    pub fn with_bind_ip(mut self, ip: impl Into<String>) -> Self {
136        self.bind_ip = ip.into();
137        self
138    }
139
140    pub fn with_bind_port(mut self, port: u16) -> Self {
141        self.bind_port = port;
142        self
143    }
144    pub fn add_route(mut self, route: JambonzRoute<T>) -> Self {
145        self.routes.push(route);
146        self
147    }
148
149    pub fn start(self) -> Server {
150        start_server(self)
151    }
152}
153
154pub struct HandlerContext<T> {
155    pub uuid: Uuid,
156    pub session: Session,
157    pub request: JambonzRequest,
158    pub state: Data<T>,
159}
160
161#[derive(Clone)]
162pub struct TestData {
163    pub message:String
164}
cal_jambonz_actix/lib.rs

cal_jambonz_actix/
lib.rs
