Wrapper function that sends telemetry to the subscriber Service This takes an `hyper::Request` and transforms it into `Vec ` for the underlying `Service` to process.
(
service: Arc<Mutex<S>>,
req: Request<Incoming>,
)
| 270 | /// This takes an `hyper::Request` and transforms it into `Vec<LambdaTelemetry>` for the |
| 271 | /// underlying `Service` to process. |
| 272 | pub(crate) async fn telemetry_wrapper<S, L>( |
| 273 | service: Arc<Mutex<S>>, |
| 274 | req: Request<Incoming>, |
| 275 | ) -> Result<Response<Body>, Box<dyn std::error::Error + Send + Sync>> |
| 276 | where |
| 277 | S: Service<Vec<LambdaTelemetry<L>>, Response = ()>, |
| 278 | S::Error: Into<Box<dyn std::error::Error + Send + Sync>> + fmt::Debug, |
| 279 | S::Future: Send, |
| 280 | L: DeserializeOwned, |
| 281 | { |
| 282 | trace!("Received telemetry request"); |
| 283 | // Parse the request body as a Vec<LambdaTelemetry> |
| 284 | let body = match req.into_body().collect().await { |
| 285 | Ok(body) => body, |
| 286 | Err(e) => { |
| 287 | error!("Error reading telemetry request body: {}", e); |
| 288 | return Ok(hyper::Response::builder() |
| 289 | .status(hyper::StatusCode::BAD_REQUEST) |
| 290 | .body(Body::empty()) |
| 291 | .unwrap()); |
| 292 | } |
| 293 | }; |
| 294 | |
| 295 | let telemetry: Vec<LambdaTelemetry<L>> = match serde_json::from_slice(&body.to_bytes()) { |
| 296 | Ok(telemetry) => telemetry, |
| 297 | Err(e) => { |
| 298 | error!("Error parsing telemetry: {}", e); |
| 299 | return Ok(hyper::Response::builder() |
| 300 | .status(hyper::StatusCode::BAD_REQUEST) |
| 301 | .body(Body::empty()) |
| 302 | .unwrap()); |
| 303 | } |
| 304 | }; |
| 305 | |
| 306 | { |
| 307 | let mut service = service.lock().await; |
| 308 | match service.call(telemetry).await { |
| 309 | Ok(_) => (), |
| 310 | Err(err) => println!("{err:?}"), |
| 311 | } |
| 312 | } |
| 313 | |
| 314 | Ok(hyper::Response::new(Body::empty())) |
| 315 | } |
| 316 | |
| 317 | #[cfg(test)] |
| 318 | mod deserialization_tests { |