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

Issue:
Migrate from old_io::Timer to whatever replaces it
#74 covered the migration from `old_io` to `io`, but there is not currently a migration path for `Timer`. This is needed for Heartbeating unless we switch to an evented system.



Discussion / hints:

Fix patch:
diff --git a/.gitignore b/.gitignore
index 1360ad7..a0e9596 100644
--- a/.gitignore
+++ b/.gitignore
@@ -2,4 +2,5 @@ src/**/target/**
 target/**
 examples/**/target/**
 **.swp
+**callgrind.out.*
 Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
index fad0337..dea4cd0 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -12,5 +12,14 @@ license = "MIT"
 [lib]
 name = "stomp"
 
+[dependencies]
+"unicode-segmentation" = "*"
+
 [dependencies.log]
 log = "*"
+
+[dependencies.mio]
+git = "https://github.com/carllerche/mio"
+
+[dependencies.bytes]
+git = "https://github.com/carllerche/bytes"
diff --git a/examples/send_and_subscribe/src/main.rs b/examples/send_and_subscribe/src/main.rs
index 5ca3ded..4d45517 100644
--- a/examples/send_and_subscribe/src/main.rs
+++ b/examples/send_and_subscribe/src/main.rs
@@ -27,27 +27,27 @@ fn main() {
     println!("Something went horribly wrong: {}", frame);
   });
 
-  let sub_id = session.subscription(destination, |frame: &Frame| {
+  let _ = session.subscription(destination, |frame: &Frame| {
     message_count += 1;
     println!("Received message #{}:\n{}", message_count, frame);
     Ack
   })
   .with(AckMode::Client)
   .with(Header::new("custom-subscription-header", "lozenge"))
-  .with(ReceiptHandler::new(|frame: &Frame| println!("Subscribed successfully.")))
+  .with(ReceiptHandler::new(|_: &Frame| println!("Subscribed successfully.")))
   .start();
 
-  session.message(destination, "Animal").send();
-  session.message(destination, "Vegetable").send();
-  session.message(destination, "Mineral").send();
+  let _ = session.message(destination, "Animal").send();
+  let _ = session.message(destination, "Vegetable").send();
+  let _ = session.message(destination, "Mineral").send();
 
-  session.message(destination, "Hypoteneuse".as_bytes())
+  let _ = session.message(destination, "Hypoteneuse".as_bytes())
     .with(ContentType("text/plain"))
     .with(Header::new("persistent", "true"))
-    .with(ReceiptHandler::new(|frame: &Frame| println!("Got a RECEIPT for 'Hypoteneuse'")))
+    .with(ReceiptHandler::new(|_: &Frame| println!("Got a RECEIPT for 'Hypoteneuse'")))
     .send();
 
-  session.listen(); // Loops infinitely, awaiting messages
+  let _ = session.listen(); // Loops infinitely, awaiting messages
 
-  session.disconnect();
+  let _ = session.disconnect();
 }
diff --git a/examples/send_lots/Cargo.toml b/examples/send_lots/Cargo.toml
new file mode 100644
index 0000000..786b4e4
--- /dev/null
+++ b/examples/send_lots/Cargo.toml
@@ -0,0 +1,10 @@
+[package]
+name = "send_lots"
+version = "0.1.0"
+authors = ["Zack Slayton <zack.slayton@gmail.com>"]
+
+[dependencies.stomp]
+stomp = "*"
+
+[dependencies.env_logger]
+env_logger = "*"
diff --git a/examples/send_lots/src/main.rs b/examples/send_lots/src/main.rs
new file mode 100644
index 0000000..b6c5433
--- /dev/null
+++ b/examples/send_lots/src/main.rs
@@ -0,0 +1,63 @@
+#![feature(scoped)]
+#![allow(unused_variables)]
+extern crate env_logger;
+extern crate stomp;
+use stomp::header::{ContentType};
+use stomp::subscription::AckOrNack::Ack;
+use stomp::frame::Frame;
+use std::thread;
+
+const TOTAL_MESSAGES : u64 = 10_000;
+const INTERVAL : u64 = 100;
+
+fn main() {
+  env_logger::init().unwrap();
+
+  let destination = "/queue/sullivan";
+  let mut messages_received: u64 = 0;
+
+  let mut subscribe_session = match stomp::session("127.0.0.1", 61613)
+    .start() {
+      Ok(session) => session,
+      Err(error)  => panic!("Could not connect to the server: {}", error)
+    };
+
+  let subscription = subscribe_session.subscription(destination, |_: &Frame| {
+    messages_received += 1;
+    if messages_received % INTERVAL == 0 {
+      println!("{} messages received ", messages_received);
+    }
+    if messages_received >= TOTAL_MESSAGES {
+      println!("Receive complete.");
+    }
+    Ack
+  }).start();
+
+  let join_guard = thread::scoped(move || {
+    let mut messages_sent: u64 = 0;
+    let mut publish_session = match stomp::session("127.0.0.1", 61613)
+      .start() {
+        Ok(session) => session,
+        Err(error)  => panic!("Could not connect to the server: {}", error)
+      };
+    loop {
+      let _ = publish_session.message(destination, "Modern major general")
+      .with(ContentType("text/plain"))
+      .send();
+      messages_sent += 1;
+      if messages_sent % INTERVAL == 0 {
+        println!("{} messages sent", messages_sent);
+      }
+      if messages_sent >= TOTAL_MESSAGES {
+        println!("Send complete.");
+        break;
+      }
+    }
+    let _ = publish_session.disconnect();
+    println!("Disconnected.");
+  });
+
+  let _ = subscribe_session.listen(); // Loops infinitely, awaiting messages
+
+  let _ = subscribe_session.disconnect();
+}
diff --git a/src/connection.rs b/src/connection.rs
index afbeb79..e3a1b2e 100644
--- a/src/connection.rs
+++ b/src/connection.rs
@@ -5,7 +5,7 @@ use std::io::BufWriter;
 use frame::Transmission;
 use std::io::Result;
 use std::io::Error;
-use std::io::ErrorKind::InvalidInput;
+use std::io::ErrorKind;
 use frame::Frame;
 use std::cmp::max;
 use header::{self, StompHeaderSet};
@@ -59,12 +59,13 @@ impl Connection {
         Transmission::CompleteFrame(frame) => {
           connected_frame = frame;
           break;
-        }
+        },
+        Transmission::ConnectionClosed => return Err(Error::new(ErrorKind::ConnectionAborted, "Connection closed by remote host."))
       } 
     }
     match connected_frame.command.as_ref() {
       "CONNECTED" => debug!("Received CONNECTED frame: {}", connected_frame),
-       _ => return Err(Error::new(InvalidInput, "Could not connect."))
+       _ => return Err(Error::new(ErrorKind::InvalidInput, "Could not connect."))
     }
     match connected_frame.headers.get_heart_beat() {
       Some(header::HeartBeat(tx_ms, rx_ms)) => Ok((tx_ms, rx_ms)),
diff --git a/src/frame.rs b/src/frame.rs
index 1f80781..b05c3bf 100644
--- a/src/frame.rs
+++ b/src/frame.rs
@@ -8,7 +8,6 @@ use std::io::Error;
 use std::io::ErrorKind::InvalidInput;
 use std::io::Write;
 use std::io::Read;
-use std::io::BufWriter;
 use std::io::BufRead;
 use std::str::from_utf8;
 use std::fmt;
@@ -46,7 +45,8 @@ pub struct Frame {
 
 pub enum Transmission {
   HeartBeat,
-  CompleteFrame(Frame)
+  CompleteFrame(Frame),
+  ConnectionClosed
 }
 
 impl fmt::Display for Frame {
@@ -88,7 +88,7 @@ impl Frame {
     frame_string
   }
 
-  pub fn write<W: Write>(&self, stream: &mut BufWriter<W>) -> Result<()> {
+  pub fn write<W: Write>(&self, stream: &mut W) -> Result<()> {
     debug!("Sending frame:\n{}", self.to_str());
     try!(stream.write(self.command.as_bytes()));
     try!(stream.write("\n".as_bytes()));
@@ -119,7 +119,10 @@ impl Frame {
   pub fn read<B: BufRead>(stream: &mut B) -> Result<Transmission> {
     let mut line : String = String::new();
     // Empty lines are interpreted as heartbeats
-    try!(stream.read_line(&mut line));
+    let bytes: usize = try!(stream.read_line(&mut line));
+    if bytes == 0 {
+      return Ok(Transmission::ConnectionClosed);
+    }
     line = Frame::chomp_line(line);
     if line.len() == 0 {
       return Ok(Transmission::HeartBeat);
diff --git a/src/frame_buffer.rs b/src/frame_buffer.rs
new file mode 100644
index 0000000..dfd7de7
--- /dev/null
+++ b/src/frame_buffer.rs
@@ -0,0 +1,276 @@
+use header::HeaderList;
+use header::Header;
+use header::ContentLength;
+use header::StompHeaderSet;
+use std::str::from_utf8;
+use frame::{Frame, Transmission};
+use std::collections::VecDeque;
+
+pub struct FrameBuffer {
+  buffer: VecDeque<u8>, 
+  parse_state: ParseState
+}
+
+struct ParseState {
+  command: Option<String>,
+  headers: HeaderList,
+  body: Option<Vec<u8>>,
+  section: FrameSection,
+}
+
+impl ParseState {
+  fn new() -> ParseState {
+    ParseState {
+      command: None,
+      headers: header_list![],
+      body: None,
+      section: FrameSection::Command
+    }
+  }
+}
+
+enum FrameSection {
+  Command,
+  Headers,
+  Body
+}
+
+enum ReadCommandResult {
+  HeartBeat,
+  Command(String),
+  Incomplete
+}
+
+enum ReadHeaderResult {
+  Header(Header),
+  EndOfHeaders,
+  Incomplete
+}
+
+enum ReadBodyResult {
+  Body(Vec<u8>),
+  Incomplete
+}
+
+impl FrameBuffer {
+  pub fn new() -> FrameBuffer {
+    FrameBuffer::with_capacity(1024 * 64)
+  }
+ 
+  pub fn with_capacity(capacity: usize) -> FrameBuffer {
+    FrameBuffer {
+      buffer: VecDeque::with_capacity(capacity),
+      parse_state: ParseState::new()
+    }
+  }
+
+  pub fn len(&self) -> usize {
+    self.buffer.len()
+  }
+
+  pub fn append(&mut self, bytes: &[u8]) {
+    for byte in bytes {
+      self.buffer.push_back(*byte);
+    }
+    debug!("Copied {} bytes into the frame buffer.", bytes.len());
+  }
+
+  pub fn read_transmission(&mut self) -> Option<Transmission> {
+    match self.parse_state.section {
+      FrameSection::Command => self.resume_parsing_at_command(), 
+      FrameSection::Headers => self.resume_parsing_at_headers(),
+      FrameSection::Body    => self.resume_parsing_at_body()
+    }
+  }
+
+  fn resume_parsing_at_command(&mut self) -> Option<Transmission> {
+    debug!("Parsing command.");
+    match self.read_command() {
+      ReadCommandResult::HeartBeat => Some(Transmission::HeartBeat),
+      ReadCommandResult::Command(command_string) => {
+        self.parse_state.command = Some(command_string);
+        self.parse_state.section = FrameSection::Headers;
+        self.resume_parsing_at_headers()
+      },
+      ReadCommandResult::Incomplete => None
+    }
+  }
+
+  fn resume_parsing_at_headers(&mut self) -> Option<Transmission> {
+    debug!("Parsing headers.");
+    match self.read_header() {
+      ReadHeaderResult::Header(header) => {
+        self.parse_state.headers.push(header);
+        self.resume_parsing_at_headers()
+      },
+      ReadHeaderResult::EndOfHeaders => {
+        self.parse_state.section = FrameSection::Body;
+        self.resume_parsing_at_body()
+      },
+      ReadHeaderResult::Incomplete => None
+    }
+  }
+
+//TODO: Eliminate clone()s
+  fn resume_parsing_at_body(&mut self) -> Option<Transmission> {
+    debug!("Parsing body.");
+    match self.read_body() {
+      ReadBodyResult::Body(body_bytes) => {
+        let command = match self.parse_state.command {
+          Some(ref command) => command.to_string(),
+          None => panic!("No COMMAND found.")
+        };
+        let headers = self.parse_state.headers.clone();
+        let body = body_bytes;
+        let frame = Frame {
+          command: command,
+          headers: headers,
+          body: body
+        };
+        self.reset_parse_state();
+        Some(Transmission::CompleteFrame(frame))
+      },
+      ReadBodyResult::Incomplete => None
+    }
+  }
+
+  fn reset_parse_state(&mut self) {
+    self.parse_state.command = None;
+    self.parse_state.headers = header_list![];
+    self.parse_state.body = None;
+    self.parse_state.section = FrameSection::Command;
+  }
+
+  // Replace these methods with bridge-buffer concept
+  fn read_into_vec(&mut self, n: usize) -> Vec<u8> {
+    let mut vec = Vec::with_capacity(n);
+    for _ in 0..n {
+      let byte = match self.buffer.pop_front() {
+        Some(byte) => byte,
+        None => panic!("Attempted to read beyond the end of the buffer!")
+      };
+      vec.push(byte);
+    }
+    debug!("Removed {} bytes from frame buffer, new size: {}", n, self.buffer.len());
+    vec 
+  }
+ 
+  // Wasteful with allocations 
+  fn read_into_string(&mut self, n: usize) -> String {
+    let vec = self.read_into_vec(n); 
+    let string = from_utf8(&vec)
+      .ok()
+      .expect("Attempted to read a string that was not utf8.");
+    return string.to_string();
+  }
+
+  fn chomp(mut line: String) -> String {
+    // This may be suboptimal, but fine for now
+    let chomped_length : usize;
+    {
+      let chars_to_remove : &[char] = &['\r', '\n'];
+      let trimmed_line : &str = line.trim_right_matches(chars_to_remove);
+      chomped_length = trimmed_line.len();
+    }
+    line.truncate(chomped_length);
+    line
+  }
+
+  fn read_command(&mut self) -> ReadCommandResult {
+    use frame_buffer::ReadCommandResult::*;
+    match self.find_next('\n' as u8) {
+      Some(index) => {
+        debug!("Found command ending @ index {}", index);
+        let num_bytes = index + 1;
+        let command = self.read_into_string(num_bytes as usize);
+        let command = FrameBuffer::chomp(command);
+        debug!("Chomped length: {}", command.len());
+        if command == "" {
+          debug!("Found HeartBeat");
+          return HeartBeat;
+        }
+        if command.len() == 1 {
+          debug!("Byte: {}", command.as_bytes()[0]);
+        }
+        debug!("Command -> '{}'", command);
+        Command(command)
+      },
+      None => Incomplete
+    }
+  }
+
+  fn read_header(&mut self) -> ReadHeaderResult {
+    match self.find_next('\n' as u8) {
+      Some(index) => {
+        debug!("Found header ending @ index {}", index);
+        let num_bytes = index + 1;
+        let header_string = self.read_into_string(num_bytes as usize);
+        let header_string = FrameBuffer::chomp(header_string); 
+        debug!("Header -> '{}'", header_string);
+        if header_string == "" {
+          return ReadHeaderResult::EndOfHeaders;
+        }
+        let header = Header::decode_string(&header_string).expect("Invalid header encountered.");
+        ReadHeaderResult::Header(header)
+      },
+      None => ReadHeaderResult::Incomplete
+    }
+  }
+  
+  fn read_body(&mut self) -> ReadBodyResult {
+    let maybe_body : Option<Vec<u8>> = match self.parse_state.headers.get_content_length() {
+      Some(ContentLength(num_bytes)) => self.read_body_by_content_length(num_bytes as usize),
+      None => self.read_body_by_null_octet()
+    };
+    match maybe_body {
+      Some(body) => ReadBodyResult::Body(body),
+      None => ReadBodyResult::Incomplete
+    }
+  }
+
+  fn read_body_by_content_length(&mut self, content_length: usize) -> Option<Vec<u8>> {
+    debug!("Reading body by content length.");
+    let bytes_needed = content_length + 1; // null octet
+    if self.buffer.len() < bytes_needed {
+      debug!("Not enough bytes to form body; needed {}, only had {}.", content_length, self.buffer.len());
+      return None;
+    }
+    let mut body = self.read_into_vec(bytes_needed);
+    body.pop(); // Discard null octet
+    debug!("Body -> '{}'", FrameBuffer::body_as_string(&body));
+    Some(body)
+  }
+
+  fn read_body_by_null_octet(&mut self) -> Option<Vec<u8>> {
+    debug!("Reading body by null octet.");
+    match self.find_next(0u8) {
+      Some(index) => {
+        debug!("Found body ending @ index {}", index);
+        let num_bytes = index + 1;
+        let mut body = self.read_into_vec(num_bytes as usize);
+        body.pop(); // Discard null octet
+        debug!("Body -> '{}'", FrameBuffer::body_as_string(&body));
+        Some(body) 
+      },
+      None => None
+    }
+  }
+
+  fn body_as_string(body: &Vec<u8>) -> &str {
+    match from_utf8(&body) {
+      Ok(ref s) => *s,
+      Err(_) => "<Non-utf8 Binary Content>"
+    }
+  }
+
+  fn find_next(&self, needle: u8) -> Option<u32> {
+    let mut step = 0u32;
+    for byte in &self.buffer {
+      if *byte == needle {
+        return Some(step);
+      }
+      step += 1;
+    }
+    None
+  }
+}
diff --git a/src/header.rs b/src/header.rs
index 4f6b82b..3b0dc4b 100644
--- a/src/header.rs
+++ b/src/header.rs
@@ -2,6 +2,7 @@
 #![macro_use]
 #![allow(non_camel_case_types)]
 use collections::slice::Iter;
+use unicode_segmentation::UnicodeSegmentation;
 
 // Ideally this would be a simple typedef. However:
 // See Rust bug #11047: https://github.com/mozilla/rust/issues/11047
@@ -90,22 +91,43 @@ impl Header {
     }
   }
 
-  //TODO: Optimize this method.
   fn decode_value(value: &str) -> String {
-    value
-      .replace(r"\c", ":")
-      .replace(r"\n", "\n")
-      .replace(r"\r", "\r")
-      .replace(r"\\", "\\")
+    let mut is_escaped = false;
+    let mut decoded = String::new();
+    for grapheme in UnicodeSegmentation::graphemes(value, true) {
+      if !is_escaped {
+        match grapheme {
+          r"\" => is_escaped = true,
+          g => decoded.push_str(g)
+        }
+        continue;
+      }
+    
+      match grapheme {
+        r"c" => decoded.push_str(":"),
+        r"r" => decoded.push_str("\r"),
+        r"n" => decoded.push_str("\n"),
+        r"\" => decoded.push_str("\\"),
+        g => panic!("Unrecognized escape sequence encountered: '\\{}'.", g)
+      }
+      
+      is_escaped = false;
+    }
+    decoded
   }
 
-//TODO: Optimize this method.
   fn encode_value(value: &str) -> String {
-    value
-      .replace("\\", r"\\") // Order is significant
-      .replace("\r", r"\r")
-      .replace("\n", r"\n")
-      .replace(":", r"\c")
+    let mut encoded = String::new();
+    for grapheme in UnicodeSegmentation::graphemes(value, true) {
+      match grapheme {
+        "\\" => encoded.push_str(r"\\"),// Order is significant
+        "\r" => encoded.push_str(r"\r"),
+        "\n" => encoded.push_str(r"\n"),
+        ":" => encoded.push_str(r"\c"),
+        g => encoded.push_str(g)
+      }
+    }
+    encoded
   }
 
   pub fn get_raw<'a>(&'a self) -> &'a str {
@@ -333,7 +355,7 @@ impl StompHeaderSet for HeaderList {
 #[macro_export]
 macro_rules! header_list [
   ($($header: expr), *) => ({
-    let mut header_list = HeaderList::new();
+    let header_list = HeaderList::new();
     $(header_list.push($header);)*
     header_list
   });
@@ -344,3 +366,59 @@ macro_rules! header_list [
   })
 
 ];
+
+#[test]
+fn encode_return_carriage() {
+  let unencoded = "Hello\rWorld";
+  let encoded = r"Hello\rWorld";
+  assert!(encoded == Header::encode_value(unencoded));
+}
+
+#[test]
+fn decode_return_carriage() {
+  let unencoded = "Hello\rWorld";
+  let encoded = r"Hello\rWorld";
+  assert!(unencoded == Header::decode_value(encoded));
+}
+
+#[test]
+fn encode_newline() {
+  let unencoded = "Hello\nWorld";
+  let encoded = r"Hello\nWorld";
+  assert!(encoded == Header::encode_value(unencoded));
+}
+
+#[test]
+fn decode_newline() {
+  let unencoded = "Hello\nWorld";
+  let encoded = r"Hello\nWorld";
+  assert!(unencoded == Header::decode_value(encoded));
+}
+
+#[test]
+fn encode_colon() {
+  let unencoded = "Hello:World";
+  let encoded = r"Hello\cWorld";
+  assert!(encoded == Header::encode_value(unencoded));
+}
+
+#[test]
+fn decode_colon() {
+  let unencoded = "Hello:World";
+  let encoded = r"Hello\cWorld";
+  assert!(unencoded == Header::decode_value(encoded));
+}
+
+#[test]
+fn encode_slash() {
+  let unencoded = r"Hello\World";
+  let encoded = r"Hello\\World";
+  assert!(encoded == Header::encode_value(unencoded));
+}
+
+#[test]
+fn decode_slash() {
+  let unencoded = r"Hello\World";
+  let encoded = r"Hello\\World";
+  assert!(unencoded == Header::decode_value(encoded));
+}
diff --git a/src/session.rs b/src/session.rs
index 11addde..51fe6bb 100644
--- a/src/session.rs
+++ b/src/session.rs
@@ -1,15 +1,9 @@
-use std::thread;
 use std::collections::hash_map::HashMap;
-use std::time::Duration;
-use std::io::BufReader;
+use std::io::Read;
 use std::io::Write;
-use std::io::BufWriter;
-use std::sync::mpsc::{Sender, Receiver, channel};
-use std::old_io::Timer;
 use std::io::Result;
 use std::io::Error;
 use std::io::ErrorKind::{Other};
-use std::net::TcpStream;
 use std::marker::PhantomData;
 use connection::Connection;
 use subscription::AckMode;
@@ -18,9 +12,8 @@ use subscription::AckOrNack;
 use subscription::AckOrNack::{Ack, Nack};
 use subscription::{Subscription, MessageHandler, ToMessageHandler};
 use frame::Frame;
-use frame::Transmission;
 use frame::ToFrameBody;
-use frame::Transmission::{HeartBeat, CompleteFrame};
+use frame::Transmission::{HeartBeat, CompleteFrame, ConnectionClosed};
 use header;
 use header::HeaderList;
 use header::ReceiptId;
@@ -28,6 +21,9 @@ use header::StompHeaderSet;
 use transaction::Transaction;
 use message_builder::MessageBuilder;
 use subscription_builder::SubscriptionBuilder;
+use frame_buffer::FrameBuffer;
+
+use mio::{EventLoop, Handler, Token, ReadHint, Timeout};
 
 pub trait FrameHandler {
   fn on_frame(&mut self, &Frame);
@@ -55,132 +51,140 @@ pub struct ReceiptHandler<'a, T> where T: 'a + ToFrameHandler<'a> {
 }
 
 impl<'a, T> ReceiptHandler<'a, T> where T: 'a + ToFrameHandler<'a> {
-  pub fn new(val: T) -> ReceiptHandler<'a T> {
+  pub fn new(val: T) -> ReceiptHandler<'a, T> {
     let r = ReceiptHandler { handler: val, _marker: PhantomData };
     r
   }
 }
 
+const READ_BUFFER_SIZE: usize = 64 * 1024;
+const GRACE_PERIOD_MULTIPLIER: f64 = 2.0;
+
 pub struct Session <'a> { 
   pub connection : Connection,
-  sender: Sender<Frame>,
-  receiver: Receiver<Frame>,
+  read_buffer: [u8; READ_BUFFER_SIZE],
+  frame_buffer: FrameBuffer,
   next_transaction_id: u32,
   next_subscription_id: u32,
   next_receipt_id: u32,
+  rx_heartbeat_ms: u64,
+  rx_heartbeat_timeout: Option<Timeout>,
+  tx_heartbeat_ms: u64,
   pub subscriptions: HashMap<String, Subscription <'a>>,
   pub receipt_handlers: HashMap<String, Box<FrameHandler + 'a>>,
   error_callback: Box<FrameHandler + 'a>
 }
 
-pub static GRACE_PERIOD_MULTIPLIER : f64 = 2.0f64;
+pub enum StompTimeout {
+  SendHeartBeat,
+  ReceiveHeartBeat
+}
+
+impl <'a> Handler for Session<'a> {
+  type Timeout = StompTimeout;
+  type Message = ();
+
+  fn timeout(&mut self, event_loop: &mut EventLoop<Session<'a>>, timeout: StompTimeout) {
+    match timeout {
+      StompTimeout::SendHeartBeat => self.send_heartbeat(event_loop),
+      StompTimeout::ReceiveHeartBeat => {
+        debug!("Did not receive a heartbeat in time.");
+      },
+    }
+  }
+
+  fn readable(&mut self, event_loop: &mut EventLoop<Session<'a>>, _token: Token, _: ReadHint) {
+    debug!("Readable!");
+    debug!("Buffer size: {}", &mut self.read_buffer.len());
+    debug!("Frame buffer length: {}", &mut self.frame_buffer.len());
+    let bytes_read = match self.connection.tcp_stream.read(&mut self.read_buffer){
+      Ok(0) => panic!("Read 0 bytes. Connection closed by remote host."),
+      Ok(bytes_read) => bytes_read,
+      Err(error) => panic!("Error while reading: {}", error)
+    };
+    debug!("Read {} bytes", bytes_read);
+    self.frame_buffer.append(&self.read_buffer[..bytes_read]);
+    let mut num_frames = 0u32;
+    loop {
+      debug!("Reading from frame buffer");
+      match self.frame_buffer.read_transmission() {
+        Some(HeartBeat) => self.on_heartbeat(event_loop),
+        Some(CompleteFrame(frame)) => {
+          debug!("Received frame!:\n{}", frame);
+          self.reset_rx_heartbeat_timeout(event_loop);
+          self.dispatch(frame);
+          num_frames += 1;
+        },
+        Some(ConnectionClosed) => panic!("Connection closed by remote host."),
+        None => {
+          debug!("Done. Read {} frames.", num_frames);
+          break;
+        }
+      }
+    }
+  } 
+}
 
 impl <'a> Session <'a> {
   pub fn new(connection: Connection, tx_heartbeat_ms: u32, rx_heartbeat_ms: u32) -> Session<'a> {
-    let reading_stream = connection.tcp_stream.try_clone().unwrap();
-    let writing_stream = reading_stream.try_clone().unwrap();
-    let (sender_tx, sender_rx) : (Sender<Frame>, Receiver<Frame>) = channel();
-    let (receiver_tx, receiver_rx) : (Sender<Frame>, Receiver<Frame>) = channel();
-
     let modified_rx_heartbeat_ms : u32 = ((rx_heartbeat_ms as f64) * GRACE_PERIOD_MULTIPLIER) as u32;
-    let _ = thread::spawn(move || {
-      match modified_rx_heartbeat_ms {
-        0 => Session::receive_loop(receiver_tx, reading_stream),
-        _ => Session::receive_loop_with_heartbeat(receiver_tx, reading_stream, Duration::milliseconds(modified_rx_heartbeat_ms as i64))
-      } 
-    });
-    let _ = thread::spawn(move || {
-      match tx_heartbeat_ms {
-        0 => Session::send_loop(sender_rx, writing_stream),
-        _ => Session::send_loop_with_heartbeat(sender_rx, writing_stream, Duration::milliseconds(tx_heartbeat_ms as i64))
-      } 
-    });
 
     Session {
       connection: connection,
-      sender : sender_tx,
-      receiver : receiver_rx,
+      frame_buffer: FrameBuffer::new(),
+      //TODO: Make this configurable
+      read_buffer: [0; READ_BUFFER_SIZE],
       next_transaction_id: 0,
       next_subscription_id: 0,
       next_receipt_id: 0,
+      rx_heartbeat_ms: modified_rx_heartbeat_ms as u64,
+      rx_heartbeat_timeout: None,
+      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>
     }
   }
- 
- fn send_loop(frames_to_send: Receiver<Frame>, tcp_stream: TcpStream){
-    let mut writer : BufWriter<TcpStream> = BufWriter::new(tcp_stream);
-    loop {
-      let frame_to_send = frames_to_send.recv().ok().expect("Could not receive the next frame: communication was lost with the receiving thread.");
-      frame_to_send.write(&mut writer).ok().expect("Couldn't send message!");
-    }
-  }
 
-  fn send_loop_with_heartbeat(frames_to_send: Receiver<Frame>, tcp_stream: TcpStream, heartbeat: Duration){
-    let mut writer : BufWriter<TcpStream> = BufWriter::new(tcp_stream);
-    let mut timer = Timer::new().unwrap(); 
-    loop {
-      let timeout = timer.oneshot(heartbeat);
-      select! {
-        _ = timeout.recv() => {
-          debug!("Sending heartbeat...");
-          writer.write(&['\n' as u8]).ok().expect("Failed to send heartbeat.");
-          let _ = writer.flush();
-        },
-        frame_to_send = frames_to_send.recv() => {
-          frame_to_send.unwrap().write(&mut writer).ok().expect("Couldn't send message!");
-        }
-      }
+  fn register_tx_heartbeat_timeout(&self, event_loop: &mut EventLoop<Session<'a>>) {
+    if self.tx_heartbeat_ms <= 0 {
+      debug!("Heartbeat transmission ms is {}, no need to register a callback.", self.tx_heartbeat_ms);
+      return;
     }
+    let _ = event_loop.timeout_ms(StompTimeout::SendHeartBeat, self.tx_heartbeat_ms);
   }
 
-   fn receive_loop(frame_recipient: Sender<Frame>, tcp_stream: TcpStream){
-    let (trans_tx, trans_rx) : (Sender<Transmission>, Receiver<Transmission>) = channel();
-    let _ = thread::spawn(move || {
-      Session::read_loop(trans_tx, tcp_stream); 
-    });
-    loop {
-      match trans_rx.recv() {
-        Ok(HeartBeat) => debug!("Received heartbeat"),
-        Ok(CompleteFrame(frame)) => frame_recipient.send(frame).unwrap(),
-        Err(_) => panic!("Could not read Transmission from remote host: the reading thread has died.")
-      }
+  fn register_rx_heartbeat_timeout(&mut self, event_loop: &mut EventLoop<Session<'a>>) {
+    if self.rx_heartbeat_ms <= 0 {
+      debug!("Heartbeat receipt ms is {}, no need to register a callback.", self.tx_heartbeat_ms);
+      return;
     }
+    let timeout = event_loop
+      .timeout_ms(StompTimeout::ReceiveHeartBeat, self.rx_heartbeat_ms)
+      .ok()
+      .expect("Could not register a timeout to receive a heartbeat.");
+    self.rx_heartbeat_timeout = Some(timeout);
   }
- 
-
-  fn receive_loop_with_heartbeat(frame_recipient: Sender<Frame>, tcp_stream: TcpStream, heartbeat: Duration){
-    let (trans_tx, trans_rx) : (Sender<Transmission>, Receiver<Transmission>) = channel();
-    let _ = thread::spawn(move || {
-      Session::read_loop(trans_tx, tcp_stream); 
-    });
 
+  fn send_heartbeat(&mut self, event_loop: &mut EventLoop<Session<'a>>) {
+    debug!("Sending heartbeat");
+    self.connection.tcp_stream.write("\n".as_bytes()).ok().expect("Could not send a heartbeat. Connection failed.");
+    let _ = self.connection.tcp_stream.flush();
+    self.register_tx_heartbeat_timeout(event_loop);
+  }
 
-    let mut timer = Timer::new().unwrap(); 
-    loop {
-      let timeout = timer.oneshot(heartbeat);
-      select! {
-        _ = timeout.recv() => error!("Did not receive expected heartbeat!"),
-        transmission = trans_rx.recv() => {
-          match transmission {
-            Ok(HeartBeat) => debug!("Received heartbeat"),
-            Ok(CompleteFrame(frame)) => frame_recipient.send(frame).unwrap(),
-            Err(_) => panic!("Could not read Transmission from remote host: the readin thread has died.")
-          }
-        }
-      }
-    }
+  fn on_heartbeat(&mut self, event_loop: &mut EventLoop<Session<'a>>) {
+    debug!("Received HeartBeat");
+    self.reset_rx_heartbeat_timeout(event_loop);
   }
 
-  fn read_loop(transmission_listener: Sender<Transmission>, tcp_stream: TcpStream){
-    let mut reader : BufReader<TcpStream> = BufReader::new(tcp_stream);
-    loop {
-      match Frame::read(&mut reader){
-         Ok(transmission) => transmission_listener.send(transmission).unwrap(),
-         Err(error) => panic!("Couldn't read from server!: {}", error)
-      }
-    }
+  fn reset_rx_heartbeat_timeout(&mut self, event_loop: &mut EventLoop<Session<'a>>) {
+    debug!("Resetting heartbeat rx timeout");
+    self.rx_heartbeat_timeout.map(|timeout| {
+      let result = event_loop.clear_timeout(timeout);
+      debug!("Reset complete -> {}", result);
+    });
+    self.register_rx_heartbeat_timeout(event_loop);
   }
 
   fn default_error_callback(frame : &Frame) {
@@ -263,25 +267,18 @@ impl <'a> Session <'a> {
   }
 
   pub fn begin_transaction<'b>(&'b mut self) -> Result<Transaction<'b, 'a>> {
-    let transaction = Transaction::new(self.generate_transaction_id(), self);
+    let mut transaction = Transaction::new(self.generate_transaction_id(), self);
     let _ = try!(transaction.begin());
     Ok(transaction)
   }
 
-  pub fn send(&self, frame: Frame) -> Result<()> {
-    match self.sender.send(frame) {
+  pub fn send(&mut self, frame: Frame) -> Result<()> {
+    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."))
     }
   }
 
-  pub fn receive(&self) -> Result<Frame> {
-    match self.receiver.recv() {
-      Ok(frame) => Ok(frame),//FIXME: Replace 'Other' below with a more meaningful ErrorKind
-      Err(_) => Err(Error::new(Other, "Could not receive frame: the connection to the server was lost."))
-    }
-  }
-
   pub fn dispatch(&mut self, frame: Frame) {
     // Check for ERROR frame
     match frame.command.as_ref() {
@@ -343,10 +340,10 @@ impl <'a> Session <'a> {
   }
 
   pub fn listen(&mut self) -> Result<()> {
-    loop {
-      let frame = try!(self.receive());
-      debug!("Received '{}' frame, dispatching.", frame.command);
-      self.dispatch(frame)
-    }
+    let mut event_loop : EventLoop<Session<'a>> = EventLoop::new().unwrap();
+    let _ = event_loop.register(&self.connection.tcp_stream, Token(0));
+    self.register_tx_heartbeat_timeout(&mut event_loop);
+    self.register_rx_heartbeat_timeout(&mut event_loop);
+    event_loop.run(self)
   }
 }
diff --git a/src/stomp.rs b/src/stomp.rs
index 86cb4cb..aed8c76 100644
--- a/src/stomp.rs
+++ b/src/stomp.rs
@@ -2,13 +2,14 @@
 #![crate_type = "lib"]
 
 #![feature(collections)]
-#![feature(std_misc)]
-#![feature(old_io)]
 #![feature(slice_patterns)]
 
 #[macro_use]
 extern crate log;
 extern crate collections;
+extern crate bytes;
+extern crate mio;
+extern crate unicode_segmentation;
 
 use session_builder::SessionBuilder;
 
@@ -19,6 +20,7 @@ pub fn session<'a>(host: &'a str, port: u16) -> SessionBuilder<'a>{
 pub mod connection;
 pub mod header;
 pub mod frame;
+pub mod frame_buffer;
 pub mod session;
 pub mod subscription;
 pub mod transaction;
diff --git a/src/transaction.rs b/src/transaction.rs
index 02669c0..14b0ef4 100644
--- a/src/transaction.rs
+++ b/src/transaction.rs
@@ -27,7 +27,7 @@ impl <'a, 'session> Transaction<'a, 'session> {
     }
   }
 
-  pub fn begin<'b>(&'b self) -> Result<()> {
+  pub fn begin<'b>(&'b mut self) -> Result<()> {
     let begin_frame = Frame::begin(self.id.as_ref());
     self.session.send(begin_frame)
   }


Test patch: