Refactor dispatch process (#12)

* refactor server impl

* refactor link handling

CHANGES.md

@@ -1,5 +1,13 @@
 # Changes
 
+## [0.5.0-b.5] - 2021-08-11
+
+* Refactor server dispatch process
+
+## [codec-0.6.2] - 2021-08-11
+
+* Add helper methods to Transfer type
+
 ## [0.5.0-b.3] - 2021-08-10
 
 * Add Session::connection() method, returns ref to Connection

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-amqp"
-version = "0.5.0-b.4"
+version = "0.5.0-b.5"
 authors = ["ntex contributors <[email protected]>"]
 description = "AMQP 1.0 Client/Server framework"
 documentation = "https://docs.rs/ntex-amqp"
@@ -25,7 +25,7 @@ frame-trace = []
 
 [dependencies]
 ntex = "0.4.0-b.1"
-ntex-amqp-codec = "0.6.1"
+ntex-amqp-codec = "0.6.2"
 
 bitflags = "1.2"
 derive_more = "0.99"

codec/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-amqp-codec"
-version = "0.6.1"
+version = "0.6.2"
 description = "AMQP 1.0 Protocol Codec"
 authors = ["Nikolay Kim <[email protected]>", "Max Gortman <[email protected]>", "Mike Yagley <[email protected]>"]
 license = "MIT/Apache-2.0"

codec/src/protocol/mod.rs

@@ -5,7 +5,7 @@ use derive_more::From;
 use ntex_bytes::{BufMut, ByteString, Bytes, BytesMut};
 use uuid::Uuid;
 
-use super::codec::{self, DecodeFormatted, Encode};
+use super::codec::{self, Decode, DecodeFormatted, Encode};
 use super::error::AmqpParseError;
 use super::message::Message;
 use super::types::*;
@@ -253,22 +253,44 @@ impl TransferBody {
 }
 
 impl From<Message> for TransferBody {
+    #[inline]
     fn from(msg: Message) -> Self {
         Self::Message(Box::new(msg))
     }
 }
 
 impl Encode for TransferBody {
+    #[inline]
     fn encoded_size(&self) -> usize {
         match self {
             TransferBody::Data(ref data) => data.len(),
             TransferBody::Message(ref data) => data.encoded_size(),
         }
     }
+    #[inline]
     fn encode(&self, dst: &mut BytesMut) {
         match *self {
             TransferBody::Data(ref data) => dst.put_slice(data),
             TransferBody::Message(ref data) => data.encode(dst),
         }
     }
 }
+
+impl Transfer {
+    #[inline]
+    pub fn get_body(&self) -> Option<&Bytes> {
+        match self.body {
+            Some(TransferBody::Data(ref b)) => Some(b),
+            _ => None,
+        }
+    }
+
+    #[inline]
+    pub fn load_message<T: Decode>(&self) -> Result<T, AmqpParseError> {
+        if let Some(TransferBody::Data(ref b)) = self.body {
+            Ok(T::decode(b)?.1)
+        } else {
+            Err(AmqpParseError::UnexpectedType("body"))
+        }
+    }
+}

examples/server.rs

@@ -7,7 +7,7 @@ async fn server(
 ) -> Result<
     Box<
         dyn Service<
-                Request = server::Transfer<()>,
+                Request = server::Transfer,
                 Response = server::Outcome,
                 Error = AmqpError,
                 Future = Ready<server::Outcome, AmqpError>,

src/client/connection.rs

@@ -16,7 +16,7 @@ pub struct Client<Io, St = ()> {
     keepalive: u16,
     remote_config: Configuration,
     timer: Timer,
-    st: State<St>,
+    _st: State<St>,
 }
 
 impl<T> Client<T, ()>
@@ -41,7 +41,7 @@ where
             keepalive,
             remote_config,
             timer,
-            st: State::new(()),
+            _st: State::new(()),
         }
     }
 }
@@ -68,7 +68,7 @@ where
             keepalive: self.keepalive,
             remote_config: self.remote_config,
             timer: self.timer,
-            st: State::new(st),
+            _st: State::new(st),
         }
     }
 
@@ -77,9 +77,8 @@ where
     /// Default handler closes connection on any control message.
     pub async fn start_default(self) -> Result<(), DispatcherError> {
         let dispatcher = Dispatcher::new(
-            self.st,
             self.connection,
-            fn_service(|_| Ready::<_, LinkError>::Err(LinkError::force_detach())),
+            fn_service(|_| Ready::<_, LinkError>::Ok(())),
             fn_service(|_| Ready::<_, LinkError>::Ok(())),
             self.remote_config.timeout_remote_secs(),
         )