serial_listener.h

Go to the documentation of this file.

#ifndef SERIAL_LISTENER_H
#define SERIAL_LISTENER_H

#ifndef SERIAL_LISTENER_DEBUG
#define SERIAL_LISTENER_DEBUG 0
#endif

// STL
#include <vector>
#include <stdint.h>
#include <iostream>

// Serial
#include <serial/serial.h>

// Concurrent Queue
#include "serial/utils/concurrent_queue.h"

// Boost
#include <boost/function.hpp>
#include <boost/algorithm/string.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/thread.hpp>

#if SERIAL_LISTENER_DEBUG
# warning SerialListener in debug mode
#endif

namespace serial {
  namespace utils {

typedef boost::shared_ptr<const std::string> TokenPtr;

typedef boost::function<void(const std::string&)> DataCallback;

typedef boost::function<bool(const std::string&)> ComparatorType;

typedef boost::function<void(const std::string&, std::vector<TokenPtr>&)>
TokenizerType;

typedef boost::function<void(const std::exception&)> ExceptionCallback;

class Filter
{
public:
  Filter (ComparatorType comparator, DataCallback callback)
  : comparator_(comparator), callback_(callback) {}
  virtual ~Filter () {}

  ComparatorType comparator_;
  DataCallback callback_;

private:
  // Disable copy constructors
  Filter(const Filter&);
  void operator=(const Filter&);
  const Filter& operator=(Filter);
};

typedef boost::shared_ptr<Filter> FilterPtr;

class BlockingFilter; // Forward declaration

typedef boost::shared_ptr<BlockingFilter> BlockingFilterPtr;

class BufferedFilter; // Forward declaration

typedef boost::shared_ptr<BufferedFilter> BufferedFilterPtr;

class SerialListenerException : public std::exception {
  const std::string e_what_;
public:
  SerialListenerException(const std::string &e_what) : e_what_(e_what) {}
  ~SerialListenerException() throw() {}

  virtual const char* what() const throw() {
    std::stringstream ss;
    ss << "SerialListenerException: " << this->e_what_;
    return ss.str().c_str();
  }
};

class SerialListener
{
public:
  SerialListener (size_t num_threads = 0);

  virtual ~SerialListener ();

/***** Configurations ******/

  void
  setTokenizer (TokenizerType tokenizer) {
    this->tokenize = tokenizer;
  }

  void
  setChunkSize (size_t chunk_size) {
    this->chunk_size_ = chunk_size;
  }

/***** Start and Stop Listening ******/

  void
  startListening (serial::Serial &serial_port);

  void
  stopListening ();

  bool
  isListening () {
    return this->listening;
  }

/***** Filter Functions ******/

  FilterPtr
  createFilter (ComparatorType comparator, DataCallback callback);

  BlockingFilterPtr
  createBlockingFilter (ComparatorType comparator);

  BufferedFilterPtr
  createBufferedFilter (ComparatorType comparator, size_t buffer_size = 1024);

  void
  removeFilter (FilterPtr filter_ptr);

  void
  removeFilter (BlockingFilterPtr blocking_filter);

  void
  removeFilter (BufferedFilterPtr buffered_filter);

  void
  removeAllFilters ();

/***** Hooks and Handlers ******/

  void setDefaultHandler(DataCallback default_handler) {
    this->_default_handler = default_handler;
  }

  void
  setExceptionHandler (ExceptionCallback exception_handler) {
    this->handle_exc = exception_handler;
  }

/***** Static Functions ******/

  static void
  sleep (long milliseconds) {
    boost::int64_t ms(milliseconds);
    boost::this_thread::sleep(boost::posix_time::milliseconds(ms));
  }

  static TokenizerType
  delimeter_tokenizer (std::string delimeter) {
    return boost::bind(&SerialListener::_delimeter_tokenizer,
                       _1, _2, delimeter);
  }

  static ComparatorType
  exactly (std::string exact_str) {
    return boost::bind(&SerialListener::_exactly, _1, exact_str);
  }

  static ComparatorType
  startsWith (std::string prefix) {
    return boost::bind(&SerialListener::_startsWith, _1, prefix);
  }

  static ComparatorType
  endsWith (std::string postfix) {
    return boost::bind(&SerialListener::_endsWith, _1, postfix);
  }

  static ComparatorType
  contains (std::string substr) {
    return boost::bind(_contains, _1, substr);
  }

private:
  // Disable copy constructors
  SerialListener(const SerialListener&);
  void operator=(const SerialListener&);
  const SerialListener& operator=(SerialListener);
  // delimeter tokenizer function
  static void
  _delimeter_tokenizer (const std::string &data,
                        std::vector<TokenPtr> &tokens,
                        std::string delimeter)
  {
    typedef std::vector<std::string> find_vector_type;
    find_vector_type t;
    boost::split(t, data, boost::is_any_of(delimeter));
    for (find_vector_type::iterator it = t.begin(); it != t.end(); ++it)
      tokens.push_back(TokenPtr( new std::string(*it) ));
  }
  // exact comparator function
  static bool
  _exactly (const std::string& token, std::string exact_str) {
#if SERIAL_LISTENER_DEBUG
    std::cerr << "In exactly callback(" << token.length() << "): `";
    std::cerr << token << "` == `" << exact_str << "`?: ";
    if (token == exact_str)
      std::cerr << "True";
    else
      std::cerr << "False";
    std::cerr << std::endl;
#endif
    return token == exact_str;
  }
  // startswith comparator function
  static bool
  _startsWith (const std::string& token, std::string prefix) {
#if SERIAL_LISTENER_DEBUG
    std::cerr << "In startsWith callback(" << token.length() << "): `";
    std::cerr << token << "` starts with `" << prefix;
    std::cerr << "`?: ";
    if (token.substr(0,prefix.length()) == prefix)
      std::cerr << "True";
    else
      std::cerr << "False";
    std::cerr << std::endl;
#endif
    return token.substr(0,prefix.length()) == prefix;
  }
  // endswith comparator function
  static bool
  _endsWith (const std::string& token, std::string postfix) {
#if SERIAL_LISTENER_DEBUG
    std::cerr << "In endsWith callback(";
    std::cerr << token.length();
    std::cerr << "): `" << token;
    std::cerr << "` ends with `" << postfix << "`?: ";
    if (token.substr(token.length()-postfix.length()) == postfix)
      std::cerr << "True";
    else
      std::cerr << "False";
    std::cerr << std::endl;
#endif
    return token.substr(token.length()-postfix.length()) == postfix;
  }
  // contains comparator function
  static bool
  _contains (const std::string& token, std::string substr) {
#if SERIAL_LISTENER_DEBUG
    std::cerr << "In contains callback(";
    std::cerr << token.length();
    std::cerr << "): `" << token;
    std::cerr << "` contains `" << substr << "`?: ";
    if (token.find(substr) != std::string::npos)
      std::cerr << "True";
    else
      std::cerr << "False";
    std::cerr << std::endl;
#endif
    return token.find(substr) != std::string::npos;
  }

