+49

crates/gordian-jetstream/src/client.rs

reviewed

··· 36 36 } 37 37 38 38 /// Add a DID to the Jetstream filters. 39 39 + /// 40 40 + /// # Errors 41 41 + /// 42 42 + /// Returns an error if the Jetstream task is no longer active. 43 43 + /// 44 44 + /// # Panics 45 45 + /// 46 46 + /// Panics if the [`Mutex`] for the client options has been poisoned. 47 47 + /// 39 48 pub async fn add_did(&self, did: impl Into<OwnedDid>) -> Result<(), JetstreamClientError> { 40 49 if self.options.lock().unwrap().add_did(did.into())? { 41 50 // The DID is new to the client, notify the task to update. ··· 54 45 } 55 46 56 47 /// Remove a DID from the Jetstream filters. 48 48 + /// 49 49 + /// # Errors 50 50 + /// 51 51 + /// Returns an error if the Jetstream task is no longer active. 52 52 + /// 53 53 + /// # Panics 54 54 + /// 55 55 + /// Panics if the [`Mutex`] for the client options has been poisoned. 56 56 + /// 57 57 pub async fn remove_did(&self, did: impl Into<OwnedDid>) -> Result<(), JetstreamClientError> { 58 58 if self.options.lock().unwrap().remove_did(&did.into()) { 59 59 self.update_task().await?; ··· 71 53 } 72 54 73 55 /// Add a collection to the Jetstream filters. 56 56 + /// 57 57 + /// # Errors 58 58 + /// 59 59 + /// Returns an error if the Jetstream task is no longer active. 60 60 + /// 61 61 + /// # Panics 62 62 + /// 63 63 + /// Panics if the [`Mutex`] for the client options has been poisoned. 64 64 + /// 74 65 pub async fn add_collection( 75 66 &self, 76 67 collection: impl Into<Box<Nsid>>, ··· 97 70 } 98 71 99 72 /// Remove a collection from the Jetstream filters. 73 73 + /// 74 74 + /// # Errors 75 75 + /// 76 76 + /// Returns an error if the Jetstream task is no longer active. 77 77 + /// 78 78 + /// # Panics 79 79 + /// 80 80 + /// Panics if the [`Mutex`] for the client options has been poisoned. 81 81 + /// 100 82 pub async fn remove_collection( 101 83 &self, 102 84 collection: impl Into<Box<Nsid>>, ··· 126 90 self.metrics.export() 127 91 } 128 92 93 93 + /// Shutdown the Jetstream client. 94 94 + /// 95 95 + /// # Errors 96 96 + /// 97 97 + /// Returns an error if the Jetstream task is no longer active. 98 98 + /// 129 99 pub async fn shutdown(self) -> Result<(), JetstreamClientError> { 130 100 let (command, complete) = ClientCommand::shutdown(); 131 101 self.client_tx.send(command)?; ··· 257 215 } 258 216 259 217 impl<'a> JetstreamEvent { 218 218 + /// Deserialize the event. 219 219 + /// 220 220 + /// # Errors 221 221 + /// 222 222 + /// Returns an error if the event cannot be deserialized from it JSON 223 223 + /// representation. 224 224 + /// 260 225 pub fn deserialize(&'a self) -> Result<Event<'a>, serde_json::Error> { 261 226 let value = serde_json::from_slice(&self.bytes)?; 262 227 Ok(value)

+2 -4

crates/gordian-jetstream/src/client_config.rs

reviewed

··· 13 13 }; 14 14 15 15 #[derive(Clone, Debug, Default)] 16 16 + #[must_use] 16 17 pub struct JetstreamConfig { 17 18 pub client_options: ClientOptions, 18 19 pub subscriber_options: SubscriberOptions, ··· 22 21 impl JetstreamConfig { 23 22 /// Create default a [`JetstreamConfig`] connecting to jetstream{1,2}.us-east.bsky.network 24 23 /// instances. 25 25 - #[must_use] 26 24 pub fn us_east() -> Self { 27 25 Self { 28 26 client_options: ClientOptions { ··· 37 37 38 38 /// Create default a [`JetstreamConfig`] connecting to jetstream{1,2}.us-west.bsky.network 39 39 /// instances. 40 40 - #[must_use] 41 40 pub fn us_west() -> Self { 42 41 Self { 43 42 client_options: ClientOptions { ··· 50 51 } 51 52 } 52 53 53 53 - #[must_use] 54 54 + #[must_use] 54 55 pub fn connect(self) -> (JetstreamClient, JetstreamReceiver, JetstreamTask) { 55 56 let (event_tx, event_rx) = flume::bounded(8); 56 57 let (client_tx, client_rx) = flume::bounded(8); ··· 77 78 (client, receiver, task) 78 79 } 79 80 80 80 - #[must_use] 81 81 pub const fn with_cursor(mut self, cursor: Option<u128>) -> Self { 82 82 self.subscriber_options.cursor = cursor; 83 83 self

+5 -10

crates/gordian-jetstream/src/de.rs

reviewed

··· 178 178 #[must_use] 179 179 pub const fn ts(&self) -> OffsetDateTime { 180 180 match self { 181 181 - Self::Create(commit) => commit.ts, 182 182 - Self::Update(commit) => commit.ts, 181 181 + Self::Create(commit) | Self::Update(commit) => commit.ts, 183 182 Self::Delete(commit) => commit.ts, 184 183 } 185 184 } ··· 186 187 #[must_use] 187 188 pub const fn did(&'a self) -> &'a Did { 188 189 match self { 189 189 - Self::Create(commit) => commit.did, 190 190 - Self::Update(commit) => commit.did, 190 190 + Self::Create(commit) | Self::Update(commit) => commit.did, 191 191 Self::Delete(commit) => commit.did, 192 192 } 193 193 } ··· 200 202 #[must_use] 201 203 pub const fn collection(&self) -> &str { 202 204 match self { 203 203 - Self::Create(commit) => commit.collection, 204 204 - Self::Update(commit) => commit.collection, 205 205 + Self::Create(commit) | Self::Update(commit) => commit.collection, 205 206 Self::Delete(commit) => commit.collection, 206 207 } 207 208 } ··· 208 211 #[must_use] 209 212 pub const fn rkey(&self) -> &str { 210 213 match self { 211 211 - Self::Create(commit) => commit.rkey, 212 212 - Self::Update(commit) => commit.rkey, 214 214 + Self::Create(commit) | Self::Update(commit) => commit.rkey, 213 215 Self::Delete(commit) => commit.rkey, 214 216 } 215 217 } ··· 216 220 #[must_use] 217 221 pub const fn rev(&self) -> &str { 218 222 match self { 219 219 - Self::Create(commit) => commit.rev, 220 220 - Self::Update(commit) => commit.rev, 223 223 + Self::Create(commit) | Self::Update(commit) => commit.rev, 221 224 Self::Delete(commit) => commit.rev, 222 225 } 223 226 }

+7 -2

crates/gordian-jetstream/src/metrics.rs

reviewed

