Fix large transfers

Cargo.toml

@@ -37,7 +37,9 @@ uuid = { version = "1", features = ["v4"] }
 
 [dev-dependencies]
 env_logger = "0.11"
+rand = "0.8"
 ntex = { version = "1", features = ["tokio"] }
+ntex-amqp = { path = ".", features = ["frame-trace"] }
 
 [patch.crates-io]
 ntex-amqp = { path = "." }

src/delivery.rs

@@ -297,12 +297,6 @@ impl DeliveryBuilder {
 
         if let Some(ref err) = inner.error {
             Err(err.clone())
-        } else if inner
-            .max_message_size
-            .map(|l| self.data.len() > l as usize)
-            .unwrap_or_default()
-        {
-            Err(AmqpProtocolError::BodyTooLarge)
         } else {
             let id = self
                 .sender

src/session.rs

@@ -1278,24 +1278,20 @@ impl SessionInner {
             );
 
             loop {
-                let chunk = body.split_to(std::cmp::min(max_frame_size, body.len()));
-
                 // last chunk
                 if body.is_empty() {
-                    log::trace!("{}: Sending last tranfer for {:?}", self.tag(), tag);
-
-                    let mut transfer = Transfer(Default::default());
-                    transfer.0.more = false;
-                    self.post_frame(Frame::Transfer(transfer));
+                    log::trace!("{}: Last tranfer for {:?} is sent", self.tag(), tag);
                     break;
                 }
 
-                log::trace!("{}: Sending chunk tranfer for {:?}", self.tag(), tag);
+                let chunk = body.split_to(std::cmp::min(max_frame_size, body.len()));
 
+                log::trace!("{}: Sending chunk tranfer for {:?}", self.tag(), tag);
                 let mut transfer = Transfer(Default::default());
+                transfer.0.delivery_id = Some(delivery_id);
                 transfer.0.handle = link_handle;
                 transfer.0.body = Some(TransferBody::Data(chunk));
-                transfer.0.more = true;
+                transfer.0.more = !body.is_empty();
                 transfer.0.batchable = true;
                 self.post_frame(Frame::Transfer(transfer));
             }

tests/test_server.rs

@@ -8,6 +8,7 @@ use ntex::{http::Uri, rt, time::sleep, time::Millis};
 use ntex_amqp::{
     client, codec::protocol, error::LinkError, server, types, ControlFrame, ControlFrameKind,
 };
+use rand::{distributions::Alphanumeric, thread_rng, Rng};
 
 async fn server(
     _link: types::Link<()>,
@@ -91,6 +92,71 @@ async fn test_simple() -> std::io::Result<()> {
     Ok(())
 }
 
+#[ntex::test]
+async fn test_large_transfer() -> std::io::Result<()> {
+    let _ = env_logger::init();
+
+    let mut rng = thread_rng();
+    let data: String = (0..2048)
+        .map(|_| rng.sample(Alphanumeric) as char)
+        .collect();
+
+    let count = Arc::new(AtomicUsize::new(0));
+    let count2 = count.clone();
+    let srv = test_server(move || {
+        let count = count2.clone();
+        server::Server::build(|con: server::Handshake| async move {
+            match con {
+                server::Handshake::Amqp(con) => {
+                    let con = con.open().await.unwrap();
+                    Ok(con.ack(()))
+                }
+                server::Handshake::Sasl(_) => Err(()),
+            }
+        })
+        .control(|msg: ControlFrame| async move {
+            if let ControlFrameKind::AttachReceiver(_, rcv) = msg.kind() {
+                rcv.set_max_message_size(1024);
+            }
+            Ok::<_, ()>(())
+        })
+        .finish(
+            server::Router::<()>::new()
+                .service(
+                    "test",
+                    fn_factory_with_config(move |_: types::Link<()>| server_count(count.clone())),
+                )
+                .finish(),
+        )
+    });
+
+    let uri = Uri::try_from(format!("amqp://{}:{}", srv.addr().ip(), srv.addr().port())).unwrap();
+    let client = client::Connector::new().connect(uri).await.unwrap();
+    let sink = client.sink();
+    ntex::rt::spawn(async move {
+        let _ = client.start_default().await;
+    });
+
+    let session = sink.open_session().await.unwrap();
+    let link = session
+        .build_sender_link("test", "test")
+        .attach()
+        .await
+        .unwrap();
+
+    let delivery = link
+        .delivery(Bytes::from(data.clone()))
+        .send()
+        .await
+        .unwrap();
+    let st = delivery.wait().await.unwrap().unwrap();
+    assert_eq!(st, protocol::DeliveryState::Accepted(protocol::Accepted {}));
+    sleep(Millis(250)).await;
+
+    assert_eq!(count.load(Ordering::Relaxed), 1);
+    Ok(())
+}
+
 async fn sasl_auth(auth: server::Sasl) -> Result<server::HandshakeAck<()>, server::HandshakeError> {
     let init = auth
         .mechanism("PLAIN")