wip

Author: fafhrd91

src/connection.rs

@@ -245,7 +245,7 @@ impl ConnectionInner {
     pub(crate) fn register_remote_session(
         &mut self,
         remote_channel_id: u16,
-        begin: &Begin,
+        begin: Begin,
         cell: &Cell<ConnectionInner>,
     ) -> Result<(), AmqpProtocolError> {
         log::trace!(
@@ -256,6 +256,7 @@ impl ConnectionInner {
 
         let entry = self.sessions.vacant_entry();
         let local_token = entry.key();
+        let outgoing_window = begin.incoming_window();
 
         let session = Cell::new(SessionInner::new(
             local_token,
@@ -268,10 +269,10 @@ impl ConnectionInner {
         self.sessions_map.insert(remote_channel_id, local_token);
 
         let begin = Begin(Box::new(codec::BeginInner {
+            outgoing_window,
             remote_channel: Some(remote_channel_id),
             next_outgoing_id: 1,
             incoming_window: u32::MAX,
-            outgoing_window: begin.incoming_window(),
             handle_max: u32::MAX,
             offered_capabilities: None,
             desired_capabilities: None,
@@ -291,7 +292,7 @@ impl ConnectionInner {
         &mut self,
         local_channel_id: u16,
         remote_channel_id: u16,
-        begin: &Begin,
+        begin: Begin,
     ) {
         log::trace!(
             "{}: Begin response received: local {:?} remote {:?}",
@@ -377,9 +378,9 @@ impl ConnectionInner {
                 // the remote-channel property in the frame is the local channel id
                 // we previously sent to the remote
                 if let Some(local_channel_id) = begin.remote_channel() {
-                    self.complete_session_creation(local_channel_id, channel_id, &begin);
+                    self.complete_session_creation(local_channel_id, channel_id, begin);
                 } else {
-                    self.register_remote_session(channel_id, &begin, inner)?;
+                    self.register_remote_session(channel_id, begin, inner)?;
                 }
                 return Ok(Action::None);
             }

src/session.rs

@@ -6,9 +6,9 @@ use ntex_util::{HashMap, future::Either, future::Ready};
 use slab::Slab;
 
 use ntex_amqp_codec::protocol::{
-    self as codec, Accepted, Attach, DeliveryNumber, DeliveryState, Detach, Disposition, End,
-    Error, Flow, Frame, Handle, MessageFormat, ReceiverSettleMode, Role, SenderSettleMode, Source,
-    Transfer, TransferBody, TransferNumber,
+    self as codec, Accepted, Attach, Begin, DeliveryNumber, DeliveryState, Detach, Disposition,
+    End, Error, Flow, Frame, Handle, MessageFormat, ReceiverSettleMode, Role, SenderSettleMode,
+    Source, Transfer, TransferBody, TransferNumber,
 };
 use ntex_amqp_codec::{AmqpFrame, Encode};
 
@@ -33,6 +33,7 @@ pub(crate) struct SessionInner {
     sink: ConnectionRef,
     next_outgoing_id: TransferNumber,
     flags: Flags,
+    begin: Begin,
 
     remote_channel_id: u16,
     next_incoming_id: TransferNumber,
@@ -291,9 +292,9 @@ impl SessionInner {
         begin: Begin,
     ) -> SessionInner {
         SessionInner {
-            next_incoming_id: begin.next_incoming_id(),
-            remote_incoming_window: begin.remote_incoming_window(),
-            remote_outgoing_window: begin.remote_outgoing_window(),
+            next_incoming_id: begin.next_outgoing_id(),
+            remote_incoming_window: begin.incoming_window(),
+            remote_outgoing_window: begin.outgoing_window(),
             flags: if local { Flags::LOCAL } else { Flags::empty() },
             next_outgoing_id: INITIAL_NEXT_OUTGOING_ID,
             unsettled_snd_deliveries: HashMap::default(),