Adding files for serial_listener.

examples/serial_listener_example.cc

@@ -0,0 +1,46 @@
+#include <iostream>
+
+#include <serial/serial.h>
+#include <serial/serial_listener.h>
+
+using namespace serial;
+
+void default_handler(std::string line) {
+  std::cout << "default_handler got a: " << line << std::endl;
+}
+
+void callback(std::string line) {
+  std::cout << "callback got a: " << line << std::endl;
+}
+
+bool comparator(std::string line) {
+  if (line.substr(0,2) == "V=")
+    return true;
+  return false;
+}
+
+int main(void) {
+  Serial serial("/dev/tty.usbmodemfd1231", 115200);
+
+  SerialListener listener;
+  // Set the time to live for messages to 1 second
+  listener.setTimeToLive(1000);
+  listener.startListening(serial);
+
+  listener.listenFor(comparator, callback);
+
+  serial.write("?$1E\r");
+  if (!listener.listenForOnce("?$1E")) {
+    std::cerr << "Didn't get conformation of device version!" << std::endl;
+    return;
+  }
+
+}
+
+/*
+TODO:
+
+listenForOnce -> listenForStringOnce
+listenForOnce(ComparatorType comparator, std::string& result, size_t timeout)
+
+*/

include/serial/serial_listener.h

