Repository: zslayton/stomp-rs (language: rust)

Issue:
Nack requeue false
Sending a Nack frame with Rabbit MQ automatically enables [`requeue`](https://www.rabbitmq.com/stomp.html). If this is not desired behaviour, like in my case, then there is [no way to change it to false as default](https://groups.google.com/forum/#!msg/rabbitmq-users/Y8k0W3Oflto/bq0nkxrKq5QJ) on the server-side. So it must be implemented client-side.

The [STOMP spec is ambiguous on what the default setting should be](https://stomp.github.io/stomp-specification-1.2.html#NACK) and leaves it up to the server. It does even not specify the name of header that should be send if you do want control over the behaviour.

So with respect to the library, I would suggest adding a `on_frame` method to the `session` struct. This enables anyone to hack into the frame for these kind of situations: when the spec leaves it open how to implement something.

From the perspective of the user it then can look like this.

``` rust
session.on_frame(|frame : &mut Frame| {
    match frame.command.as_ref() {
        "NACK" => {
            &frame.headers.push(Header::new("requeue", "false"));
        },
        _ => {}
    }
});
```



Discussion / hints:
Thanks for reporting this. I'd be open to adding callbacks for `on_before_send` and `on_before_receive` that each get the opportunity to manipulate `&mut Frame` before the main handlers are called. I wonder if the `AckOrNack` enum should have a third option to allow a custom frame to be specified.

Out of interest, in your situation could you simply respond with an `Ack` even though you encountered trouble processing the message? That way the server would drop the message rather than re-queuing it. Or are you relying on the `Nack` in other ways as well?

@zslayton Interesting thought: to send an `Ack` even though the message fails. Actually, that would work in my case. However, I do think it is not the best solution. Probably the `Ack` would end up in the Rabbit MQ logs which indicates that the message was processed successfully (which is not the case).

Personally, I would be in favour of `on_before_send` and `on_before_receive`. That solves any case where the spec is not clear/leaves room for own implementation. A third option to the `AckOrNack` enum, would only help you in that particular case. I do not know the spec extremely well (I just chose STOMP because this package handles communication with Rabbit MQ best in my opinion.), so I cannot assess whether are many cases like that.

I don't use RabbitMQ as often, but in ActiveMQ there aren't any records kept of the individual messages being handled (without enabling debug logging).

`on_before_send` and `on_before_receive` are certainly the most flexible approach. I'll plan on adding these.

@zslayton Wonderful, I will see if I can those in with a PR. My Rust skills are [improving](https://github.com/genkgo/gworkerd).

Great! This change is going to require a `FrameHandlerMut` variant of `FrameHandler`, which [currently lives in session.rs](https://github.com/zslayton/stomp-rs/blob/79bcc9e1333db9e7d69e8188abc48436bd3718e2/src/session.rs#L32-L50).

If present, the `on_before_receive` callback should be invoked [just before the `dispatch` method is](https://github.com/zslayton/stomp-rs/blob/79bcc9e1333db9e7d69e8188abc48436bd3718e2/src/session.rs#L126).

Similarly, `on_before_send` would be invoked in the `send` method [just before the write](https://github.com/zslayton/stomp-rs/blob/79bcc9e1333db9e7d69e8188abc48436bd3718e2/src/session.rs#L334).

@zslayton Yes, I noticed that should be a new variant of FrameHandler. My first try failed miserably however. Storing closures with lifetimes etc are concepts that I am not used to.

@zslayton But thanks for providing the right names and pointing into the right directions. That will help to create the correct solution.


Fix patch:
diff --git a/.gitignore b/.gitignore
index ef56e01..6d815bb 100644
--- a/.gitignore
+++ b/.gitignore
@@ -5,3 +5,4 @@ examples/**/target/**
 **callgrind.out.*
 **perf.data*
 Cargo.lock
+.idea
\ No newline at end of file
diff --git a/README.md b/README.md
index 79a691e..be1adbf 100644
--- a/README.md
+++ b/README.md
@@ -109,13 +109,29 @@ session.message(destination, "text/plain", "Hypoteneuse".as_bytes())
   .with(ReceiptHandler::new(|frame: &Frame| println!("Got a receipt for 'Hypoteneuse'.")))
   .send();
 ```
-### Handling ERROR frames
+### Handling frames
+To handle errors, you can register an error handler
 ```rust
 session.on_error(|frame: &Frame| {
   panic!("ERROR frame received:\n{}", frame);
 });
 ```
-
+To change the frame that will be sent, use the before send handler 
+```rust
+session.on_before_send(|frame: &mut Frame| {
+  match frame.command.as_ref() {
+    "NACK" => {
+      frame.headers.push(Header::new("requeue", "false"));
+    },
+    _ => {}
+  }
+});
+```
+And to change the frame that is received, use the before receive handler
+```rust
+session.on_before_receive(|frame: &mut Frame| {
+});
+```
 ### Cargo.toml
 ```toml
 [package]
diff --git a/src/session.rs b/src/session.rs
index 73f4822..8eb5b7e 100644
--- a/src/session.rs
+++ b/src/session.rs
@@ -33,22 +33,42 @@ pub trait FrameHandler {
   fn on_frame(&mut self, &Frame);
 }
 
+pub trait FrameHandlerMut {
+  fn on_frame(&mut self, &mut Frame);
+}
+
 impl <F> FrameHandler for F where F: FnMut(&Frame) {
   fn on_frame(&mut self, frame: &Frame) {
     self(frame)
   }
 }
 
+impl <F> FrameHandlerMut for F where F: FnMut(&mut Frame) {
+  fn on_frame(&mut self, frame: &mut Frame) {
+    self(frame)
+  }
+}
+
 pub trait ToFrameHandler <'a> {
   fn to_frame_handler(self) -> Box<FrameHandler + 'a>;
 }
 
+pub trait ToFrameHandlerMut <'a> {
+  fn to_frame_handler_mut(self) -> Box<FrameHandlerMut + 'a>;
+}
+
 impl <'a, T: 'a> ToFrameHandler <'a> for T where T: FrameHandler {
   fn to_frame_handler(self) -> Box<FrameHandler + 'a> {
     Box::new(self) as Box<FrameHandler>
   }
 }
 
+impl <'a, T: 'a> ToFrameHandlerMut <'a> for T where T: FrameHandlerMut {
+  fn to_frame_handler_mut(self) -> Box<FrameHandlerMut + 'a> {
+    Box::new(self) as Box<FrameHandlerMut>
+  }
+}
+
 pub struct ReceiptHandler<'a, T> where T: 'a + ToFrameHandler<'a> {
   pub handler: T,
   _marker: PhantomData<&'a T>
@@ -77,7 +97,9 @@ pub struct Session <'a> {
   tx_heartbeat_ms: u64,
   pub subscriptions: HashMap<String, Subscription <'a>>,
   pub receipt_handlers: HashMap<String, Box<FrameHandler + 'a>>,
-  error_callback: Box<FrameHandler + 'a>
+  error_callback: Box<FrameHandler + 'a>,
+  frame_send_callback: Box<FrameHandlerMut + 'a>,
+  frame_receive_callback: Box<FrameHandlerMut + 'a>
 }
 
 pub enum StompTimeout {
@@ -124,6 +146,7 @@ impl <'a> Handler for Session<'a> {
         Some(CompleteFrame(mut frame)) => {
           debug!("Received frame!:\n{}", frame);
           self.reset_rx_heartbeat_timeout(event_loop);
+          self.frame_receive_callback.on_frame(&mut frame);
           self.dispatch(&mut frame);
           self.frame_buffer.recycle_frame(frame);
           num_frames += 1;
@@ -159,7 +182,9 @@ impl <'a> Session <'a> {
       tx_heartbeat_ms: (tx_heartbeat_ms as f64 / 2f64) as u64, //FIXME: Make this configurable, change units
       subscriptions: HashMap::new(),
       receipt_handlers: HashMap::new(),
-      error_callback: Box::new(Session::default_error_callback) as Box<FrameHandler>
+      error_callback: Box::new(Session::default_error_callback) as Box<FrameHandler>,
+      frame_send_callback: Box::new(Session::default_frame_send_callback) as Box<FrameHandlerMut>,
+      frame_receive_callback: Box::new(Session::default_frame_receive_callback) as Box<FrameHandlerMut>
     }
   }
 
@@ -251,12 +276,28 @@ impl <'a> Session <'a> {
   fn default_error_callback(frame : &Frame) {
     error!("ERROR received:\n{}", frame);
   }
-  
+
+  fn default_frame_send_callback(frame : &mut Frame) {
+  }
+
+  fn default_frame_receive_callback(frame : &mut Frame) {
+  }
+
   pub fn on_error<T: 'a>(&mut self, handler_convertible: T) where T : ToFrameHandler<'a> + 'a {
     let handler = handler_convertible.to_frame_handler();
     self.error_callback = handler;
   }
 
+  pub fn on_before_send<T: 'a>(&mut self, handler_convertible: T) where T : ToFrameHandlerMut<'a> + 'a {
+    let handler = handler_convertible.to_frame_handler_mut();
+    self.frame_send_callback = handler;
+  }
+
+  pub fn on_before_receive<T: 'a>(&mut self, handler_convertible: T) where T : ToFrameHandlerMut<'a> + 'a {
+    let handler = handler_convertible.to_frame_handler_mut();
+    self.frame_receive_callback = handler;
+  }
+
   fn handle_receipt(&mut self, frame: &mut Frame) {
     match frame.headers.get_receipt_id() {
       Some(ReceiptId(ref receipt_id)) => {
@@ -331,7 +372,8 @@ impl <'a> Session <'a> {
     Ok(transaction)
   }
 
-  pub fn send(&mut self, frame: Frame) -> Result<()> {
+  pub fn send(&mut self, mut frame: Frame) -> Result<()> {
+    self.frame_send_callback.on_frame(&mut frame);
     match frame.write(&mut self.connection.tcp_stream) {
       Ok(_) => Ok(()),//FIXME: Replace 'Other' below with a more meaningful ErrorKind
       Err(_) => Err(Error::new(Other, "Could not send frame: the connection to the server was lost."))


Test patch: