SocketType cleanups

Cargo.toml

@@ -25,8 +25,6 @@ rand = "0.8"
 bytes = "1"
 tokio = { version = "1", features = ["full"], optional = true }
 tokio-util = { version = "0.7", features = ["compat"], optional = true }
-num-traits = "0.2"
-enum-primitive-derive = "0.2"
 dashmap = "5"
 crossbeam-queue = "0.3"
 uuid = { version = "1", features = ["v4"] }

src/lib.rs

@@ -41,22 +41,36 @@ use crate::codec::*;
 use crate::transport::AcceptStopHandle;
 use util::PeerIdentity;
 
-#[macro_use]
-extern crate enum_primitive_derive;
-
 use async_trait::async_trait;
 use asynchronous_codec::FramedWrite;
 use futures::channel::mpsc;
 use futures::FutureExt;
-use num_traits::ToPrimitive;
 use parking_lot::Mutex;
+
 use std::collections::HashMap;
 use std::convert::TryFrom;
 use std::fmt::{Debug, Display};
+use std::str::FromStr;
 use std::sync::Arc;
 
+const COMPATIBILITY_MATRIX: [u8; 121] = [
+    // PAIR, PUB, SUB, REQ, REP, DEALER, ROUTER, PULL, PUSH, XPUB, XSUB
+    1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, // PAIR
+    0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 1, // PUB
+    0, 1, 0, 0, 0, 0, 0, 0, 0, 1, 0, // SUB
+    0, 0, 0, 0, 1, 0, 1, 0, 0, 0, 0, // REQ
+    0, 0, 0, 1, 0, 1, 0, 0, 0, 0, 0, // REP
+    0, 0, 0, 0, 1, 1, 1, 0, 0, 0, 0, // DEALER
+    0, 0, 0, 1, 0, 1, 1, 0, 0, 0, 0, // ROUTER
+    0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, // PULL
+    0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, // PUSH
+    0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 1, // XPUB
+    0, 1, 0, 0, 0, 0, 0, 0, 0, 1, 0, // XSUB
+];
+
 #[allow(clippy::upper_case_acronyms)]
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Primitive)]
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
+#[repr(usize)]
 pub enum SocketType {
     PAIR = 0,
     PUB = 1,
@@ -89,25 +103,48 @@ impl SocketType {
             SocketType::STREAM => "STREAM",
         }
     }
+
+    /// Checks if two sockets are compatible with each other
+    /// ```
+    /// use zeromq::SocketType;
+    /// assert!(SocketType::PUB.compatible(SocketType::SUB));
+    /// assert!(SocketType::REQ.compatible(SocketType::REP));
+    /// assert!(SocketType::DEALER.compatible(SocketType::ROUTER));
+    /// assert!(!SocketType::PUB.compatible(SocketType::REP));
+    /// ```
+    pub fn compatible(&self, other: SocketType) -> bool {
+        let row_index = *self as usize;
+        let col_index = other as usize;
+        COMPATIBILITY_MATRIX[row_index * 11 + col_index] != 0
+    }
+}
+
+impl FromStr for SocketType {
+    type Err = ZmqError;
+
+    #[inline]
+    fn from_str(s: &str) -> Result<Self, ZmqError> {
+        Self::try_from(s.as_bytes())
+    }
 }
 