@@ -0,0 +1,194 @@
+/*!
+ * \file serial/serial_listener.h
+ * \author William Woodall <wjwwood@gmail.com>
+ * \version 0.1
+ *
+ * \section LICENSE
+ *
+ * The BSD License
+ *
+ * Copyright (c) 2011 William Woodall
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a 
+ * copy of this software and associated documentation files (the "Software"), 
+ * to deal in the Software without restriction, including without limitation 
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense, 
+ * and/or sell copies of the Software, and to permit persons to whom the 
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING 
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER 
+ * DEALINGS IN THE SOFTWARE.
+ *
+ * \section DESCRIPTION
+ *
+ * This provides a class that allows for asynchronous serial port reading.
+ * 
+ */
+
+// Serial
+#include <serial/serial.h>
+
+// Boost
+#include <boost/function.hpp>
+#include <boost/algorithm/string.hpp>
+#include <boost/uuid/uuid.hpp>
+#include <boost/uuid/uuid_generators.hpp>
+#include <boost/date_time/posix_time/posix_time.hpp>
+
+#ifndef SERIAL_LISTENER_TEST
+#define SERIAL_LISTENER_TEST false
+#endif
+
+namespace serial {
+
+/*!
+ * This is a general function type that is used both as the callback prototype 
+ * for asynchronous functions like the default handler callback and the 
+ * listenFor callbacks.
+ * 
+ * The function takes a std::string reference and returns nothing, it is 
+ * simply passing the resulting line detected by the comparator to the user's 
+ * callback for processing.
+ * 
+ * \see SerialListener::listenFor, SerialListener::setDefaultHandler
+ */
+typedef boost::function<void(const std::string&)> SerialCallback;
+
+/*!
+ * This is a general function type that is used as the comparator callback 
+ * prototpe for the listenFor* type functions.
+ * 
+ * The function takes a std::string reference and returns true if the string 
+ * matches what the comparator is looking for and false if it does not, unless 
+ * otherwise specified.
+ * 
+ * \see SerialListener::listenFor, SerialListener::listenForOnce
+ */
+typedef boost::function<bool(const std::string&)> ComparatorType;
+
+
+typedef boost::function<void(const std::string&)> InfoCallback;
+typedef boost::function<void(const std::string&)> WarningCallback;
+typedef boost::function<void(const std::string&)> DebugCallback;
+typedef boost::function<void(const std::exception&)> ExceptionCallback;
+
+typedef boost::uuids::uuid uuid_t;
+
+class SerialListenerException : public std::exception {
+  const char * e_what;
+public:
+  SerialListenerException(const char * e_what) {this->e_what = e_what;}
+
+  virtual const char* what() const throw() {
+    std::stringstream ss;
+    ss << "Error listening to serial port: " << this->e_what;
+    return ss.str().c_str();
+  }
+};
+
+/*!
+ * Listens to a serial port, facilitates asynchronous reading
+ */
+class SerialListener
+{
+public:
+  /*!
+   * Creates a new Serial Listener.
+   */
+  SerialListener ();
+  
+  /*!
+   * Destructor.
+   */
+  virtual ~SerialListener ();
+
+  /*!
+   * Sets the time-to-live (ttl) for messages waiting to be processsed.
+   * 
+   * \param ms Time in milliseconds until messages are purged from the buffer.
+   */
+  void setTimeToLive (size_t ms);
+
+  /*!
+   * Starts a thread to listen for messages and process them through filters.
+   * 
+   * \param serial_port Pointer to a serial::Serial object that is used to 
+   * retrieve new data.
+   */
+  void startListening (serial::Serial * serial_port);
+
+  /*!
+   * Stops the listening thread and blocks until it completely stops.
+   */
+  void stopListening ();
+
+  /*!
+   * Blocks until the given string is detected or until the timeout occurs.
+   * 
+   * \param token std::string that should be watched for, this string must 
+   * match the message exactly.
+   * 
+   * \param timeout in milliseconds before timing out and returning false.
+   * Defaults to 1000 milliseconds or 1 second.
+   * 
+   * \return bool If true then the token was detected before the token, false 
+   * if the token was not heard and the timeout occured.
+   */
+  bool listenForStringOnce (std::string token, size_t timeout = 1000);
+
+  boost::uuids::uuid listenFor (ComparatorType, SerialCallback);
+  void stopListeningFor (boost::uuids::uuid filter_uuid);
+
+  InfoCallback info;
+  WarningCallback warn;
+  DebugCallback debug;
+  ExceptionCallback handle_exc;
+  SerialCallback default_handler;
+
+private:
+  void listen();
+  std::string listenOnce(std::string data);
+  size_t determineAmountToRead();
+  bool listenForOnceComparator(std::string line);
+
+  bool listening;
+
+  serial::Serial * serial_port;
+
+  boost::thread listen_thread;
+  boost::uuids::random_generator random_generator();
+
+  std::string buffer;
+  std::map<const uuid_t,std::string> lines;
+  std::map<const uuid_t,boost::posix_time::ptime> ttls;
+  boost::posix_time::time_duration ttl;
+
+  // map<uuid, filter type (blocking/non-blocking)>
+  std::map<const uuid_t,std::string> filters;
+  // map<uuid, comparator>
+  std::map<const uuid_t,ComparatorType> comparators;
+  // map<uuid, callback>
+  std::map<const uuid_t,SerialCallback> callbacks;
+  // map<uuid, conditional_variables>
+  std::map<const uuid_t,boost::condition_variable*>
+  condition_vars;
+
+  // ptime time_start(microsec_clock::local_time());
+  // //... execution goes here ...
+  // ptime time_end(microsec_clock::local_time());
+  // time_duration duration(time_end - time_start);
+
+  std::string current_listen_for_one_target;
+
+  boost::mutex filter_mux;
+};
+
+}

src/serial_listener.cc

