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

Issue:
Change callback to take &Frame instead of Frame
This will allow a small number of allocations per frame dispatch to be avoided.

Use session.send() wherever possible
A number of old uses of `some_frame.write(&mut session.connection.writer)` are still lying around when `session.send(some_frame)` does the same thing and is easier to read.



Discussion / hints:

Fix patch:
diff --git a/Cargo.toml b/Cargo.toml
index e29f5fd..79d5eee 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,7 +1,7 @@
 [package]
 
 name = "stomp"
-version = "0.1.1"
+version = "0.2.0"
 authors = [ "zslayton" ]
 
 [lib]
diff --git a/README.md b/README.md
index 6c310fa..1d31757 100644
--- a/README.md
+++ b/README.md
@@ -26,7 +26,7 @@ fn main() {
     Err(error) => fail!("Could not connect to the server: {}", error)
   };
   
-  fn on_message(frame: Frame) -> AckOrNack {
+  fn on_message(frame: &Frame) -> AckOrNack {
     println!("Received a message:\n{}", frame);
     Ack
   }
@@ -67,7 +67,7 @@ fn main() {
 
 ### Handling RECEIPT frames
 ```rust
-  fn on_receipt(frame: Frame) {
+  fn on_receipt(frame: &Frame) {
     debug!("RECEIPT frame received:\n{}", frame);
   }
 
@@ -77,7 +77,7 @@ fn main() {
 
 ### Handling ERROR frames
 ```rust
-  fn on_error(frame: Frame) {
+  fn on_error(frame: &Frame) {
     fail!("ERROR frame received:\n{}", frame);
   }
 
diff --git a/src/frame.rs b/src/frame.rs
index 46942b8..6bfb3c4 100644
--- a/src/frame.rs
+++ b/src/frame.rs
@@ -43,18 +43,18 @@ impl Frame {
   pub fn to_str(&self) -> String {
     let space_required = self.character_count();
     let mut frame_string = String::with_capacity(space_required);
-    frame_string = frame_string.append(self.command.as_slice());
-    frame_string = frame_string.append("\n");
+    frame_string.push_str(self.command.as_slice());
+    frame_string.push_str("\n");
     for header in self.headers.iter() {
-      frame_string = frame_string.append(header.get_raw());
-      frame_string = frame_string.append("\n");
+      frame_string.push_str(header.get_raw());
+      frame_string.push_str("\n");
     }
-    frame_string = frame_string.append("\n");
+    frame_string.push_str("\n");
     let body_string : &str = match from_utf8(self.body.as_slice()) {
       Some(ref s) => *s,
       None => "<Binary content>" // Space is wasted in this case. Could shrink to fit?
     };
-    frame_string = frame_string.append(body_string);
+    frame_string.push_str(body_string);
     frame_string
   }
 
@@ -201,7 +201,7 @@ impl Frame {
     let send_frame = Frame {
       command : "SEND".to_string(),
       headers : header_list,
-      body : Vec::from_slice(body)
+      body : body.to_vec()
     };
     send_frame
   }
diff --git a/src/session.rs b/src/session.rs
index d7f80b6..9023f82 100644
--- a/src/session.rs
+++ b/src/session.rs
@@ -5,11 +5,13 @@ use connection::Connection;
 use subscription::AckMode;
 use subscription::Auto;
 use subscription::AckOrNack;
+use subscription::Ack;
+use subscription::Nack;
 use subscription::Client;
 use subscription::ClientIndividual;
+use subscription::Subscription;
+use header;
 use header::Header;
-use header::Subscription;
-use header::Ack;
 use header::ReceiptId;
 use header::StompHeaderSet;
 use transaction::Transaction;
@@ -24,11 +26,11 @@ pub struct Session {
   next_transaction_id: uint,
   next_subscription_id: uint,
   next_receipt_id: uint,
-  subscriptions: HashMap<String, ::subscription::Subscription>,
+  subscriptions: HashMap<String, Subscription>,
   outstanding_receipts: HashMap<String, ReceiptStatus>, 
   pub connection : Connection,
-  receipt_callback: fn(Frame) -> (), 
-  error_callback: fn(Frame) -> ()
+  receipt_callback: fn(&Frame) -> (), 
+  error_callback: fn(&Frame) -> ()
 }
 
 impl Session {
@@ -45,15 +47,15 @@ impl Session {
     }
   }
  
-  fn default_error_callback(frame : Frame) {
+  fn default_error_callback(frame : &Frame) {
     error!("ERROR received:\n{}", frame);
   }
   
-  fn default_receipt_callback(frame : Frame) {
+  fn default_receipt_callback(frame : &Frame) {
     info!("RECEIPT received:\n{}", frame);
   }
 
-  pub fn on_error(&mut self, callback: fn(Frame)) {
+  pub fn on_error(&mut self, callback: fn(&Frame)) {
     self.error_callback = callback;
   }
 
@@ -71,10 +73,10 @@ impl Session {
       },
       None => fail!("Received RECEIPT frame without a receipt-id")
     };
-    (self.receipt_callback)(frame);
+    (self.receipt_callback)(&frame);
   }
 
-  pub fn on_receipt(&mut self, callback: fn(Frame)) {
+  pub fn on_receipt(&mut self, callback: fn(&Frame)) {
      self.receipt_callback = callback;
   }
 
@@ -124,9 +126,9 @@ impl Session {
     Ok(())
   }
 
-  pub fn subscribe(&mut self, topic: &str, ack_mode: AckMode, callback: fn(Frame)->AckOrNack)-> IoResult<String> {
+  pub fn subscribe(&mut self, topic: &str, ack_mode: AckMode, callback: fn(&Frame)->AckOrNack)-> IoResult<String> {
     let next_id = self.generate_subscription_id();
-    let sub = ::subscription::Subscription::new(next_id, topic, ack_mode, callback);
+    let sub = Subscription::new(next_id, topic, ack_mode, callback);
     let subscribe_frame = Frame::subscribe(sub.id.as_slice(), sub.topic.as_slice(), ack_mode);
     debug!("Sending frame:\n{}", subscribe_frame);
     try!(subscribe_frame.write(&mut self.connection.writer));
@@ -165,69 +167,58 @@ impl Session {
   pub fn dispatch(&mut self, frame: Frame) -> () {
     // Check for ERROR frame
     match frame.command.as_slice() {
-       "ERROR" => return (self.error_callback)(frame),
+       "ERROR" => return (self.error_callback)(&frame),
        "RECEIPT" => return self.handle_receipt(frame),
         _ => {} // No operation
     };
+ 
+    let ack_mode : AckMode;
+    let callback : fn(&Frame) -> AckOrNack;
+    { // This extra scope is required to free up `frame` and `subscription`
+      // following a borrow.
+      let header::Subscription(sub_id) = 
+        frame.headers
+        .get_subscription()
+        .expect("Frame did not contain a subscription header.");
+
+      let (a, c) = 
+         self.subscriptions
+         .find_equiv(&sub_id)
+         .map(|sub| (sub.ack_mode, sub.callback))
+         .expect("Received a message for an unknown subscription.");
+      ack_mode = a;
+      callback = c;
+    }
 
-    let sub : &::subscription::Subscription;
-    let sub_id = match frame.headers.get_subscription() {
-      Some(Subscription(ref s)) => s.to_string(),
-      None => { 
-        debug!("Error: frame did not contain a subscription header.");
-        debug!("Frame: {}", frame);
-        return;
-      }
-    };
-    sub = match self.subscriptions.find_equiv(&sub_id.as_slice()) {
-      Some(sub) => sub,
-      None => {
-        debug!("Error: Received message for unknown subscription: {}", sub_id);
-        return;
-      }
-    };
     debug!("Executing.");
-    match sub.ack_mode {
+    let callback_result : AckOrNack = (callback)(&frame);
+    match ack_mode {
       Auto => {
         debug!("Auto ack, no frame sent.");
-        let _ = (sub.callback)(frame);
       }
       Client | ClientIndividual => {
-        let ack_id = match frame.headers.get_ack() {
-          Some(Ack(ack_id)) => ack_id.to_string(),
-          _ => {
-            debug!("Error: Message did not have an ack header.");
-            return;
-          }
-        };
-        match (sub.callback)(frame) {
-          ::subscription::Ack => {
-            let ack_frame = Frame::ack(ack_id.as_slice());
-            //match self.send(ack_frame) {
-            match ack_frame.write(&mut self.connection.writer) {
-              Err(error) => {
-                debug!("Couldn't send ACK: {}", error);
-                return;
-              },
-              _ => debug!("ACK sent.")
-            }
-          },
-          ::subscription::Nack => {
-            let nack_frame = Frame::nack(ack_id.as_slice());
-            //match self.send(nack_frame) {
-            match nack_frame.write(&mut self.connection.writer) {
-              Err(error) => {
-                debug!("Couldn't send NACK: {}", error);
-                return;
-              },
-              _ => debug!("NACK sent.")
-            }
-          } // Nack
-        } // match
+        let header::Ack(ack_id) = 
+          frame.headers
+          .get_ack()
+          .expect("Message did not have an 'ack' header.");
+        match callback_result {
+          Ack =>  self.acknowledge_frame(ack_id),
+          Nack => self.negatively_acknowledge_frame(ack_id)
+        }.unwrap_or_else(|error|fail!(format!("Could not acknowledge frame: {}", error)));
       } // Client | ...
     }
   } 
 
+  fn acknowledge_frame(&mut self, ack_id: &str) -> IoResult<()> {
+    let ack_frame = Frame::ack(ack_id);
+    self.send(ack_frame)
+  }
+
+  fn negatively_acknowledge_frame(&mut self, ack_id: &str) -> IoResult<()>{
+    let nack_frame = Frame::nack(ack_id);
+    self.send(nack_frame)
+  }
+
   pub fn receive(&mut self) -> IoResult<()>{
     let frame = try!(self.receive_frame());
     debug!("Received '{}' frame, dispatching.", frame.command);
diff --git a/src/subscription.rs b/src/subscription.rs
index f6d3a4f..06d9994 100644
--- a/src/subscription.rs
+++ b/src/subscription.rs
@@ -25,11 +25,11 @@ pub struct Subscription {
   pub id : String,
   pub topic: String,
   pub ack_mode: AckMode,
-  pub callback: fn(Frame)-> AckOrNack
+  pub callback: fn(&Frame)-> AckOrNack
 }
 
 impl Subscription {
-  pub fn new(id: uint, topic: &str, ack_mode: AckMode, callback: fn(Frame)->AckOrNack) -> Subscription {
+  pub fn new(id: uint, topic: &str, ack_mode: AckMode, callback: fn(&Frame)->AckOrNack) -> Subscription {
     Subscription {
       id: format!("stomp-rs/{}",id),
       topic: topic.to_string(),
diff --git a/src/transaction.rs b/src/transaction.rs
index d72c122..7e8dd52 100644
--- a/src/transaction.rs
+++ b/src/transaction.rs
@@ -21,7 +21,7 @@ impl <'a> Transaction<'a> {
     send_frame.headers.push(
       Header::encode_key_value("transaction", self.id.as_slice())
     );
-    Ok(try!(send_frame.write(&mut self.session.connection.writer)))
+    Ok(try!(self.session.send(send_frame)))
   }
 
   pub fn send_text(&mut self, topic: &str, body: &str) -> IoResult<()> {
@@ -30,16 +30,16 @@ impl <'a> Transaction<'a> {
 
   pub fn begin(&mut self) -> IoResult<()> {
     let begin_frame = Frame::begin(self.id.as_slice());
-    Ok(try!(begin_frame.write(&mut self.session.connection.writer)))
+    Ok(try!(self.session.send(begin_frame)))
   }
 
   pub fn commit(&mut self) -> IoResult<()> {
     let commit_frame = Frame::commit(self.id.as_slice());
-    Ok(try!(commit_frame.write(&mut self.session.connection.writer)))
+    Ok(try!(self.session.send(commit_frame)))
   }
 
   pub fn abort(&mut self) -> IoResult<()> {
-    let abort_frame = Frame::commit(self.id.as_slice());
-    Ok(try!(abort_frame.write(&mut self.session.connection.writer)))
+    let abort_frame = Frame::abort(self.id.as_slice());
+    Ok(try!(self.session.send(abort_frame)))
   }
 }


Test patch: