···1+[package]
2+authors = ["videah <videah@selfish.systems>"]
3+name = "jetstream-oxide"
4+version = "0.1.1"
5+edition = "2021"
6+license = "MIT"
7+description = "Library for easily interacting with and consuming the Bluesky Jetstream service."
8+repository = "https://github.com/videah/jetstream-oxide"
9+readme = "README.md"
10+11+[dependencies]
12+async-trait = "0.1.83"
13+atrium-api = { version = "0.24.7", default-features = false, features = [
14+ "namespace-appbsky",
15+] }
16+tokio = { version = "1.41.1", features = ["full", "sync", "time"] }
17+tokio-tungstenite = { version = "0.24.0", features = [
18+ "connect",
19+ "native-tls-vendored",
20+ "url",
21+] }
22+futures-util = "0.3.31"
23+url = "2.5.4"
24+serde = { version = "1.0.215", features = ["derive"] }
25+serde_json = "1.0.132"
26+chrono = "0.4.38"
27+zstd = "0.13.2"
28+thiserror = "2.0.3"
29+flume = "0.11.1"
30+log = "0.4.22"
31+tokio-util = "0.7.13"
32+33+[dev-dependencies]
34+anyhow = "1.0.93"
35+clap = { version = "4.5.20", features = ["derive"] }
+21
jetstream/LICENSE
···000000000000000000000
···1+MIT License
2+3+Copyright (c) 2024 videah
4+5+Permission is hereby granted, free of charge, to any person obtaining a copy
6+of this software and associated documentation files (the "Software"), to deal
7+in the Software without restriction, including without limitation the rights
8+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9+copies of the Software, and to permit persons to whom the Software is
10+furnished to do so, subject to the following conditions:
11+12+The above copyright notice and this permission notice shall be included in all
13+copies or substantial portions of the Software.
14+15+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
21+SOFTWARE.
···1+# jetstream-oxide
2+3+[](https://crates.io/crates/jetstream-oxide)
4+[](https://docs.rs/jetstream-oxide/latest/jetstream_oxide)
5+6+A typed Rust library for easily interacting with and consuming the
7+Bluesky [Jetstream](https://github.com/bluesky-social/jetstream)
8+service.
9+10+```rust
11+let config = JetstreamConfig {
12+ endpoint: DefaultJetstreamEndpoints::USEastOne.into(),
13+ compression: JetstreamCompression::Zstd,
14+ ..Default::default()
15+};
16+17+let jetstream = JetstreamConnector::new(config).unwrap();
18+let receiver = jetstream.connect().await?;
19+20+while let Ok(event) = receiver.recv_async().await {
21+ if let Commit(commit) = event {
22+ match commit {
23+ CommitEvent::Create { info, commit } => {
24+ println!("Received create event: {:#?}", info);
25+ }
26+ CommitEvent::Update { info, commit } => {
27+ println!("Received update event: {:#?}", info);
28+ }
29+ CommitEvent::Delete { info, commit } => {
30+ println!("Received delete event: {:#?}", info);
31+ }
32+ }
33+ }
34+}
35+```
36+37+## Example
38+39+A small example CLI utility to show how to use this crate can be found in the `examples` directory. To run it, use the
40+following command:
41+42+```sh
43+cargo run --example basic -- --nsid "app.bsky.feed.post"
44+```
45+46+This will display a real-time feed of every single post that is being made or deleted in the entire Bluesky network,
47+right in your terminal!
48+49+You can filter it down to just specific accounts like this:
50+51+```sh
52+cargo run --example basic -- \
53+--nsid "app.bsky.feed.post" \
54+--did "did:plc:inze6wrmsm7pjl7yta3oig77"
55+```
56+57+This listens for posts that *I personally make*. You can substitute your own DID and make a few test posts yourself if
58+you'd
59+like of course!
···1+//! Various error types.
2+use std::io;
3+4+use thiserror::Error;
5+6+/// Possible errors that can occur when a [JetstreamConfig](crate::JetstreamConfig) that is passed
7+/// to a [JetstreamConnector](crate::JetstreamConnector) is invalid.
8+#[derive(Error, Debug)]
9+pub enum ConfigValidationError {
10+ #[error("too many wanted collections: {0} > 100")]
11+ TooManyWantedCollections(usize),
12+ #[error("too many wanted DIDs: {0} > 10,000")]
13+ TooManyDids(usize),
14+}
15+16+/// Possible errors that can occur in the process of connecting to a Jetstream instance over
17+/// WebSockets.
18+///
19+/// See [JetstreamConnector::connect](crate::JetstreamConnector::connect).
20+#[derive(Error, Debug)]
21+pub enum ConnectionError {
22+ #[error("invalid endpoint: {0}")]
23+ InvalidEndpoint(#[from] url::ParseError),
24+ #[error("failed to connect to Jetstream instance: {0}")]
25+ WebSocketFailure(#[from] tokio_tungstenite::tungstenite::Error),
26+ #[error("the Jetstream config is invalid (this really should not happen here): {0}")]
27+ InvalidConfig(#[from] ConfigValidationError),
28+}
29+30+/// Possible errors that can occur when receiving events from a Jetstream instance over WebSockets.
31+///
32+/// See [websocket_task](crate::websocket_task).
33+#[derive(Error, Debug)]
34+pub enum JetstreamEventError {
35+ #[error("received websocket message that could not be deserialized as JSON: {0}")]
36+ ReceivedMalformedJSON(#[from] serde_json::Error),
37+ #[error("failed to load built-in zstd dictionary for decoding: {0}")]
38+ CompressionDictionaryError(io::Error),
39+ #[error("failed to decode zstd-compressed message: {0}")]
40+ CompressionDecoderError(io::Error),
41+ #[error("all receivers were dropped but the websocket connection failed to close cleanly")]
42+ WebSocketCloseFailure,
43+}
+40
jetstream/src/events/account.rs
···0000000000000000000000000000000000000000
···1+use chrono::Utc;
2+use serde::Deserialize;
3+4+use crate::{
5+ events::EventInfo,
6+ exports,
7+};
8+9+/// An event representing a change to an account.
10+#[derive(Deserialize, Debug)]
11+pub struct AccountEvent {
12+ /// Basic metadata included with every event.
13+ #[serde(flatten)]
14+ pub info: EventInfo,
15+ /// Account specific data bundled with this event.
16+ pub account: AccountData,
17+}
18+19+/// Account specific data bundled with an account event.
20+#[derive(Deserialize, Debug)]
21+pub struct AccountData {
22+ /// Whether the account is currently active.
23+ pub active: bool,
24+ /// The DID of the account.
25+ pub did: exports::Did,
26+ pub seq: u64,
27+ pub time: chrono::DateTime<Utc>,
28+ /// If `active` is `false` this will be present to explain why the account is inactive.
29+ pub status: Option<AccountStatus>,
30+}
31+32+/// The possible reasons an account might be listed as inactive.
33+#[derive(Deserialize, Debug)]
34+#[serde(rename_all = "lowercase")]
35+pub enum AccountStatus {
36+ Deactivated,
37+ Deleted,
38+ Suspended,
39+ TakenDown,
40+}
···1+use atrium_api::record::KnownRecord;
2+use serde::Deserialize;
3+4+use crate::{
5+ events::EventInfo,
6+ exports,
7+};
8+9+/// An event representing a repo commit, which can be a `create`, `update`, or `delete` operation.
10+#[derive(Deserialize, Debug)]
11+#[serde(untagged, rename_all = "snake_case")]
12+pub enum CommitEvent {
13+ Create {
14+ #[serde(flatten)]
15+ info: EventInfo,
16+ commit: CommitData,
17+ },
18+ Update {
19+ #[serde(flatten)]
20+ info: EventInfo,
21+ commit: CommitData,
22+ },
23+ Delete {
24+ #[serde(flatten)]
25+ info: EventInfo,
26+ commit: CommitInfo,
27+ },
28+}
29+30+/// The type of commit operation that was performed.
31+#[derive(Deserialize, Debug)]
32+#[serde(rename_all = "snake_case")]
33+pub enum CommitType {
34+ Create,
35+ Update,
36+ Delete,
37+}
38+39+/// Basic commit specific info bundled with every event, also the only data included with a `delete`
40+/// operation.
41+#[derive(Deserialize, Debug)]
42+pub struct CommitInfo {
43+ /// The type of commit operation that was performed.
44+ pub operation: CommitType,
45+ pub rev: String,
46+ pub rkey: String,
47+ /// The NSID of the record type that this commit is associated with.
48+ pub collection: exports::Nsid,
49+}
50+51+/// Detailed data bundled with a commit event. This data is only included when the event is
52+/// `create` or `update`.
53+#[derive(Deserialize, Debug)]
54+pub struct CommitData {
55+ #[serde(flatten)]
56+ pub info: CommitInfo,
57+ /// The CID of the record that was operated on.
58+ pub cid: exports::Cid,
59+ /// The record that was operated on.
60+ pub record: KnownRecord,
61+}
+28
jetstream/src/events/identity.rs
···0000000000000000000000000000
···1+use chrono::Utc;
2+use serde::Deserialize;
3+4+use crate::{
5+ events::EventInfo,
6+ exports,
7+};
8+9+/// An event representing a change to an identity.
10+#[derive(Deserialize, Debug)]
11+pub struct IdentityEvent {
12+ /// Basic metadata included with every event.
13+ #[serde(flatten)]
14+ pub info: EventInfo,
15+ /// Identity specific data bundled with this event.
16+ pub identity: IdentityData,
17+}
18+19+/// Identity specific data bundled with an identity event.
20+#[derive(Deserialize, Debug)]
21+pub struct IdentityData {
22+ /// The DID of the identity.
23+ pub did: exports::Did,
24+ /// The handle associated with the identity.
25+ pub handle: Option<exports::Handle>,
26+ pub seq: u64,
27+ pub time: chrono::DateTime<Utc>,
28+}
+31
jetstream/src/events/mod.rs
···0000000000000000000000000000000
···1+pub mod account;
2+pub mod commit;
3+pub mod identity;
4+5+use serde::Deserialize;
6+7+use crate::exports;
8+9+/// Basic data that is included with every event.
10+#[derive(Deserialize, Debug)]
11+pub struct EventInfo {
12+ pub did: exports::Did,
13+ pub time_us: u64,
14+ pub kind: EventKind,
15+}
16+17+#[derive(Deserialize, Debug)]
18+#[serde(untagged)]
19+pub enum JetstreamEvent {
20+ Commit(commit::CommitEvent),
21+ Identity(identity::IdentityEvent),
22+ Account(account::AccountEvent),
23+}
24+25+#[derive(Deserialize, Debug)]
26+#[serde(rename_all = "snake_case")]
27+pub enum EventKind {
28+ Commit,
29+ Identity,
30+ Account,
31+}
+8
jetstream/src/exports.rs
···00000000
···1+//! Useful exports for third-party crates used by this project.
2+3+pub use atrium_api::types::string::{
4+ Cid,
5+ Did,
6+ Handle,
7+ Nsid,
8+};
···1+pub mod error;
2+pub mod events;
3+pub mod exports;
4+5+use std::{
6+ io::{Cursor, Read},
7+ sync::Arc,
8+ time::Duration,
9+};
10+11+use chrono::Utc;
12+use futures_util::{stream::StreamExt, SinkExt};
13+use tokio::{net::TcpStream, sync::Mutex};
14+use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream};
15+use tokio_util::sync::CancellationToken;
16+use url::Url;
17+use zstd::dict::DecoderDictionary;
18+19+use crate::{
20+ error::{ConfigValidationError, ConnectionError, JetstreamEventError},
21+ events::JetstreamEvent,
22+};
23+24+/// The Jetstream endpoints officially provided by Bluesky themselves.
25+///
26+/// There are no guarantees that these endpoints will always be available, but you are free
27+/// to run your own Jetstream instance in any case.
28+pub enum DefaultJetstreamEndpoints {
29+ /// `jetstream1.us-east.bsky.network`
30+ USEastOne,
31+ /// `jetstream2.us-east.bsky.network`
32+ USEastTwo,
33+ /// `jetstream1.us-west.bsky.network`
34+ USWestOne,
35+ /// `jetstream2.us-west.bsky.network`
36+ USWestTwo,
37+}
38+39+impl From<DefaultJetstreamEndpoints> for String {
40+ fn from(endpoint: DefaultJetstreamEndpoints) -> Self {
41+ match endpoint {
42+ DefaultJetstreamEndpoints::USEastOne => {
43+ "wss://jetstream1.us-east.bsky.network/subscribe".to_owned()
44+ }
45+ DefaultJetstreamEndpoints::USEastTwo => {
46+ "wss://jetstream2.us-east.bsky.network/subscribe".to_owned()
47+ }
48+ DefaultJetstreamEndpoints::USWestOne => {
49+ "wss://jetstream1.us-west.bsky.network/subscribe".to_owned()
50+ }
51+ DefaultJetstreamEndpoints::USWestTwo => {
52+ "wss://jetstream2.us-west.bsky.network/subscribe".to_owned()
53+ }
54+ }
55+ }
56+}
57+58+/// The maximum number of wanted collections that can be requested on a single Jetstream connection.
59+const MAX_WANTED_COLLECTIONS: usize = 100;
60+/// The maximum number of wanted DIDs that can be requested on a single Jetstream connection.
61+const MAX_WANTED_DIDS: usize = 10_000;
62+63+/// The custom `zstd` dictionary used for decoding compressed Jetstream messages.
64+///
65+/// Sourced from the [official Bluesky Jetstream repo.](https://github.com/bluesky-social/jetstream/tree/main/pkg/models)
66+const JETSTREAM_ZSTD_DICTIONARY: &[u8] = include_bytes!("../zstd/dictionary");
67+68+/// A receiver channel for consuming Jetstream events.
69+pub type JetstreamReceiver = flume::Receiver<JetstreamEvent>;
70+71+/// An internal sender channel for sending Jetstream events to [JetstreamReceiver]'s.
72+type JetstreamSender = flume::Sender<JetstreamEvent>;
73+74+/// A wrapper connector type for working with a WebSocket connection to a Jetstream instance to
75+/// receive and consume events. See [JetstreamConnector::connect] for more info.
76+pub struct JetstreamConnector {
77+ /// The configuration for the Jetstream connection.
78+ config: JetstreamConfig,
79+}
80+81+pub enum JetstreamCompression {
82+ /// No compression, just raw plaintext JSON.
83+ None,
84+ /// Use the `zstd` compression algorithm, which can result in a ~56% smaller messages on
85+ /// average. See [here](https://github.com/bluesky-social/jetstream?tab=readme-ov-file#compression) for more info.
86+ Zstd,
87+}
88+89+impl From<JetstreamCompression> for bool {
90+ fn from(compression: JetstreamCompression) -> Self {
91+ match compression {
92+ JetstreamCompression::None => false,
93+ JetstreamCompression::Zstd => true,
94+ }
95+ }
96+}
97+98+pub struct JetstreamConfig {
99+ /// A Jetstream endpoint to connect to with a WebSocket Scheme i.e.
100+ /// `wss://jetstream1.us-east.bsky.network/subscribe`.
101+ pub endpoint: String,
102+ /// A list of collection [NSIDs](https://atproto.com/specs/nsid) to filter events for.
103+ ///
104+ /// An empty list will receive events for *all* collections.
105+ ///
106+ /// Regardless of desired collections, all subscribers receive
107+ /// [AccountEvent](events::account::AccountEvent) and
108+ /// [IdentityEvent](events::identity::Identity) events.
109+ pub wanted_collections: Vec<exports::Nsid>,
110+ /// A list of repo [DIDs](https://atproto.com/specs/did) to filter events for.
111+ ///
112+ /// An empty list will receive events for *all* repos, which is a lot of events!
113+ pub wanted_dids: Vec<exports::Did>,
114+ /// The compression algorithm to request and use for the WebSocket connection (if any).
115+ pub compression: JetstreamCompression,
116+ /// An optional timestamp to begin playback from.
117+ ///
118+ /// An absent cursor or a cursor from the future will result in live-tail operation.
119+ ///
120+ /// When reconnecting, use the time_us from your most recently processed event and maybe
121+ /// provide a negative buffer (i.e. subtract a few seconds) to ensure gapless playback.
122+ pub cursor: Option<chrono::DateTime<Utc>>,
123+}
124+125+impl Default for JetstreamConfig {
126+ fn default() -> Self {
127+ JetstreamConfig {
128+ endpoint: DefaultJetstreamEndpoints::USEastOne.into(),
129+ wanted_collections: Vec::new(),
130+ wanted_dids: Vec::new(),
131+ compression: JetstreamCompression::None,
132+ cursor: None,
133+ }
134+ }
135+}
136+137+impl JetstreamConfig {
138+ /// Constructs a new endpoint URL with the given [JetstreamConfig] applied.
139+ pub fn construct_endpoint(&self, endpoint: &str) -> Result<Url, url::ParseError> {
140+ let did_search_query = self
141+ .wanted_dids
142+ .iter()
143+ .map(|s| ("wantedDids", s.to_string()));
144+145+ let collection_search_query = self
146+ .wanted_collections
147+ .iter()
148+ .map(|s| ("wantedCollections", s.to_string()));
149+150+ let compression = (
151+ "compress",
152+ match self.compression {
153+ JetstreamCompression::None => "false".to_owned(),
154+ JetstreamCompression::Zstd => "true".to_owned(),
155+ },
156+ );
157+158+ let cursor = self
159+ .cursor
160+ .map(|c| ("cursor", c.timestamp_micros().to_string()));
161+162+ let params = did_search_query
163+ .chain(collection_search_query)
164+ .chain(std::iter::once(compression))
165+ .chain(cursor)
166+ .collect::<Vec<(&str, String)>>();
167+168+ Url::parse_with_params(endpoint, params)
169+ }
170+171+ /// Validates the configuration to make sure it is within the limits of the Jetstream API.
172+ ///
173+ /// # Constants
174+ /// The following constants are used to validate the configuration and should only be changed
175+ /// if the Jetstream API has itself changed.
176+ /// - [MAX_WANTED_COLLECTIONS]
177+ /// - [MAX_WANTED_DIDS]
178+ pub fn validate(&self) -> Result<(), ConfigValidationError> {
179+ let collections = self.wanted_collections.len();
180+ let dids = self.wanted_dids.len();
181+182+ if collections > MAX_WANTED_COLLECTIONS {
183+ return Err(ConfigValidationError::TooManyWantedCollections(collections));
184+ }
185+186+ if dids > MAX_WANTED_DIDS {
187+ return Err(ConfigValidationError::TooManyDids(dids));
188+ }
189+190+ Ok(())
191+ }
192+}
193+194+impl JetstreamConnector {
195+ /// Create a Jetstream connector with a valid [JetstreamConfig].
196+ ///
197+ /// After creation, you can call [connect] to connect to the provided Jetstream instance.
198+ pub fn new(config: JetstreamConfig) -> Result<Self, ConfigValidationError> {
199+ // We validate the configuration here so any issues are caught early.
200+ config.validate()?;
201+ Ok(JetstreamConnector { config })
202+ }
203+204+ /// Connects to a Jetstream instance as defined in the [JetstreamConfig].
205+ ///
206+ /// A [JetstreamReceiver] is returned which can be used to respond to events. When all instances
207+ /// of this receiver are dropped, the connection and task are automatically closed.
208+ pub async fn connect(&self) -> Result<JetstreamReceiver, ConnectionError> {
209+ // We validate the config again for good measure. Probably not necessary but it can't hurt.
210+ self.config
211+ .validate()
212+ .map_err(ConnectionError::InvalidConfig)?;
213+214+ // TODO: Run some benchmarks and look into using a bounded channel instead.
215+ let (send_channel, receive_channel) = flume::unbounded();
216+217+ let configured_endpoint = self
218+ .config
219+ .construct_endpoint(&self.config.endpoint)
220+ .map_err(ConnectionError::InvalidEndpoint)?;
221+222+ tokio::task::spawn(async move {
223+ let max_retries = 10;
224+ let base_delay_ms = 1_000; // 1 second
225+ let max_delay_ms = 30_000; // 30 seconds
226+227+ for retry_attempt in 0..max_retries {
228+ let dict = DecoderDictionary::copy(JETSTREAM_ZSTD_DICTIONARY);
229+230+ if let Ok((ws_stream, _)) = connect_async(&configured_endpoint).await {
231+ let _ = websocket_task(dict, ws_stream, send_channel.clone()).await;
232+ }
233+234+ // Exponential backoff
235+ let delay_ms = base_delay_ms * (2_u64.pow(retry_attempt));
236+237+ log::error!("Connection failed, retrying in {delay_ms}ms...");
238+ tokio::time::sleep(Duration::from_millis(delay_ms.min(max_delay_ms))).await;
239+ log::info!("Attempting to reconnect...")
240+ }
241+ log::error!("Connection retries exhausted. Jetstream is disconnected.");
242+ });
243+244+ Ok(receive_channel)
245+ }
246+}
247+248+/// The main task that handles the WebSocket connection and sends [JetstreamEvent]'s to any
249+/// receivers that are listening for them.
250+async fn websocket_task(
251+ dictionary: DecoderDictionary<'_>,
252+ ws: WebSocketStream<MaybeTlsStream<TcpStream>>,
253+ send_channel: JetstreamSender,
254+) -> Result<(), JetstreamEventError> {
255+ // TODO: Use the write half to allow the user to change configuration settings on the fly.
256+ let (socket_write, mut socket_read) = ws.split();
257+ let shared_socket_write = Arc::new(Mutex::new(socket_write));
258+259+ let ping_cancellation_token = CancellationToken::new();
260+ let mut ping_interval = tokio::time::interval(Duration::from_secs(30));
261+ let ping_cancelled = ping_cancellation_token.clone();
262+ let ping_shared_socket_write = shared_socket_write.clone();
263+ tokio::spawn(async move {
264+ loop {
265+ ping_interval.tick().await;
266+ let false = ping_cancelled.is_cancelled() else {
267+ break;
268+ };
269+ log::trace!("Sending ping");
270+ match ping_shared_socket_write
271+ .lock()
272+ .await
273+ .send(Message::Ping("ping".as_bytes().to_vec()))
274+ .await
275+ {
276+ Ok(_) => (),
277+ Err(error) => {
278+ log::error!("Ping failed: {error}");
279+ break;
280+ }
281+ }
282+ }
283+ });
284+285+ let mut closing_connection = false;
286+ loop {
287+ match socket_read.next().await {
288+ Some(Ok(message)) => {
289+ match message {
290+ Message::Text(json) => {
291+ let event = serde_json::from_str::<JetstreamEvent>(&json)
292+ .map_err(JetstreamEventError::ReceivedMalformedJSON)?;
293+294+ if send_channel.send(event).is_err() {
295+ // We can assume that all receivers have been dropped, so we can close the
296+ // connection and exit the task.
297+ log::info!(
298+ "All receivers for the Jetstream connection have been dropped, closing connection."
299+ );
300+ closing_connection = true;
301+ }
302+ }
303+ Message::Binary(zstd_json) => {
304+ let mut cursor = Cursor::new(zstd_json);
305+ let mut decoder = zstd::stream::Decoder::with_prepared_dictionary(
306+ &mut cursor,
307+ &dictionary,
308+ )
309+ .map_err(JetstreamEventError::CompressionDictionaryError)?;
310+311+ let mut json = String::new();
312+ decoder
313+ .read_to_string(&mut json)
314+ .map_err(JetstreamEventError::CompressionDecoderError)?;
315+316+ let event = serde_json::from_str::<JetstreamEvent>(&json)
317+ .map_err(JetstreamEventError::ReceivedMalformedJSON)?;
318+319+ if send_channel.send(event).is_err() {
320+ // We can assume that all receivers have been dropped, so we can close the
321+ // connection and exit the task.
322+ log::info!(
323+ "All receivers for the Jetstream connection have been dropped, closing connection..."
324+ );
325+ closing_connection = true;
326+ }
327+ }
328+ Message::Ping(vec) => {
329+ log::trace!("Ping recieved, responding");
330+ _ = shared_socket_write
331+ .lock()
332+ .await
333+ .send(Message::Pong(vec))
334+ .await;
335+ }
336+ Message::Close(close_frame) => {
337+ if let Some(close_frame) = close_frame {
338+ let reason = close_frame.reason;
339+ let code = close_frame.code;
340+ log::trace!("Connection closed. Reason: {reason}, Code: {code}");
341+ }
342+ }
343+ Message::Pong(pong) => {
344+ let pong_payload =
345+ String::from_utf8(pong).unwrap_or("Invalid payload".to_string());
346+ log::trace!("Pong recieved. Payload: {pong_payload}");
347+ }
348+ Message::Frame(_) => (),
349+ }
350+ }
351+ Some(Err(error)) => {
352+ log::error!("Web socket error: {error}");
353+ ping_cancellation_token.cancel();
354+ closing_connection = true;
355+ }
356+ None => {
357+ log::error!("No web socket result");
358+ ping_cancellation_token.cancel();
359+ closing_connection = true;
360+ }
361+ }
362+ if closing_connection {
363+ _ = shared_socket_write.lock().await.close().await;
364+ return Ok(());
365+ }
366+ }
367+}