  // Gets some data from the serial port
  void readSomeData (std::string &temp, size_t this_many) {
    // Make sure there is a serial port
    if (this->serial_port_ == NULL && this->listening) {
      this->handle_exc(SerialListenerException("Invalid serial port."));
    }
    // Make sure the serial port is open
    if (!this->serial_port_->isOpen() && this->listening) {
      this->handle_exc(SerialListenerException("Serial port not open."));
    }
    if (!this->listening) {
      temp = "";
      return;
    }
    temp = this->serial_port_->read(this_many);
  }
  // Runs the new_tokens through all the filters
  void filter (std::vector<TokenPtr> &tokens);
  // Function that loops while listening is true
  void listen ();
  // Target of callback thread
  void callback (size_t);
  // Determines how much to read on each loop of listen
  size_t determineAmountToRead ();

  // Tokenizer
  TokenizerType tokenize;

  // Exception handler
  ExceptionCallback handle_exc;

  // Default handler
  FilterPtr default_filter;
  DataCallback _default_handler;
  ComparatorType default_comparator;
  void default_handler(const std::string &token);

  // Persistent listening variables
  bool listening;
  char serial_port_padding[7];
  serial::Serial * serial_port_;
  boost::thread listen_thread;
  std::string data_buffer;
  size_t chunk_size_;

  // Callback related variables
  // filter id, token
  // filter id == 0 is going to be default handled
  ConcurrentQueue<std::pair<FilterPtr,TokenPtr> >
  callback_queue;
  size_t num_threads_;
  std::vector<boost::thread*> callback_threads;

  // Mutex for locking use of filters
  boost::mutex filter_mux;
  // vector of filter ids
  std::vector<FilterPtr> filters;

};

class BlockingFilter
{
public:
  BlockingFilter (ComparatorType comparator, SerialListener &listener) {
    this->listener_ = &listener;
    DataCallback cb = boost::bind(&BlockingFilter::callback, this, _1);
    this->filter_ptr = this->listener_->createFilter(comparator, cb);
  }

  virtual ~BlockingFilter () {
    this->listener_->removeFilter(filter_ptr);
    this->result = "";
    this->cond.notify_all();
  }

 std::string wait(long ms) {
    this->result = "";
    boost::unique_lock<boost::mutex> lock(this->mutex);
    this->cond.timed_wait(lock, boost::posix_time::milliseconds(ms));
    return this->result;
  }

  FilterPtr filter_ptr;

  void callback(const std::string& token) {
#if SERIAL_LISTENER_DEBUG
    std::cerr << "In BlockingFilter callback(" << token.length() << "): ";
    std::cerr << token << std::endl;
#endif
    this->cond.notify_all();
    this->result = token;
  }

private:
  SerialListener * listener_;
  boost::condition_variable cond;
  boost::mutex mutex;
  std::string result;

};

class BufferedFilter
{
public:
  BufferedFilter (ComparatorType comparator, size_t buffer_size,
                  SerialListener &listener)
  : buffer_size_(buffer_size)
  {
    this->listener_ = &listener;
    DataCallback cb = boost::bind(&BufferedFilter::callback, this, _1);
    this->filter_ptr = this->listener_->createFilter(comparator, cb);
  }

  virtual ~BufferedFilter () {
    this->listener_->removeFilter(filter_ptr);
    this->queue.clear();
    this->result = "";
  }

  std::string wait(long ms) {
    if (ms == 0) {
      if (!this->queue.try_pop(this->result)) {
        this->result = "";
      }
    } else {
      if (!this->queue.timed_wait_and_pop(this->result, ms)) {
        this->result = "";
      }
    }
    return result;
  }

  void clear() {
    queue.clear();
  }

  size_t count() {
    return queue.size();
  }

  size_t capacity() {
    return buffer_size_;
  }

  FilterPtr filter_ptr;

  void callback(const std::string &token) {
#if SERIAL_LISTENER_DEBUG
    std::cerr << "In BufferedFilter callback(" << token.length() << "): ";
    std::cerr << token << std::endl;
#endif
    std::string throw_away;
    if (this->queue.size() == this->buffer_size_) {
      this->queue.wait_and_pop(throw_away);
    }
    this->queue.push(token);
  }

private:
  size_t buffer_size_;
  SerialListener * listener_;
  ConcurrentQueue<std::string> queue;
  std::string result;

};

  } // namespace utils
} // namespace serial

#endif // SERIAL_LISTENER_H