@@ -0,0 +1,267 @@
+#include "mdc2250/serial_listener.h"
+
+/***** Inline Functions *****/
+
+inline void defaultWarningCallback(const std::string& msg) {
+  std::cout << "SerialListener Warning: " << msg << std::endl;
+}
+
+inline void defaultDebugCallback(const std::string& msg) {
+  std::cout << "SerialListener Debug: " << msg << std::endl;
+}
+
+inline void defaultInfoCallback(const std::string& msg) {
+  std::cout << "SerialListener Info: " << msg << std::endl;
+}
+
+inline void defaultExceptionCallback(const std::exception &error) {
+  std::cerr << "SerialListener Unhandled Exception: " << error.what();
+  std::cerr << std::endl;
+  throw(error);
+}
+
+using namespace serial;
+
+/***** Listener Class Functions *****/
+
+SerialListener::SerialListener() : listening(false) {
+  // Set default callbacks
+  this->handle_exc = defaultExceptionCallback;
+  this->info = defaultInfoCallback;
+  this->debug = defaultDebugCallback;
+  this->warn = defaultWarningCallback;
+  this->default_handler = NULL;
+  
+  // Set default ttl
+  using namespace boost::posix_time;
+  this->ttl = time_duration(milliseconds(1000));
+}
+
+SerialListener::~SerialListener() {
+  
+}
+
+void SerialListener::setTimeToLive(size_t ms) {
+  this->ttl = time_duration(boost::posix_time::milliseconds(ms));
+}
+
+void SerialListener::startListening(Serial * serial_port) {
+  if (this->listening) {
+    throw(SerialListenerException("Already listening."));
+    return;
+  }
+  this->listening = true;
+  
+  this->serial_port = serial_port;
+  if (!this->serial_port->isOpen()) {
+    throw(SerialListenerException("Serial port not open."));
+    return;
+  }
+  
+  listen_thread = boost::thread(boost::bind(&SerialListener::listen, this));
+}
+
+void SerialListener::stopListening() {
+  listening = false;
+  listen_thread.join();
+  this->serial_port = NULL;
+}
+
+void SerialListener::listen() {
+  // Make sure there is a serial port
+  if (this->serial_port == NULL) {
+    this->handle_exc(SerialListenerException("Invalid serial port."));
+  }
+  // Make sure the serial port is open
+  if (!this->serial_port->isOpen()) {
+    this->handle_exc(SerialListenerException("Serial port not open."));
+  }
+  try {
+    while (this->listening) {
+      // Determine how much to read in
+      size_t amount_to_read = determineAmountToRead();
+      // Read some
+      std::string temp = this->serial_port->read(amount_to_read);
+      if (temp.length() == 0) {
+        // If nothing was read don't interate through the filters
+        continue;
+      }
+      this->buffer += temp;
+      if (this->buffer.find("\r") == std::string::npos) {
+        // If there is no return carrage in the buffer, then a command hasn't 
+        //  been completed.
+        continue;
+      }
+      // Listen once
+      buffer = this->listenOnce(buffer);
+      // Done parsing lines and buffer should now be set to the left overs
+    } // while (this->listening)
+  } catch (std::exception &e) {
+    this->handle_exc(SerialListenerException(e.what()));
+  }
+}
+
+std::string SerialListener::listenOnce(std::string data) {
+  std::string left_overs;
+// TODO: Make the delimeter settable
+  // Split the buffer by the delimeter
+  std::vector<std::string> new_lines;
+  boost::split(new_lines, data, boost::is_any_of("\r")); // it only uses \r
+  // Iterate through new lines and add times to them
+  std::vector<std::string>::iterator it_lines;
+  for(it_lines=new_lines.begin(); it_lines!=new_lines.end(); it_lines++) {
+    // The last line needs to be put back in the buffer always:
+    //  In the case that the string ends with \r the last element will be 
+    //  empty ("").  In the case that it does not the last element will be 
+    //  what is left over from the next message that hasn't sent 
+    //  everything.  Ex.: "?$1E\r" -> ["?$1E", ""] and 
+    //  "?$1E\r$1E=Robo" -> ["?$1E","$1E=Robo"]
+    if (it_lines == new_lines.end()-1) {
+      left_overs = (*it_lines);
+      continue;
+    }
+    uuid_t uuid = random_generator();
+    lines.insert(std::pair<const uuid_t,std::string>(uuid,(*it_lines)));
+    using namespace boost::posix_time;
+    ttls.insert(std::pair<const uuid_t,ptime>
+                  (uuid,ptime(microsec_clock::local_time())));
+  }
+  // Iterate through the lines checking for a match
+  for(it_lines=lines.begin(); it_lines!=lines.end(); it_lines++) {
+    std::string line = (*it_lines).second;
+    uuid_t uuid = (*it_lines).first
+    // If the line is empty, continue
+    if (line.length() == 0) {
+      continue;
+    }
+    bool matched = false;
+    bool erased = false;
+    // Get the filter lock
+    boost::mutex::scoped_lock l(filter_mux);
+    // Iterate through each filter
+    std::map<const uuid_t,std::string>::iterator it;
+    for(it=filters.begin(); it!=filters.end(); it++) {
+      if (comparators[(*it).first](line)) { // If comparator matches line
+        if ((*it).second == "non-blocking") {
+// TODO: Put this callback execution into a queue
+          // If non-blocking run the callback
+          callbacks[(*it).first](line);
+          lines.erase(uuid);
+          ttls.erase(uuid);
+          erased = true;
+        } else if ((*it).second == "blocking") {
+          // If blocking then notify the waiting call to continue
+          condition_vars[(*it).first]->notify_all();
+          lines.erase(uuid);
+          ttls.erase(uuid);
+          erased = true;
+        }
+        matched = true;
+        break; // It matched, continue to next line
+      }
+    } // for(it=filters.begin(); it!=filters.end(); it++)
+    // If the comparator doesn't match try another
+    if (!matched) { // Try to send to default handler
+      if (this->default_handler) {
+        this->default_handler(line);
+        lines.erase(uuid);
+        ttls.erase(uuid);
+        erased = true;
+      }
+    }
+    // If not already erased check how old it is, remove the too old
+    if (!erased) {
+      using namespace boost::posix_time;
+      if (ptime(microsec_clock::local_time())-ttls[uuid] > ttl) {
+        lines.erase(uuid);
+        ttls.erase(uuid);
+      }
+    }
+  } // for(it_lines=lines.begin(); it_lines!=lines.end(); it_lines++)
+  return left_overs;
+}
+
+size_t SerialListener::determineAmountToRead() {
+  // TODO: Make a more intelligent method based on the length of the things 
+  //  filters are looking for.  i.e.: if the filter is looking for 'V=XX\r' 
+  //  make the read amount at least 5.
+  return 5;
+}
+
+bool SerialListener::listenForOnceComparator(std::string line) {
+  if (line == current_listen_for_one_target)
+    return true;
+  return false;
+}
+
+bool SerialListener::listenForOnce(std::string token, size_t milliseconds) {
+  boost::condition_variable cond;
+  boost::mutex mut;
+  current_listen_for_one_target = token;
+
+  // Create blocking filter
+  uuid_t uuid = random_generator();
+  std::pair<const uuid_t,std::string>
+   filter_pair(uuid, "blocking");
+  std::pair<const uuid_t,ComparatorType>
+   comparator_pair(uuid,
+    boost::bind(&SerialListener::listenForOnceComparator, this, _1));
+  std::pair<const uuid_t,boost::condition_variable*>
+   condition_pair(uuid, &cond);
+  {
+    boost::mutex::scoped_lock l(filter_mux);
+    filters.insert(filter_pair);
+    comparators.insert(comparator_pair);
+    condition_vars.insert(condition_pair);
+  }
+
+  bool result = false;
+
+  // Wait
+  boost::unique_lock<boost::mutex> lock(mut);
+  if (cond.timed_wait(lock, boost::posix_time::milliseconds(milliseconds)))
+    result = true;
+
+  // Destroy the filter
+  {
+    boost::mutex::scoped_lock l(filter_mux);
+    filters.erase(uuid);
+    comparators.erase(uuid);
+    condition_vars.erase(uuid);
+  }
+
+  return result;
+}
+
+boost::uuids::uuid
+SerialListener::listenFor(ComparatorType comparator,
+                          SerialCallback callback)
+{
+  // Create Filter
+  uuid_t uuid = random_generator();
+  std::pair<const uuid_t,std::string>
+   filter_pair(uuid, "non-blocking");
+  std::pair<const uuid_t,ComparatorType>
+   comparator_pair(uuid, comparator);
+  std::pair<const uuid_t,SerialCallback>
+   callback_pair(uuid, callback);
+
+  {
+    boost::mutex::scoped_lock l(filter_mux);
+    filters.insert(filter_pair);
+    comparators.insert(comparator_pair);
+    callbacks.insert(callback_pair);
+  }
+
+  return uuid;
+}
+
+void SerialListener::stopListeningFor(boost::uuids::uuid filter_uuid) {
+  // Delete filter
+  boost::mutex::scoped_lock l(filter_mux);
+  filters.erase(filter_uuid);
+  comparators.erase(filter_uuid);
+  callbacks.erase(filter_uuid);
+}
+
+

