fix(sdk): being unable to add a new listener after dropping old ones

Author: Enduriel

filen-sdk-rs/src/auth/mod.rs

@@ -402,6 +402,10 @@ impl Client {
 		self.http_client.clone()
 	}
 
+	pub(crate) fn arc_client_ref(&self) -> &Arc<AuthClient> {
+		&self.http_client
+	}
+
 	pub fn crypter(&self) -> Arc<impl MetaCrypter> {
 		self.auth_info
 			.read()

filen-sdk-rs/src/socket/native.rs

@@ -27,16 +27,19 @@ impl Client {
 		event_types: Option<Vec<Cow<'static, str>>>,
 	) -> Result<ListenerHandle, Error> {
 		let request_sender = {
-			let mut socket_handle = self.socket_handle.lock().unwrap();
-			socket_handle.get_request_sender(self.arc_client())
+			let mut socket_handle = self.socket_handle.lock().unwrap_or_else(|e| e.into_inner());
+			socket_handle.get_request_sender(self.arc_client_ref())
 		};
 		request_sender
 			.add_event_listener(callback, event_types)
 			.await
 	}
 
 	pub fn is_socket_connected(&self) -> bool {
-		self.socket_handle.lock().unwrap().request_sender.is_some()
+		self.socket_handle
+			.lock()
+			.unwrap_or_else(|e| e.into_inner())
+			.is_connected()
 	}
 
 	// we need to expose this for v3 because most of the returned events are encrypted
@@ -54,7 +57,7 @@ impl Client {
 #[derive(Default)]
 pub(crate) struct WebSocketHandle {
 	// closes websocket thread on drop
-	request_sender: Option<tokio::sync::mpsc::Sender<SocketRequest>>,
+	request_sender: Option<tokio::sync::mpsc::WeakSender<SocketRequest>>,
 }
 
 struct RequestSender(tokio::sync::mpsc::Sender<SocketRequest>);
@@ -93,15 +96,21 @@ impl RequestSender {
 }
 
 impl WebSocketHandle {
-	fn get_request_sender(&mut self, client: Arc<AuthClient>) -> RequestSender {
-		RequestSender(match &self.request_sender {
-			Some(s) => s.clone(),
-			None => {
-				let request_sender = spawn_websocket_thread(client);
-				self.request_sender = Some(request_sender.clone());
-				request_sender
-			}
-		})
+	fn get_request_sender(&mut self, client: &Arc<AuthClient>) -> RequestSender {
+		if let Some(weak_sender) = &self.request_sender
+			&& let Some(sender) = weak_sender.upgrade()
+		{
+			return RequestSender(sender);
+		}
+		let sender = spawn_websocket_thread(Arc::clone(client));
+		self.request_sender = Some(sender.downgrade());
+		RequestSender(sender)
+	}
+
+	fn is_connected(&self) -> bool {
+		self.request_sender
+			.as_ref()
+			.is_some_and(|weak| weak.strong_count() > 0)
 	}
 }
 
@@ -319,7 +328,7 @@ fn handle_message(
 /// Handles a socket request, modifying the listener manager as needed.
 ///
 /// Returns true if there are no more listeners after handling the request.
-fn handle_request(request: SocketRequest, listeners: &mut impl ListenerManagerExt) -> bool {
+fn handle_request(request: SocketRequest, listeners: &mut impl ListenerManagerExt) {
 	match request {
 		SocketRequest::AddListener {
 			id_sender,
@@ -338,7 +347,6 @@ fn handle_request(request: SocketRequest, listeners: &mut impl ListenerManagerEx
 			listeners.remove_listener(idx);
 		}
 	}
-	listeners.is_empty()
 }
 
 /// Handles the initialized websocket connection, processing incoming messages and managing listeners.
@@ -360,11 +368,7 @@ async fn handle_initialized_websocket(
 					should_retry = false;
 					break;
 				};
-				should_retry = !handle_request(request, listeners);
-				if !should_retry {
-					// no more listeners, shutting down websocket task
-					break;
-				}
+				handle_request(request, listeners);
 			}
 			message_result = streams.read.next() => {
 				if let Err((should_continue, error)) = handle_message(message_result, listeners) {

filen-sdk-rs/src/socket/shared.rs

@@ -122,8 +122,8 @@ mod listener_manager {
 			this_id
 		}
 
-		fn remove_listener(&mut self, id: u64) {
-			self.callbacks_mut().remove(&id);
+		fn remove_listener(&mut self, id: u64) -> Option<EventListenerCallback> {
+			let callback = self.callbacks_mut().remove(&id);
 			let mut empty_event_types = Vec::new();
 			for (event_type, callbacks) in self.callbacks_for_event_mut().iter_mut() {
 				callbacks.retain(|&callback_id| callback_id != id);
@@ -136,6 +136,8 @@ mod listener_manager {
 			}
 			self.global_callbacks_mut()
 				.retain(|&callback_id| callback_id != id);
+
+			callback
 		}
 	}
 
@@ -285,7 +287,10 @@ mod listener_manager {
 		}
 
 		fn remove_listener(&mut self, id: u64) {
-			ListenerManagerExtInner::remove_listener(self, id);
+			let callback = ListenerManagerExtInner::remove_listener(self, id);
+			if let Some(callback) = callback {
+				callback(&SocketEvent::Unsubscribed);
+			}
 			self.new_ids.retain(|new_id_struct| new_id_struct.id != id);
 		}
 	}
@@ -369,7 +374,10 @@ mod listener_manager {
 		}
 
 		fn remove_listener(&mut self, id: u64) {
-			ListenerManagerExtInner::remove_listener(self, id);
+			let callback = ListenerManagerExtInner::remove_listener(self, id);
+			if let Some(callback) = callback {
+				callback(&SocketEvent::Unsubscribed);
+			}
 		}
 	}
 }

filen-sdk-rs/src/socket/uniffi/events.rs

@@ -27,6 +27,8 @@ pub enum SocketEvent {
 	AuthFailed,
 	/// Sent when the socket has unexpectedly closed and begins attempting to reconnect
 	Reconnecting,
+	/// Sent when the handle to the event listener has been dropped and the listener is removed
+	Unsubscribed,
 	NewEvent(NewEvent),
 	FileRename(FileRename),
 	FileArchiveRestored(FileArchiveRestored),
@@ -76,6 +78,7 @@ impl From<&BorrowedSocketEvent<'_>> for SocketEvent {
 			BorrowedSocketEvent::AuthSuccess => Self::AuthSuccess,
 			BorrowedSocketEvent::AuthFailed => Self::AuthFailed,
 			BorrowedSocketEvent::Reconnecting => Self::Reconnecting,
+			BorrowedSocketEvent::Unsubscribed => Self::Unsubscribed,
 			BorrowedSocketEvent::NewEvent(e) => Self::NewEvent(e.into()),
 			BorrowedSocketEvent::FileRename(e) => Self::FileRename(e.into()),
 			BorrowedSocketEvent::FileArchiveRestored(e) => Self::FileArchiveRestored(e.into()),

filen-sdk-rs/tests/socket_reconnect.rs

@@ -0,0 +1,106 @@
+use std::{borrow::Cow, time::Duration};
+
+use filen_macros::shared_test_runtime;
+use filen_types::{api::v3::socket::SocketEvent, traits::CowHelpers};
+// separate file because it needs to avoid interference with other tests
+
+async fn await_event<F>(
+	receiver: &mut tokio::sync::mpsc::UnboundedReceiver<SocketEvent<'static>>,
+	mut filter: F,
+	timeout: Duration,
+) -> Result<SocketEvent<'static>, Cow<'static, str>>
+where
+	F: FnMut(&SocketEvent) -> bool,
+{
+	let sleep_until = tokio::time::Instant::now() + timeout;
+	loop {
+		tokio::select! {
+			_ = tokio::time::sleep_until(sleep_until) => {
+				return Err("Timed out waiting for event".into());
+			}
+			event = receiver.recv() => {
+				let event = event.ok_or(Cow::Borrowed("Expected to receive event"))?;
+				if filter(&event) {
+					return Ok(event);
+				}
+			}
+		}
+	}
+}
+
+#[shared_test_runtime]
+async fn test_websocket_disconnect_reconnect() {
+	let client = test_utils::RESOURCES.client().await;
+
+	let (events_sender, mut events_receiver) = tokio::sync::mpsc::unbounded_channel();
+
+	assert!(!client.is_socket_connected());
+
+	let handle = client
+		.add_event_listener(
+			Box::new(move |event| {
+				events_sender
+					.send(event.as_borrowed_cow().into_owned_cow())
+					.unwrap();
+			}),
+			None,
+		)
+		.await
+		.unwrap();
+	assert!(client.is_socket_connected());
+
+	await_event(
+		&mut events_receiver,
+		|event| matches!(event, SocketEvent::AuthSuccess),
+		Duration::from_secs(20),
+	)
+	.await
+	.unwrap();
+
+	std::mem::drop(handle);
+	assert!(!client.is_socket_connected());
+	await_event(
+		&mut events_receiver,
+		|event| matches!(event, SocketEvent::Unsubscribed),
+		Duration::from_secs(20),
+	)
+	.await
+	.unwrap();
+
+	// making sure it can reconnect
+
+	let (events_sender, mut events_receiver) = tokio::sync::mpsc::unbounded_channel();
+
+	assert!(!client.is_socket_connected());
+
+	let handle = client
+		.add_event_listener(
+			Box::new(move |event| {
+				events_sender
+					.send(event.as_borrowed_cow().into_owned_cow())
+					.unwrap();
+			}),
+			None,
+		)
+		.await
+		.unwrap();
+	assert!(client.is_socket_connected());
+
+	await_event(
+		&mut events_receiver,
+		|event| matches!(event, SocketEvent::AuthSuccess),
+		Duration::from_secs(20),
+	)
+	.await
+	.unwrap();
+
+	std::mem::drop(handle);
+	assert!(!client.is_socket_connected());
+	await_event(
+		&mut events_receiver,
+		|event| matches!(event, SocketEvent::Unsubscribed),
+		Duration::from_secs(20),
+	)
+	.await
+	.unwrap();
+}

filen-sdk-rs/tests/socket_tests.rs

@@ -74,7 +74,7 @@ async fn test_websocket_auth() {
 	await_event(
 		&mut events_receiver,
 		|event| *event == SocketEvent::AuthSuccess,
-		Duration::from_secs(10),
+		Duration::from_secs(20),
 	)
 	.await;
 }
@@ -111,7 +111,7 @@ async fn test_websocket_event_filtering() {
 	await_event(
 		&mut events_receiver,
 		|event| *event == SocketEvent::AuthSuccess,
-		Duration::from_secs(10),
+		Duration::from_secs(20),
 	)
 	.await;
 
@@ -159,7 +159,7 @@ async fn test_websocket_file_folder_creation() {
 			}
 			_ => false,
 		},
-		Duration::from_secs(10),
+		Duration::from_secs(20),
 	)
 	.await;
 
@@ -169,7 +169,7 @@ async fn test_websocket_file_folder_creation() {
 			SocketEvent::FileNew(data) => data.bucket == file_1.bucket && data.uuid == file_1.uuid,
 			_ => false,
 		},
-		Duration::from_secs(10),
+		Duration::from_secs(20),
 	)
 	.await;
 }

filen-types/src/api/v3/socket.rs

@@ -105,6 +105,8 @@ pub enum SocketEvent<'a> {
 	AuthFailed,
 	/// Sent when the socket has unexpectedly closed and begins attempting to reconnect
 	Reconnecting,
+	/// Sent when the handle to the event listener has been dropped and the listener is removed
+	Unsubscribed,
 	#[serde(borrow)]
 	NewEvent(NewEvent<'a>),
 	#[serde(borrow)]
@@ -318,6 +320,7 @@ impl SocketEvent<'_> {
 			SocketEvent::AuthSuccess => "authSuccess",
 			SocketEvent::AuthFailed => "authFailed",
 			SocketEvent::Reconnecting => "reconnecting",
+			SocketEvent::Unsubscribed => "unsubscribed",
 			SocketEvent::NewEvent(_) => "newEvent",
 			SocketEvent::FileRename(_) => "fileRename",
 			SocketEvent::FileArchiveRestored(_) => "fileArchiveRestored",