Git repository hosting, collaboration, and discovery for the Fediverse.
begin new protocol refactor
parent: tbd commit: 2091700
| 1 | use std::fmt::Debug; |
| 2 | |
| 3 | use anyhow::Error; |
| 4 | use futures_util::{SinkExt, StreamExt}; |
| 5 | use giterated_models::{ |
| 6 | messages::{error::ConnectionError, MessageTarget}, |
| 7 | model::{ |
| 8 | authenticated::{Authenticated, AuthenticationSourceProvider}, |
| 9 | instance::Instance, |
| 10 | }, |
| 11 | }; |
| 12 | use serde::{de::DeserializeOwned, Serialize}; |
| 13 | use tokio_tungstenite::tungstenite::Message; |
| 14 | |
| 15 | use crate::DaemonConnectionPool; |
| 16 | |
| 17 | pub fn request_local<T: Serialize + DeserializeOwned + Debug + Send + Sync + MessageTarget>( |
| 18 | request: T, |
| 19 | ) -> PreparedRequest<T> { |
| 20 | PreparedRequest { |
| 21 | request: Authenticated::new_empty(request), |
| 22 | instance: None, |
| 23 | } |
| 24 | } |
| 25 | |
| 26 | pub fn request<T: Serialize + DeserializeOwned + Debug + Send + Sync + MessageTarget>( |
| 27 | instance: Instance, |
| 28 | request: T, |
| 29 | ) -> PreparedRequest<T> { |
| 30 | PreparedRequest { |
| 31 | request: Authenticated::new_empty(request), |
| 32 | instance: Some(instance), |
| 33 | } |
| 34 | } |
| 35 | |
| 36 | pub struct PreparedRequest<T: Serialize + DeserializeOwned + Send + Sync> { |
| 37 | instance: Option<Instance>, |
| 38 | request: Authenticated<T>, |
| 39 | } |
| 40 | |
| 41 | impl<T: Serialize + DeserializeOwned + Debug + Send + Sync + MessageTarget> PreparedRequest<T> { |
| 42 | pub fn authenticate<P: AuthenticationSourceProvider + 'static + Send + Sync>(mut self, source: P) -> Self { |
| 43 | self.request.append_authentication(source); |
| 44 | |
| 45 | self |
| 46 | } |
| 47 | |
| 48 | pub async fn execute_expect<R: DeserializeOwned>( |
| 49 | self, |
| 50 | pool: &DaemonConnectionPool, |
| 51 | ) -> Result<R, Error> { |
| 52 | let mut socket = pool.0.get().await.unwrap(); |
| 53 | |
| 54 | let payload = serde_json::to_vec(&self.request.into_payload()).unwrap(); |
| 55 | |
| 56 | socket.send(Message::Binary(payload)).await?; |
| 57 | |
| 58 | while let Some(message) = socket.next().await { |
| 59 | let payload = match message? { |
| 60 | Message::Binary(payload) => payload, |
| 61 | _ => { |
| 62 | continue; |
| 63 | } |
| 64 | }; |
| 65 | |
| 66 | let as_target = serde_json::from_slice::<R>(&payload).map_err(|e| Error::from(e)); |
| 67 | |
| 68 | if as_target.is_err() { |
| 69 | // Maybe we got an error payload? |
| 70 | if let Ok(error_payload) = serde_json::from_slice::<ConnectionError>(&payload) { |
| 71 | return Err(error_payload.into()); |
| 72 | } |
| 73 | } else { |
| 74 | // We did not get an error payload, forward the deserialization error from the |
| 75 | // expected type |
| 76 | return as_target; |
| 77 | } |
| 78 | } |
| 79 | |
| 80 | panic!() |
| 81 | } |
| 82 | } |