Allow to set Transfer format

CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [3.4.0] - 2025-01-08
+
+* Allow to set Transfer format
+
 ## [3.3.1] - 2025-01-02
 
 * Fix rcv-settle-mode for sender links

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-amqp"
-version = "3.3.1"
+version = "3.4.0"
 authors = ["ntex contributors <[email protected]>"]
 description = "AMQP 1.0 Client/Server framework"
 documentation = "https://docs.rs/ntex-amqp"
@@ -26,7 +26,6 @@ frame-trace = []
 
 [dependencies]
 ntex = "2.10"
-ntex-io = "2.9.2"
 ntex-amqp-codec = "0.9"
 
 bitflags = "2"

src/delivery.rs

@@ -3,7 +3,7 @@ use std::cell::Cell as StdCell;
 use ntex::{channel::pool, util::Bytes};
 use ntex_amqp_codec::protocol::{
     DeliveryNumber, DeliveryState, Disposition, DispositionInner, Error, ErrorCondition, Handle,
-    Rejected, Role, TransferBody,
+    MessageFormat, Rejected, Role, TransferBody,
 };
 use ntex_amqp_codec::types::{Str, Symbol};
 
@@ -292,6 +292,7 @@ pub struct DeliveryBuilder {
     tag: Option<Bytes>,
     settled: bool,
     data: TransferBody,
+    format: Option<MessageFormat>,
     sender: Cell<SenderLinkInner>,
 }
 
@@ -300,6 +301,7 @@ impl DeliveryBuilder {
         Self {
             tag: None,
             settled: false,
+            format: None,
             data,
             sender,
         }
@@ -315,6 +317,11 @@ impl DeliveryBuilder {
         self
     }
 
+    pub fn format(mut self, fmt: MessageFormat) -> Self {
+        self.format = Some(fmt);
+        self
+    }
+
     pub async fn send(self) -> Result<Delivery, AmqpProtocolError> {
         let inner = self.sender.get_ref();
 
@@ -332,7 +339,7 @@ impl DeliveryBuilder {
             let (id, tag) = self
                 .sender
                 .get_mut()
-                .send(self.data, self.tag, self.settled)
+                .send(self.data, self.tag, self.settled, self.format)
                 .await?;
 
             Ok(Delivery {

src/session.rs

@@ -6,8 +6,8 @@ use slab::Slab;
 
 use ntex_amqp_codec::protocol::{
     self as codec, Accepted, Attach, DeliveryNumber, DeliveryState, Detach, Disposition, End,
-    Error, Flow, Frame, Handle, ReceiverSettleMode, Role, SenderSettleMode, Source, Transfer,
-    TransferBody, TransferNumber,
+    Error, Flow, Frame, Handle, MessageFormat, ReceiverSettleMode, Role, SenderSettleMode, Source,
+    Transfer, TransferBody, TransferNumber,
 };
 use ntex_amqp_codec::{AmqpFrame, Encode};
 
@@ -1217,6 +1217,7 @@ impl SessionInner {
         tag: Bytes,
         body: TransferBody,
         settled: bool,
+        format: Option<MessageFormat>,
     ) -> Result<DeliveryNumber, AmqpProtocolError> {
         loop {
             if self.remote_incoming_window == 0 {
@@ -1247,7 +1248,11 @@ impl SessionInner {
         } else {
             None
         };
-        let message_format = body.message_format();
+        let message_format = if format.is_none() {
+            body.message_format()
+        } else {
+            format
+        };
 
         let max_frame_size = self.max_frame_size();
         let max_frame_size = if max_frame_size > 2048 {

src/sndlink.rs

@@ -3,8 +3,8 @@ use std::{collections::VecDeque, future::Future};
 use ntex::channel::{condition, oneshot, pool};
 use ntex::util::{BufMut, ByteString, Bytes, Either, PoolRef, Ready};
 use ntex_amqp_codec::protocol::{
-    self as codec, Attach, DeliveryNumber, Error, Flow, ReceiverSettleMode, Role, SenderSettleMode,
-    SequenceNo, Target, TerminusDurability, TerminusExpiryPolicy, TransferBody,
+    self as codec, Attach, DeliveryNumber, Error, Flow, MessageFormat, ReceiverSettleMode, Role,
+    SenderSettleMode, SequenceNo, Target, TerminusDurability, TerminusExpiryPolicy, TransferBody,
 };
 
 use crate::delivery::DeliveryBuilder;
@@ -321,6 +321,7 @@ impl SenderLinkInner {
         body: T,
         tag: Option<Bytes>,
         settled: bool,
+        format: Option<MessageFormat>,
     ) -> Result<(DeliveryNumber, Bytes), AmqpProtocolError> {
         if let Some(ref err) = self.error {
             Err(err.clone())
@@ -355,7 +356,7 @@ impl SenderLinkInner {
                 .session
                 .inner
                 .get_mut()
-                .send_transfer(self.id as u32, tag.clone(), body, settled)
+                .send_transfer(self.id as u32, tag.clone(), body, settled, format)
                 .await?;
 
             Ok((id, tag))