-impl TryFrom<&str> for SocketType {
+impl TryFrom<&[u8]> for SocketType {
     type Error = ZmqError;
 
-    fn try_from(s: &str) -> Result<Self, ZmqError> {
+    fn try_from(s: &[u8]) -> Result<Self, ZmqError> {
         Ok(match s {
-            "PAIR" => SocketType::PAIR,
-            "PUB" => SocketType::PUB,
-            "SUB" => SocketType::SUB,
-            "REQ" => SocketType::REQ,
-            "REP" => SocketType::REP,
-            "DEALER" => SocketType::DEALER,
-            "ROUTER" => SocketType::ROUTER,
-            "PULL" => SocketType::PULL,
-            "PUSH" => SocketType::PUSH,
-            "XPUB" => SocketType::XPUB,
-            "XSUB" => SocketType::XSUB,
-            "STREAM" => SocketType::STREAM,
+            b"PAIR" => SocketType::PAIR,
+            b"PUB" => SocketType::PUB,
+            b"SUB" => SocketType::SUB,
+            b"REQ" => SocketType::REQ,
+            b"REP" => SocketType::REP,
+            b"DEALER" => SocketType::DEALER,
+            b"ROUTER" => SocketType::ROUTER,
+            b"PULL" => SocketType::PULL,
+            b"PUSH" => SocketType::PUSH,
+            b"XPUB" => SocketType::XPUB,
+            b"XSUB" => SocketType::XSUB,
+            b"STREAM" => SocketType::STREAM,
             _ => return Err(ZmqError::Other("Unknown socket type")),
         })
     }

src/util.rs

@@ -5,7 +5,6 @@ use asynchronous_codec::FramedRead;
 use bytes::Bytes;
 use futures::stream::StreamExt;
 use futures::SinkExt;
-use num_traits::Pow;
 use rand::Rng;
 
 use std::convert::{TryFrom, TryInto};
@@ -102,36 +101,6 @@ pub(crate) struct Peer {
     pub(crate) recv_queue: FramedRead<Box<dyn FrameableRead>, ZmqCodec>,
 }
 
-const COMPATIBILITY_MATRIX: [u8; 121] = [
-    // PAIR, PUB, SUB, REQ, REP, DEALER, ROUTER, PULL, PUSH, XPUB, XSUB
-    1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, // PAIR
-    0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 1, // PUB
-    0, 1, 0, 0, 0, 0, 0, 0, 0, 1, 0, // SUB
-    0, 0, 0, 0, 1, 0, 1, 0, 0, 0, 0, // REQ
-    0, 0, 0, 1, 0, 1, 0, 0, 0, 0, 0, // REP
-    0, 0, 0, 0, 1, 1, 1, 0, 0, 0, 0, // DEALER
-    0, 0, 0, 1, 0, 1, 1, 0, 0, 0, 0, // ROUTER
-    0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, // PULL
-    0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, // PUSH
-    0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 1, // XPUB
-    0, 1, 0, 0, 0, 0, 0, 0, 0, 1, 0, // XSUB
-];
-
-/// Checks if two sokets are compatible with each other
-/// ```
-/// use zeromq::util::sockets_compatible;
-/// use zeromq::SocketType;
-/// assert!(sockets_compatible(SocketType::PUB, SocketType::SUB));
-/// assert!(sockets_compatible(SocketType::REQ, SocketType::REP));
-/// assert!(sockets_compatible(SocketType::DEALER, SocketType::ROUTER));
-/// assert!(!sockets_compatible(SocketType::PUB, SocketType::REP));
-/// ```
-pub fn sockets_compatible(one: SocketType, another: SocketType) -> bool {
-    let row_index = one.to_usize().unwrap();
-    let col_index = another.to_usize().unwrap();
-    COMPATIBILITY_MATRIX[row_index * 11 + col_index] != 0
-}
-
 /// Given the result of the greetings exchange, determines the version of the
 /// ZMTP protocol that should be used for communication with the peer according
 /// to https://rfc.zeromq.org/spec/23/#version-negotiation.
@@ -190,13 +159,10 @@ pub(crate) async fn ready_exchange(
     match ready_repl {
         Some(Ok(Message::Command(command))) => match command.name {
             ZmqCommandName::READY => {
-                let other_sock_type = command
-                    .properties
-                    .get("Socket-Type")
-                    .map(|x| {
-                        SocketType::try_from(std::str::from_utf8(x).expect("Invalid socket type"))
-                    })
-                    .unwrap_or(Err(ZmqError::Other("Failed to parse other socket type")))?;
+                let other_sock_type = match command.properties.get("Socket-Type") {
+                    Some(s) => SocketType::try_from(&s[..])?,
+                    None => Err(ZmqError::Other("Failed to parse other socket type"))?,
+                };
 
                 let peer_id = command
                     .properties
@@ -205,7 +171,7 @@ pub(crate) async fn ready_exchange(
                     .transpose()?
                     .unwrap_or_default();
 
-                if sockets_compatible(socket_type, other_sock_type) {
+                if socket_type.compatible(other_sock_type) {
                     Ok(peer_id)
                 } else {
                     Err(ZmqError::Other(
@@ -247,7 +213,7 @@ pub(crate) async fn connect_forever(endpoint: Endpoint) -> ZmqResult<(FramedIo,
                 }
                 let delay = {
                     let mut rng = rand::thread_rng();
-                    std::f64::consts::E.pow(try_num as f64 / 3.0) + rng.gen_range(0.0f64..0.1f64)
+                    std::f64::consts::E.powf(try_num as f64 / 3.0) + rng.gen_range(0.0f64..0.1f64)
                 };
                 async_rt::task::sleep(std::time::Duration::from_secs_f64(delay)).await;
                 continue;