feat: Allow overriding block_on_no_clients queue param

In the current implementation, when a SUB provider is restarted, the library just hangs without
detecting connection loss. This happens because on disconnect a client is kicked out of the
queue, as since for all the socket queues block_on_no_clients is set to true, it hangs indefinitely.

I'm unsure what would be the effect of setting block_on_no_clients to false globally (though it
seems reasonable), so in this PR add an ability to override this value is introduced via
SocketOptions.

src/dealer.rs

@@ -31,7 +31,7 @@ impl Drop for DealerSocket {
 #[async_trait]
 impl Socket for DealerSocket {
     fn with_options(options: SocketOptions) -> Self {
-        let fair_queue = FairQueue::new(true);
+        let fair_queue = FairQueue::new(options.block_on_no_clients);
         Self {
             backend: Arc::new(GenericSocketBackend::with_options(
                 Some(fair_queue.inner()),

src/lib.rs

@@ -169,16 +169,30 @@ pub enum SocketEvent {
     Disconnected(PeerIdentity),
 }
 
-#[derive(Default)]
+#[derive(Debug, Clone)]
 pub struct SocketOptions {
     pub(crate) peer_id: Option<PeerIdentity>,
+    pub(crate) block_on_no_clients: bool,
+}
+
+impl Default for SocketOptions {
+    fn default() -> Self {
+        Self {
+            peer_id: Default::default(),
+            block_on_no_clients: true,
+        }
+    }
 }
 
 impl SocketOptions {
     pub fn peer_identity(&mut self, peer_id: PeerIdentity) -> &mut Self {
         self.peer_id = Some(peer_id);
         self
     }
+    pub fn block_on_no_clients(&mut self, block_on_no_clients: bool) -> &mut Self {
+        self.block_on_no_clients = block_on_no_clients;
+        self
+    }
 }
 
 #[async_trait]

src/pull.rs

@@ -25,7 +25,7 @@ pub struct PullSocket {
 #[async_trait]
 impl Socket for PullSocket {
     fn with_options(options: SocketOptions) -> Self {
-        let fair_queue = FairQueue::new(true);
+        let fair_queue = FairQueue::new(options.block_on_no_clients);
         Self {
             backend: Arc::new(GenericSocketBackend::with_options(
                 Some(fair_queue.inner()),

src/rep.rs

@@ -43,7 +43,7 @@ impl Drop for RepSocket {
 #[async_trait]
 impl Socket for RepSocket {
     fn with_options(options: SocketOptions) -> Self {
-        let fair_queue = FairQueue::new(true);
+        let fair_queue = FairQueue::new(options.block_on_no_clients);
         Self {
             backend: Arc::new(RepSocketBackend {
                 peers: DashMap::new(),

src/router.rs

@@ -32,7 +32,7 @@ impl Drop for RouterSocket {
 #[async_trait]
 impl Socket for RouterSocket {
     fn with_options(options: SocketOptions) -> Self {
-        let fair_queue = FairQueue::new(true);
+        let fair_queue = FairQueue::new(options.block_on_no_clients);
         Self {
             backend: Arc::new(GenericSocketBackend::with_options(
                 Some(fair_queue.inner()),

src/sub.rs

@@ -9,6 +9,7 @@ use crate::transport::AcceptStopHandle;
 use crate::util::PeerIdentity;
 use crate::{
     MultiPeerBackend, Socket, SocketBackend, SocketEvent, SocketOptions, SocketRecv, SocketType,
+    ZmqError,
 };
 
 use async_trait::async_trait;
@@ -156,7 +157,7 @@ impl SubSocket {
 #[async_trait]
 impl Socket for SubSocket {
     fn with_options(options: SocketOptions) -> Self {
-        let fair_queue = FairQueue::new(true);
+        let fair_queue = FairQueue::new(options.block_on_no_clients);
         Self {
             backend: Arc::new(SubSocketBackend::with_options(
                 Some(fair_queue.inner()),