src/serial_listener.cc

   1 #include "serial/serial_listener.h"
   2
   3 /***** Inline Functions *****/
   4
   5 inline void defaultWarningCallback(const std::string& msg) {
   6   std::cout << "SerialListener Warning: " << msg << std::endl;
   7 }
   8
   9 inline void defaultDebugCallback(const std::string& msg) {
  10   std::cout << "SerialListener Debug: " << msg << std::endl;
  11 }
  12
  13 inline void defaultInfoCallback(const std::string& msg) {
  14   std::cout << "SerialListener Info: " << msg << std::endl;
  15 }
  16
  17 inline void defaultExceptionCallback(const std::exception &error) {
  18   std::cerr << "SerialListener Unhandled Exception: " << error.what();
  19   std::cerr << std::endl;
  20   throw(error);
  21 }
  22
  23 inline bool defaultComparator(const std::string &token) {
  24   return true;
  25 }
  26
  27 using namespace serial;
  28
  29 /***** Listener Class Functions *****/
  30
  31 void
  32 SerialListener::default_handler(const std::string &token) {
  33   if (this->_default_handler)
  34     this->_default_handler(token);
  35 }
  36
  37 SerialListener::SerialListener() : listening(false), chunk_size(5) {
  38   // Set default callbacks
  39   this->handle_exc = defaultExceptionCallback;
  40   this->info = defaultInfoCallback;
  41   this->debug = defaultDebugCallback;
  42   this->warn = defaultWarningCallback;
  43
  44   // Default handler stuff
  45   this->_default_handler = NULL;
  46   this->default_comparator = defaultComparator;
  47   DataCallback tmp = boost::bind(&SerialListener::default_handler, this, _1);
  48   this->default_filter = FilterPtr(new Filter(default_comparator, tmp));
  49
  50   // Set default tokenizer
  51   this->setTokenizer(delimeter_tokenizer("\r"));
  52 }
  53
  54 SerialListener::~SerialListener() {
  55   if (this->listening) {
  56     this->stopListening();
  57   }
  58 }
  59
  60 void
  61 SerialListener::callback() {
  62   try {
  63     // <filter id, token>
  64     std::pair<FilterPtr,TokenPtr> pair;
  65     while (this->listening) {
  66       if (this->callback_queue.timed_wait_and_pop(pair, 10)) {
  67         std::cout << "Got something off the callback queue: ";
  68         std::cout << (*pair.second) << std::endl;
  69         if (this->listening) {
  70           try {
  71             pair.first->callback((*pair.second));
  72           } catch (std::exception &e) {
  73             this->handle_exc(e);
  74           }// try callback
  75         } // if listening
  76       } // if popped
  77     } // while (this->listening)
  78   } catch (std::exception &e) {
  79     this->handle_exc(SerialListenerException(e.what()));
  80   }
  81 }
  82
  83 void
  84 SerialListener::startListening(Serial &serial_port) {
  85   if (this->listening) {
  86     throw(SerialListenerException("Already listening."));
  87     return;
  88   }
  89   this->listening = true;
  90
  91   this->serial_port = &serial_port;
  92   if (!this->serial_port->isOpen()) {
  93     throw(SerialListenerException("Serial port not open."));
  94     return;
  95   }
  96
  97   listen_thread = boost::thread(boost::bind(&SerialListener::listen, this));
  98
  99   // Start the callback thread
 100   callback_thread =
 101    boost::thread(boost::bind(&SerialListener::callback, this));
 102 }
 103
 104 void
 105 SerialListener::stopListening() {
 106   // Stop listening and clear buffers
 107   listening = false;
 108
 109   listen_thread.join();
 110   callback_thread.join();
 111
 112   this->data_buffer = "";
 113   this->serial_port = NULL;
 114
 115   // Delete all the filters
 116   this->removeAllFilters();
 117 }
 118
 119 size_t
 120 SerialListener::determineAmountToRead() {
 121   // TODO: Make a more intelligent method based on the length of the things
 122   //  filters are looking for.  e.g.: if the filter is looking for 'V=XX\r'
 123   //  make the read amount at least 5.
 124   return this->chunk_size;
 125 }
 126
 127 void
 128 SerialListener::readSomeData(std::string &temp, size_t this_many) {
 129   // Make sure there is a serial port
 130   if (this->serial_port == NULL) {
 131     this->handle_exc(SerialListenerException("Invalid serial port."));
 132   }
 133   // Make sure the serial port is open
 134   if (!this->serial_port->isOpen()) {
 135     this->handle_exc(SerialListenerException("Serial port not open."));
 136   }
 137   temp = this->serial_port->read(this_many);
 138   std::cout << "Read(" << temp.length() << "): " << temp << std::endl;
 139 }
 140
 141 void
 142 SerialListener::filterNewTokens (std::vector<TokenPtr> new_tokens) {
 143   // Iterate through the filters, checking each against new tokens
 144   boost::mutex::scoped_lock lock(filter_mux);
 145   std::vector<FilterPtr>::iterator it;
 146   for (it=filters.begin(); it!=filters.end(); it++) {
 147     this->filter((*it), new_tokens);
 148   } // for (it=filters.begin(); it!=filters.end(); it++)
 149 }
 150
 151 // <filter_ptr,token_ptr>
 152 void
 153 SerialListener::filter (FilterPtr filter, std::vector<TokenPtr> &tokens)
 154 {
 155   // Iterate through the token uuids and run each against the filter
 156   std::vector<TokenPtr>::iterator it;
 157   for (it=tokens.begin(); it!=tokens.end(); it++) {
 158     TokenPtr token = (*it);
 159     if (filter->comparator((*token)))
 160       callback_queue.push(std::make_pair(filter,token));
 161   }
 162 }
 163
 164 void
 165 SerialListener::listen() {
 166   try {
 167     while (this->listening) {
 168       // Read some data
 169       std::string temp;
 170       this->readSomeData(temp, determineAmountToRead());
 171       // If nothing was read then we
 172       //  don't need to iterate through the filters
 173       if (temp.length() != 0) {
 174         // Add the new data to the buffer
 175         this->data_buffer += temp;
 176         // Call the tokenizer on the updated buffer
 177         std::vector<TokenPtr> new_tokens;
 178         this->tokenize(this->data_buffer, new_tokens);
 179         // Run the new tokens through existing filters
 180         this->filterNewTokens(new_tokens);
 181       }
 182       // Done parsing lines and buffer should now be set to the left overs
 183     } // while (this->listening)
 184   } catch (std::exception &e) {
 185     this->handle_exc(SerialListenerException(e.what()));
 186   }
 187 }
 188
 189 /***** Filter Functions *****/
 190
 191 FilterPtr
 192 SerialListener::createFilter(ComparatorType comparator, DataCallback callback)
 193 {
 194   FilterPtr filter_ptr(new Filter(comparator, callback));
 195
 196   boost::mutex::scoped_lock l(filter_mux);
 197   this->filters.push_back(filter_ptr);
 198
 199   return filter_ptr;
 200 }
 201
 202 BlockingFilterPtr
 203 SerialListener::createBlockingFilter(ComparatorType comparator) {
 204   return BlockingFilterPtr(
 205     new BlockingFilter(comparator, (*this)));
 206 }
 207
 208 BufferedFilterPtr
 209 SerialListener::createBufferedFilter(ComparatorType comparator,
 210                                      size_t buffer_size)
 211 {
 212   return BufferedFilterPtr(
 213     new BufferedFilter(comparator, buffer_size, (*this)));
 214 }
 215
 216 void
 217 SerialListener::removeFilter(FilterPtr filter_ptr) {
 218   boost::mutex::scoped_lock l(filter_mux);
 219   filters.erase(std::find(filters.begin(),filters.end(),filter_ptr));
 220 }
 221
 222 void
 223 SerialListener::removeFilter(BlockingFilterPtr blocking_filter) {
 224   this->removeFilter(blocking_filter->filter_ptr);
 225 }
 226
 227 void
 228 SerialListener::removeFilter(BufferedFilterPtr buffered_filter) {
 229   this->removeFilter(buffered_filter->filter_ptr);
 230 }
 231
 232 void
 233 SerialListener::removeAllFilters() {
 234   boost::mutex::scoped_lock l(filter_mux);
 235   filters.clear();
 236   callback_queue.clear();
 237 }
 238