1 use {
2 anyhow::Result,
3 axum::{
4 routing::{get, post},
5 Json, Router,
6 },
7 serde::Deserialize,
8 std::sync::Arc,
9 streamhub::{define, define::StreamHubEventSender, utils::Uuid},
10 {
11 tokio,
12 tokio::sync::{mpsc, oneshot},
13 },
14 };
15
16 // the input to our `KickOffClient` handler
17 #[derive(Deserialize)]
18 struct KickOffClient {
19 id: String,
20 }
21
22 #[derive(Clone)]
23 struct ApiService {
24 channel_event_producer: StreamHubEventSender,
25 }
26
27 impl ApiService {
root(&self) -> String28 async fn root(&self) -> String {
29 String::from(
30 "Usage of xiu http api:
31 ./get_stream_status(get) get audio and video stream statistic information.
32 ./kick_off_client(post) kick off client by publish/subscribe id.\n",
33 )
34 }
35
get_stream_status(&self) -> Result<String>36 async fn get_stream_status(&self) -> Result<String> {
37 let (data_sender, mut data_receiver) = mpsc::unbounded_channel();
38 let (size_sender, size_receiver) = oneshot::channel();
39 let hub_event = define::StreamHubEvent::ApiStatistic {
40 data_sender,
41 size_sender,
42 };
43 if let Err(err) = self.channel_event_producer.send(hub_event) {
44 log::error!("send api event error: {}", err);
45 }
46 let mut data = Vec::new();
47 match size_receiver.await {
48 Ok(size) => {
49 if size == 0 {
50 return Ok(String::from("no stream data"));
51 }
52 loop {
53 if let Some(stream_statistics) = data_receiver.recv().await {
54 data.push(stream_statistics);
55 }
56 if data.len() == size {
57 break;
58 }
59 }
60 }
61 Err(err) => {
62 log::error!("start_api_service recv size error: {}", err);
63 }
64 }
65
66 if let Ok(data) = serde_json::to_string(&data) {
67 return Ok(data);
68 }
69
70 Ok(String::from(""))
71 }
72
kick_off_client(&self, id: KickOffClient) -> Result<String>73 async fn kick_off_client(&self, id: KickOffClient) -> Result<String> {
74 let id_result = Uuid::from_str2(&id.id);
75
76 if let Some(id) = id_result {
77 let hub_event = define::StreamHubEvent::ApiKickClient { id };
78
79 if let Err(err) = self.channel_event_producer.send(hub_event) {
80 log::error!("send api kick_off_client event error: {}", err);
81 }
82 }
83
84 Ok(String::from("ok"))
85 }
86 }
87
run(producer: StreamHubEventSender, port: usize)88 pub async fn run(producer: StreamHubEventSender, port: usize) {
89 let api = Arc::new(ApiService {
90 channel_event_producer: producer,
91 });
92
93 let api_root = api.clone();
94 let root = move || async move { api_root.root().await };
95
96 let get_status = api.clone();
97 let status = move || async move {
98 match get_status.get_stream_status().await {
99 Ok(response) => response,
100 Err(_) => "error".to_owned(),
101 }
102 };
103
104 let kick_off = api.clone();
105 let kick = move |Json(id): Json<KickOffClient>| async move {
106 match kick_off.kick_off_client(id).await {
107 Ok(response) => response,
108 Err(_) => "error".to_owned(),
109 }
110 };
111
112 let app = Router::new()
113 .route("/", get(root))
114 .route("/get_stream_status", get(status))
115 .route("/kick_off_client", post(kick));
116
117 log::info!("Http api server listening on http://0.0.0.0:{}", port);
118 axum::Server::bind(&([0, 0, 0, 0], port as u16).into())
119 .serve(app.into_make_service())
120 .await
121 .unwrap();
122 }
123