Fix control queue handling (#58)

Author: fafhrd91

CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [3.2.0] - 2024-12-03
+
+* Fix control queue handling
+
 ## [3.1.0] - 2024-12-01
 
 * Set "next_incoming_id" for Flow frame

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-amqp"
-version = "3.1.0"
+version = "3.2.0"
 authors = ["ntex contributors <[email protected]>"]
 description = "AMQP 1.0 Client/Server framework"
 documentation = "https://docs.rs/ntex-amqp"
@@ -25,7 +25,8 @@ default = []
 frame-trace = []
 
 [dependencies]
-ntex = "2"
+ntex = "2.9"
+ntex-util = "2.7"
 ntex-amqp-codec = "0.9"
 
 bitflags = "2"

src/connection.rs

@@ -6,7 +6,7 @@ use ntex::util::{HashMap, PoolRef, Ready};
 
 use crate::codec::protocol::{self as codec, Begin, Close, End, Error, Frame, Role};
 use crate::codec::{AmqpCodec, AmqpFrame};
-use crate::dispatcher::ControlQueue;
+use crate::control::ControlQueue;
 use crate::session::{Session, SessionInner, INITIAL_NEXT_OUTGOING_ID};
 use crate::sndlink::{SenderLink, SenderLinkInner};
 use crate::{cell::Cell, error::AmqpProtocolError, types::Action, Configuration};

src/control.rs

@@ -1,6 +1,6 @@
-use std::{fmt, io};
+use std::{cell::RefCell, collections::VecDeque, fmt, io};
 
-use ntex::util::Either;
+use ntex::{task::LocalWaker, util::Either};
 use ntex_amqp_codec::protocol;
 
 use crate::cell::Cell;
@@ -72,3 +72,16 @@ impl ControlFrame {
         self.0.get_ref().session.clone().map(Session::new)
     }
 }
+
+#[derive(Default, Debug)]
+pub(crate) struct ControlQueue {
+    pub(crate) pending: RefCell<VecDeque<ControlFrame>>,
+    pub(crate) waker: LocalWaker,
+}
+
+impl ControlQueue {
+    pub(crate) fn enqueue_frame(&self, frame: ControlFrame) {
+        self.pending.borrow_mut().push_back(frame);
+        self.waker.wake();
+    }
+}