··· 32 32 } 33 33 34 34 impl Metrics { 35 35 - #[must_use] 35 35 + #[must_use] 36 36 pub fn new() -> Self { 37 37 Self::default() 38 38 } 39 39 40 40 /// Create a clone of the metrics data. 41 41 - #[must_use] 41 41 + /// 42 42 + /// # Panics 43 43 + /// 44 44 + /// Panics if the contained mutex has been poisoned 45 45 + /// 46 46 + #[must_use] 42 47 pub fn export(&self) -> MetricsData { 43 48 self.inner.lock().unwrap().clone() 44 49 }

+19 -1

crates/gordian-jetstream/src/subscriber_options.rs

reviewed

··· 48 48 /// Returns an error if the maximum number of subscribed collections has been reached; `Ok(true)` 49 49 /// if the collection was newly added to the set, or `Ok(false)` if the colletion was already in the 50 50 /// the set. 51 51 + /// 52 52 + /// # Errors 53 53 + /// 54 54 + /// Returns an error if adding `collection` would cause [`SubscriberOptions`] to exceed the maximum 55 55 + /// number of subscribed collections. 56 56 + /// 51 57 pub fn add_collection(&mut self, collection: Box<Nsid>) -> Result<bool, Box<Nsid>> { 52 58 if self.wanted_collections.len() == MAX_WANTED_COLLECTIONS 53 59 && !self.wanted_collections.contains(&collection) ··· 73 67 /// Returns an error if the maximum number of subscribed DIDs has been reached; `Ok(true)` 74 68 /// if the DID was newly added to the set, or `Ok(false)` if the DID was already in the 75 69 /// the set. 70 70 + /// 71 71 + /// # Errors 72 72 + /// 73 73 + /// Returns an error if adding `did` would cause [`SubscriberOptions`] to exceed the maximum 74 74 + /// number of subscribed DIDs. 75 75 + /// 76 76 pub fn add_did(&mut self, did: OwnedDid) -> Result<bool, OwnedDid> { 77 77 if self.wanted_dids.len() == MAX_WANTED_DIDS && !self.wanted_dids.contains(&did) { 78 78 return Err(did); ··· 93 81 94 82 /// Get the normalized maximum message size. 95 83 #[must_use] 96 96 - pub fn max_message_size(&self) -> i64 { 84 84 + pub const fn max_message_size(&self) -> i64 { 97 85 normalize_max_message_size(self.max_message_size_bytes) 98 86 } 99 87 ··· 178 166 Ok(super::normalize_max_message_size(value)) 179 167 } 180 168 169 169 + #[allow(clippy::trivially_copy_pass_by_ref)] 181 170 pub fn serialize<S>(value: &i64, serializer: S) -> Result<S::Ok, S::Error> 182 171 where 183 172 S: Serializer, ··· 203 190 204 191 impl SubscriberSourcedMessage<'_> { 205 192 /// Serialize the [`SubscriberSourcedMessage`] to JSON. 193 193 + /// 194 194 + /// # Panics 195 195 + /// 196 196 + /// Panics if [`SubscriberSourcedMessage`] cannot be serialized as JSON. 197 197 + /// 206 198 #[must_use] 207 199 pub fn to_json(&self) -> String { 208 200 serde_json::to_string(self).expect("SubscriberSourcedMessage should be serializable")

+24 -16

crates/gordian-jetstream/src/task.rs

reviewed

··· 45 45 self: std::pin::Pin<&mut Self>, 46 46 cx: &mut std::task::Context<'_>, 47 47 ) -> std::task::Poll<Self::Output> { 48 48 - match self.get_mut().0 { 49 49 - Some(ref mut fut) => fut.as_mut().poll(cx), 50 50 - None => unreachable!(), 51 51 - } 48 48 + self.get_mut() 49 49 + .0 50 50 + .as_mut() 51 51 + .map_or_else(|| unreachable!(), |fut| fut.as_mut().poll(cx)) 52 52 } 53 53 } 54 54 ··· 68 68 } 69 69 } 70 70 71 71 + #[derive(Deserialize)] 72 72 + struct PartialEvent<'a> { 73 73 + time_us: u64, 74 74 + #[serde(borrow)] 75 75 + kind: &'a str, 76 76 + } 77 77 + 78 78 + #[allow(clippy::too_many_lines)] 71 79 pub async fn jetstream_subscriber( 72 80 event_tx: flume::Sender<Bytes>, 73 81 client_rx: flume::Receiver<ClientCommand>, ··· 84 76 subscriber_options: Arc<Mutex<SubscriberOptions>>, 85 77 shutdown: CancellationToken, 86 78 ) { 87 87 - let mut state = State { metrics }; 79 79 + let state = State { metrics }; 88 80 let mut reconnect_backoff = client_options.reconnect_backoff_min; 89 81 90 82 let mut instance_idx = fastrand::usize(0..client_options.instances.len()); ··· 106 98 #[cfg(not(feature = "zstd"))] 107 99 let request = ClientRequestBuilder::new(uri); 108 100 109 109 - let (socket, _) = match shutdown.run_until_cancelled(connect_async(request)).await { 101 101 + let (socket, _) = match shutdown 102 102 + .run_until_cancelled(Box::pin(connect_async(request))) 103 103 + .await 104 104 + { 110 105 Some(Ok(socket)) => { 111 106 reconnect_backoff = client_options.reconnect_backoff_min; 112 107 socket ··· 150 139 151 140 loop { 152 141 let message = tokio::select! { 153 153 - Some(Ok(outcome)) = shutdown.run_until_cancelled(handle_read_socket(&mut read, &mut state)) => { 142 142 + Some(Ok(outcome)) = shutdown.run_until_cancelled(handle_read_socket(&mut read, &state)) => { 154 143 match outcome { 155 144 ReadOutcome::Event(message) => message, 156 145 ReadOutcome::Ping(payload) => { ··· 197 186 else => break, 198 187 }; 199 188 200 200 - #[derive(Deserialize)] 201 201 - struct PartialEvent<'a> { 202 202 - time_us: u64, 203 203 - #[serde(borrow)] 204 204 - kind: &'a str, 205 205 - } 206 206 - 207 189 // Deserialize just the event timestamp and event kind. 208 190 let new_cursor = match serde_json::from_slice::<PartialEvent>(&message) { 209 191 Ok(event) => { ··· 216 212 state.metrics.modify(|mut data| data.messages_received += 1); 217 213 if let Err(error) = event_tx.send_async(message).await { 218 214 let payload = error.into_inner(); 219 219 - if let Ok(payload) = std::str::from_utf8(&payload) { tracing::error!(%payload, "Failed to dispatch event to channel") } else { tracing::error!(?payload, "Failed to dispatch event to channel") } 215 215 + if let Ok(payload) = std::str::from_utf8(&payload) { 216 216 + tracing::error!(%payload, "Failed to dispatch event to channel"); 217 217 + } else { 218 218 + tracing::error!(?payload, "Failed to dispatch event to channel"); 219 219 + } 220 220 break 'outer; 221 221 } 222 222 ··· 243 235 244 236 async fn handle_read_socket<S>( 245 237 stream: &mut S, 246 246 - state: &mut State, 238 238 + state: &State, 247 239 ) -> Result<ReadOutcome, TungsteniteError> 248 240 where 249 241 S: StreamExt<Item = Result<Message, TungsteniteError>> + Unpin,