tests/serial_listener_tests.cc

@@ -0,0 +1,122 @@
+#include "gtest/gtest.h"
+
+#include <boost/bind.hpp>
+
+#define SERIAL_LISTENER_TEST true
+
+// OMG this is so nasty...
+#define private public
+#define protected public
+
+#include "mdc2250/serial_listener.h"
+using namespace serial;
+
+static size_t global_count, global_listen_count;
+
+void default_handler(std::string line) {
+  global_count++;
+  // std::cout << "default_handler got: " << line << std::endl;
+}
+
+namespace {
+
+class SerialListenerTests : public ::testing::Test {
+protected:
+  virtual void SetUp() {
+    listener.default_handler = default_handler;
+  }
+
+  void execute_lookForOnce() {
+    listener.listenForOnce("?$1E", 1000);
+  }
+
+  SerialListener listener;
+
+};
+
+TEST_F(SerialListenerTests, ignoresEmptyString) {
+  global_count = 0;
+
+  listener.listenOnce("");
+
+  ASSERT_TRUE(global_count == 0);
+}
+
+TEST_F(SerialListenerTests, ignoresPartialMessage) {
+  global_count = 0;
+
+  listener.listenOnce("?$1E\r$1E=Robo");
+
+  ASSERT_EQ(global_count, 1);
+}
+
+TEST_F(SerialListenerTests, listenForOnceWorks) {
+  global_count = 0;
+
+  boost::thread t(
+    boost::bind(&SerialListenerTests::execute_lookForOnce, this));
+
+  boost::this_thread::sleep(boost::posix_time::milliseconds(100));
+
+  listener.listenOnce("\r+\r?$1E\r$1E=Robo");
+
+  ASSERT_TRUE(t.timed_join(boost::posix_time::milliseconds(1500)));
+
+  // Make sure the filters are getting deleted
+  ASSERT_EQ(listener.filters.size(), 0);
+
+  ASSERT_EQ(global_count, 1);
+}
+
+// lookForOnce should not find it, but timeout after 1000ms, so it should 
+//  still join.
+TEST_F(SerialListenerTests, listenForOnceTimesout) {
+  global_count = 0;
+
+  boost::thread t(
+    boost::bind(&SerialListenerTests::execute_lookForOnce, this));
+
+  boost::this_thread::sleep(boost::posix_time::milliseconds(100));
+
+  listener.listenOnce("\r+\r?$1ENOTRIGHT\r$1E=Robo");
+
+  ASSERT_TRUE(t.timed_join(boost::posix_time::milliseconds(1500)));
+
+  ASSERT_EQ(global_count, 2);
+}
+
+bool listenForComparator(std::string line) {
+  if (line.substr(0,2) == "V=") {
+    return true;
+  }
+  return false;
+}
+
+void listenForCallback(std::string line) {
+  global_listen_count++;
+}
+
+TEST_F(SerialListenerTests, listenForWorks) {
+  global_count = 0;
+  global_listen_count = 0;
+
+  boost::uuids::uuid filt_uuid = 
+    listener.listenFor(listenForComparator, listenForCallback);
+
+  listener.listenOnce("\r+\rV=05:06\r?$1E\rV=06:05\r$1E=Robo");
+
+  ASSERT_EQ(global_count, 2);
+  ASSERT_EQ(global_listen_count, 2);
+
+  listener.stopListeningFor(filt_uuid);
+
+  ASSERT_EQ(listener.filters.size(), 0);
+
+}
+
+}  // namespace
+
+int main(int argc, char **argv) {
+    ::testing::InitGoogleTest(&argc, argv);
+    return RUN_ALL_TESTS();
+}