Myth/serial
1
0
Fork 0
mirror of https://github.com/wjwwood/serial.git synced 2026-01-22 19:54:57 +08:00
Code Issues Projects Releases Wiki Activity

Merge branch 'boostless' of github.com:wjwwood/serial into boostless

Browse Source 
Conflicts:
	include/serial/impl/unix.h
	src/impl/unix.cc
	src/serial.cc
This commit is contained in:
John Harrison 2012-01-12 12:54:45 -06:00
commit 923cf7d14f
9 changed files with 761 additions and 700 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

118
examples/serial_listener_example.cc
View File

@ -5,59 +5,89 @@
using namespace serial; using namespace serial;
void default_handler(std::string line) { void default_handler(std::string token) {
std::cout << "default_handler got a: " << line << std::endl; std::cout << "default_handler got a: " << token << std::endl;
} }
void callback(std::string line) { void callback(std::string token) {
std::cout << "callback got a: " << line << std::endl; std::cout << "callback got a: " << token << std::endl;
} }
#if 1 int run() {
int main(void) { // Assuming this device prints the string 'pre-substr-post\r' at 100Hz
Serial serial("/dev/tty.usbmodemfd1231", 115200); Serial serial("/dev/tty.usbserial-A900cfJA", 115200);
SerialListener listener; SerialListener listener;
// Set the time to live for messages to 10 milliseconds
listener.setTimeToLive(10);
listener.startListening(serial); listener.startListening(serial);
listener.listenFor(SerialListener::startsWith("V="), callback); // Set the tokenizer
// This is the same as the default delimeter, so an explicit call to
// setTokenizer is not necessary if your data is \r delimited.
// You can create your own Tokenizer as well.
listener.setTokenizer(SerialListener::delimeter_tokenizer("\r"));
serial.write("?$1E\r"); // Method #1:
if (!listener.listenForStringOnce("?$1E")) { // comparator, callback - async
std::cerr << "Didn't get conformation of device version!" << std::endl; FilterPtr f1 =
listener.createFilter(SerialListener::startsWith("pre"), callback);
SerialListener::sleep(15); // Sleep 15ms, to let the data come in
listener.removeFilter(f1); // Not scoped, must be removed explicity
// Method #2:
// comparator - blocking
{
BlockingFilterPtr f2 =
listener.createBlockingFilter(SerialListener::endsWith("post"));
for (size_t i = 0; i < 3; i++) {
std::string token = f2->wait(100); // Wait for 100 ms or a matched token
if (token != "")
std::cout << "Found something ending with 'post'" << std::endl;
else
std::cout << "Did not find something ending with 'post'" << std::endl;
}
}
// BlockingFilter is scoped and will remove itself, so no removeFilter
// required, but a call like `listener.removeFilter(BlockingFilter) will
// remove it from the filter list so wait will always timeout.
// Method #3:
// comparator, token buffer size - blocking
{
// Give it a comparator, then a buffer size of 10
BufferedFilterPtr f3 =
listener.createBufferedFilter(SerialListener::contains("substr"), 10);
SerialListener::sleep(75); // Sleep 75ms, should have about 7
std::cout << "Caught " << f3->count();
std::cout << " tokens containing 'substr'" << std::endl;
for(size_t i = 0; i < 20; ++i) {
std::string token = f3->wait(5); // Pull message from the buffer
if (token == "") // If an empty string is returned, a timeout occured
break;
}
f3->clear(); // Empties the buffer
if (f3->wait(0) == "") // Non-blocking wait
std::cout << "We won the race condition!" << std::endl;
else
std::cout << "We lost the race condition..." << std::endl;
// The buffer is circular, so the oldest matches will be dropped first
}
// BufferedFilter is scoped and will remove itself just like BlockingFilter.
// Method #4:
// callback - async
// Gets called if a token doesn't match a filter
listener.setDefaultHandler(default_handler);
SerialListener::sleep(25); // Sleep 25 ms, so some default callbacks occur
return 0;
}
int main(void) {
try {
return run();
} catch (std::exception &e) {
std::cerr << e.what() << std::endl;
return 1; return 1;
} }
serial.write("?V\r");
serial.write("# 1\r");
while (true) {
// Sleep 100 ms
SerialListener::sleep(100);
} }
}
#endif
#if 0
int main(void) {
Serial serial("/dev/tty.usbmodemfd1231", 115200);
serial.write("?$1E\r");
SerialListener::sleep(10);
// if ("?$1E\r" != serial.read(5)) {
// std::cerr << "Didn't get conformation of device version!" << std::endl;
// return 1;
// }
serial.write("?V\r");
serial.write("# 1\r");
while (true) {
std::cout << serial.read(5) << std::endl;
}
}
#endif

2
include/serial/impl/unix.h
View File

@ -38,7 +38,7 @@
#include "serial/serial.h" #include "serial/serial.h"
namespace { namespace serial {
using std::string; using std::string;

6
include/serial/serial.h
View File

@ -33,7 +33,6 @@
* This provides a cross platform interface for interacting with Serial Ports. * This provides a cross platform interface for interacting with Serial Ports.
*/ */
#ifndef SERIAL_H #ifndef SERIAL_H
#define SERIAL_H #define SERIAL_H
@ -189,7 +188,9 @@ public:
std::string std::string
read (size_t size = 1); read (size_t size = 1);
std::string readline(size_t size = std::numeric_limits<std::size_t>::max(), std::string eol = "\n"); std::string readline(size_t size = std::numeric_limits<std::size_t>::max(),
std::string eol = "\n");
std::vector<std::string> readlines(std::string eol = "\n"); std::vector<std::string> readlines(std::string eol = "\n");
/*! Read a given amount of bytes from the serial port. /*! Read a given amount of bytes from the serial port.
@ -393,7 +394,6 @@ private:
// Pimpl idiom, d_pointer // Pimpl idiom, d_pointer
class SerialImpl; class SerialImpl;
// std::shared_ptr<Serial_pimpl> pimpl;
SerialImpl *pimpl; SerialImpl *pimpl;
}; };

537
include/serial/serial_listener.h
View File

@ -38,6 +38,7 @@
// STL // STL
#include <queue> #include <queue>
#include <stdint.h>
// Serial // Serial
#include <serial/serial.h> #include <serial/serial.h>
@ -45,12 +46,25 @@
// Boost // Boost
#include <boost/function.hpp> #include <boost/function.hpp>
#include <boost/algorithm/string.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> #include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/thread.hpp>
namespace serial { namespace serial {
/*!
* This is an alias to boost::shared_ptr<const std::string> used for tokens.
*
* This is the type used internally and is the type returned in a vector by
* the tokenizer. The shared_ptr allows for the token to be stored and kept
* around long enough to be used by the comparators and callbacks, but no
* longer. This internal storage is passed as a const std::string reference
* to callbacks, like the DataCallback function type, to prevent implicit
* copying.
*
* \see serial::TokenizerType, serial::SerialListener::setTokenizer
*/
typedef boost::shared_ptr<const std::string> TokenPtr;
/*! /*!
* This is a general function type that is used as the callback prototype * This is a general function type that is used as the callback prototype
* for asynchronous functions like the default handler callback and the * for asynchronous functions like the default handler callback and the
@ -105,32 +119,90 @@ typedef boost::function<void(const std::exception&)> ExceptionCallback;
* This function type describes the prototype for the tokenizer callback. * This function type describes the prototype for the tokenizer callback.
* *
* The function should take a std::string reference and tokenize it into a * The function should take a std::string reference and tokenize it into a
* several std::string's and store them in the given * several TokenPtr's and store them in the given std::vector<TokenPtr>
* std::vector<std::string> reference. There are some default ones or the * reference. There are some default ones or the user can create their own.
* user can create their own.
* *
* The last element in the std::vector of std::string's should always be * The last element in the std::vector of TokenPtr's should always be
* either an empty string ("") or the last partial message. The last element * either an empty string ("") or the last partial message. The last element
* in the std::vector will be put back into the data buffer so that if it is * in the std::vector will be put back into the data buffer so that if it is
* incomplete it can be completed when more data is read. * incomplete it can be completed when more data is read.
* *
* Example: A delimeter tokenizer with a delimeter of "\r". The result would * Example: A delimeter tokenizer with a delimeter of "\r". The result would
* be: "msg1\rmsg2\r" -> ["msg1", "msg2", ""] for all complete messages, or: * be: "msg1\rmsg2\r" -> ["msg1", "msg2", ""] for two complete messages, or:
* "msg1\rpartial_msg2" -> ["msg1","partial_msg2"] for partial messages. * "msg1\rpartial_msg2" -> ["msg1","partial_msg2"] for one complete message
* and one partial message.
* *
* \see SerialListener::setTokenizer, serial::delimeter_tokenizer * \see SerialListener::setTokenizer, serial::delimeter_tokenizer,
* serial::TokenPtr
*/ */
typedef boost::function<void(std::string&,std::vector<std::string>&)> typedef boost::function<void(const std::string&, std::vector<TokenPtr>&)>
TokenizerType; TokenizerType;
/*! This is a convenience alias for boost::uuids::uuid. */ /*!
typedef boost::uuids::uuid uuid_type; // uuid_t is already taken! =( * Represents a filter which new data is passed through.
*
* The filter consists of a comparator and a callback. The comparator takes a
* token and returns true if it matches, false if it doesn't. If a match
* occurs the serial listener will dispatch a call of the callback with the
* matched data in a another thread. The comparator should be as short as
* possible, but the callback can be longer since it is executed in a thread
* or thread pool.
*
* \param comparator A ComparatorType that matches incoming data, returns true
* for a match, false othewise.
*
* \param callback A DataCallback that gets called when a match occurs.
*
* \see serial::ComparatorType, serial::DataCallback, serial::FilterPtr
*/
class Filter
{
public:
Filter (ComparatorType comparator, DataCallback callback)
: comparator(comparator), callback(callback) {}
virtual ~Filter () {}
ComparatorType comparator;
DataCallback callback;
};
/*!
* This is an alias to boost::shared_ptr<Filter> used for tokens.
*
* This is used internally and is returned from SerialListener::listenFor like
* functions so that users can later remove those filters by passing the
* FilterPtr.
*
* \see serial::Filter, serial::SerialListener::listenFor,
* serial::SerialListener::listenForOnce
*/
typedef boost::shared_ptr<Filter> FilterPtr;
class BlockingFilter;
/*!
* Shared Pointer of BlockingFilter, returned by
* SerialListener::createBlockingFilter.
*
* \see serial::BlockingFilter, SerialListener::createBlockingFilter
*/
typedef boost::shared_ptr<BlockingFilter> BlockingFilterPtr;
class BufferedFilter;
/*!
* Shared Pointer of BufferedFilter, returned by
* SerialListener::createBufferedFilter.
*
* \see serial::BufferedFilter, SerialListener::createBufferedFilter
*/
typedef boost::shared_ptr<BufferedFilter> BufferedFilterPtr;
/*! /*!
* This is a general exception generated by the SerialListener class. * This is a general exception generated by the SerialListener class.
* *
* Check the SerialListenerException::what function for the cause. * Check the SerialListenerException::what function for the cause.
*
* \param e_what is a std::string that describes the cause of the error. * \param e_what is a std::string that describes the cause of the error.
*/ */
class SerialListenerException : public std::exception { class SerialListenerException : public std::exception {
@ -220,13 +292,6 @@ public:
} }
}; };
namespace filter_type {
typedef enum {
nonblocking, blocking
} FilterType;
}
/*! /*!
* Listens to a serial port, facilitates asynchronous reading * Listens to a serial port, facilitates asynchronous reading
*/ */
@ -245,22 +310,11 @@ public:
/***** Configurations ******/ /***** Configurations ******/
/*!
* Sets the time-to-live (ttl) for messages waiting to be processsed.
*
* Messages are processed before checking for expiration, therefore they
* will always be passed through filters once before being removed
* due to ttl expiration. The default value for this is 10 ms.
*
* \param ms Time in milliseconds until messages are purged from the buffer.
*/
void setTimeToLive (size_t ms = 10);
/*! /*!
* Sets the tokenizer to be used when tokenizing the data into tokens. * Sets the tokenizer to be used when tokenizing the data into tokens.
* *
* This function is given a std::string of data and is responsible for * This function is given a std::string of data and is responsible for
* tokenizing that data into a std::vector<std::string> of data tokens. * tokenizing that data into a std::vector<TokenPtr> of data tokens.
* The default tokenizer splits the data by the ascii return carriage. * The default tokenizer splits the data by the ascii return carriage.
* The user can create their own tokenizer or use one of the default ones. * The user can create their own tokenizer or use one of the default ones.
* *
@ -268,10 +322,21 @@ public:
* *
* \see serial::TokenizerType, serial::delimeter_tokenizer * \see serial::TokenizerType, serial::delimeter_tokenizer
*/ */
void setTokenizer (TokenizerType tokenizer) { void
setTokenizer (TokenizerType tokenizer) {
this->tokenize = tokenizer; this->tokenize = tokenizer;
} }
/*!
* Sets the number of bytes to be read at a time by the listener.
*
* \param chunk_size Number of bytes to be read at a time.
*/
void
setChunkSize (size_t chunk_size) {
this->chunk_size = chunk_size;
}
/***** Start and Stop Listening ******/ /***** Start and Stop Listening ******/
/*! /*!
@ -280,7 +345,8 @@ public:
* \param serial_port Pointer to a serial::Serial object that is used to * \param serial_port Pointer to a serial::Serial object that is used to
* retrieve new data. * retrieve new data.
*/ */
void startListening (serial::Serial &serial_port); void
startListening (serial::Serial &serial_port);
/*! /*!
* Stops the listening thread and blocks until it completely stops. * Stops the listening thread and blocks until it completely stops.
@ -288,40 +354,13 @@ public:
* This function also clears all of the active filters from listenFor and * This function also clears all of the active filters from listenFor and
* similar functions. * similar functions.
*/ */
void stopListening (); void
stopListening ();
/***** Filter Functions ******/ /***** Filter Functions ******/
/*! /*!
* Blocks until the given string is detected or until the timeout occurs. * Creates a filter that calls a callback when the comparator returns true.
*
* \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);
/*!
* Blocks until the comparator returns true or until the timeout occurs.
*
* \param comparator ComparatorType function that should return true if the
* given std::string matches otherwise false.
*
* \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 listenForOnce (ComparatorType comparator, size_t timeout = 1000);
/*!
* Setups up a filter that calls a callback when a comparator returns true.
* *
* The user provides a comparator and a callback, and every time a line is * The user provides a comparator and a callback, and every time a line is
* received the comparator is called and the comparator has to evaluate the * received the comparator is called and the comparator has to evaluate the
@ -335,23 +374,101 @@ public:
* \param callback This is the handler for when a match occurs. It is given * \param callback This is the handler for when a match occurs. It is given
* a std::string reference of the line that matched your comparator. * a std::string reference of the line that matched your comparator.
* *
* \return boost::uuids::uuid a unique identifier used to remove the filter. * \return boost::shared_ptr<Filter> so you can remove it later.
*
* \see SerialListener::removeFilter
*/ */
uuid_type listenFor (ComparatorType comparator, DataCallback callback); FilterPtr
createFilter (ComparatorType comparator, DataCallback callback);
/*! /*!
* Removes a filter by a given uuid. * Creates a BlockingFilter which blocks until the comparator returns true.
* *
* The uuid for a filter is returned by the listenFor function. * The user provides a comparator, and every time a line is
* received the comparator is called and the comparator has to evaluate the
* line and return true if it matches and false if it doesn't. If it does
* match, any threads that have called BlockingFilter::wait will be
* notified. The BlockingFilter will remove itself when its destructor is
* called, i.e. when it leaves the scope, so in those cases an explicit call
* to SerialListener::removeFilter is not needed.
* *
* \param filter_uuid The uuid of the filter to be removed. * \param comparator This is a comparator for detecting if a line matches.
* The comparartor receives a std::string reference and must return a true
* if it matches and false if it doesn't.
*
* \return BlockingFilterPtr So you can call BlockingFilter::wait on it.
*
* \see SerialListener::removeFilter, serial::BlockingFilter,
* serial::BlockingFilterPtr
*/ */
void stopListeningFor (uuid_type filter_uuid); BlockingFilterPtr
createBlockingFilter (ComparatorType comparator);
/*! /*!
* Stops listening for anything, but doesn't stop reading the serial port. * Creates a BlockingFilter blocks until the comparator returns true.
*
* The user provides a comparator, and every time a line is
* received the comparator is called and the comparator has to evaluate the
* line and return true if it matches and false if it doesn't. If it does
* match, any threads that have called BlockingFilter::wait will be
* notified. The BlockingFilter will remove itself when its destructor is
* called, i.e. when it leaves the scope, so in those cases an explicit call
* to SerialListener::removeFilter is not needed.
*
* \param comparator This is a comparator for detecting if a line matches.
* The comparartor receives a std::string reference and must return a true
* if it matches and false if it doesn't.
*
* \param buffer_size This is the number of tokens to be buffered by the
* BufferedFilter, defaults to 1024.
*
* \return BlockingFilter So you can call BlockingFilter::wait on it.
*
* \see SerialListener::removeFilter, serial::BufferedFilter,
* serial::BufferedFilterPtr
*/ */
void stopListeningForAll (); BufferedFilterPtr
createBufferedFilter (ComparatorType comparator, size_t buffer_size = 1024);
/*!
* Removes a filter by a given FilterPtr.
*
* \param filter_ptr A shared pointer to the filter to be removed.
*
* \see SerialListener::createFilter
*/
void
removeFilter (FilterPtr filter_ptr);
/*!
* Removes a BlockingFilter.
*
* The BlockingFilter will remove itself if the destructor is called.
*
* \param blocking_filter A BlockingFilter to be removed.
*
* \see SerialListener::createBlockingFilter
*/
void
removeFilter (BlockingFilterPtr blocking_filter);
/*!
* Removes a BufferedFilter.
*
* The BufferedFilter will remove itself if the destructor is called.
*
* \param buffered_filter A BufferedFilter to be removed.
*
* \see SerialListener::createBufferedFilter
*/
void
removeFilter (BufferedFilterPtr buffered_filter);
/*!
* Removes all filters.
*/
void
removeAllFilters ();
/***** Hooks and Handlers ******/ /***** Hooks and Handlers ******/
@ -368,8 +485,9 @@ public:
* *
* \see serial::DataCallback, SerialListener::setInfoHandler * \see serial::DataCallback, SerialListener::setInfoHandler
*/ */
void setDefaultHandler(DataCallback default_handler) { void
this->default_handler = default_handler; setDefaultHandler (DataCallback default_handler) {
this->_default_handler = default_handler;
} }
/*! /*!
@ -421,7 +539,8 @@ public:
* *
* \see serial::LoggingCallback * \see serial::LoggingCallback
*/ */
void setInfoHandler(LoggingCallback info_handler) { void
setInfoHandler (LoggingCallback info_handler) {
this->info = info_handler; this->info = info_handler;
} }
@ -438,7 +557,8 @@ public:
* *
* \see serial::LoggingCallback, SerialListener::setInfoHandler * \see serial::LoggingCallback, SerialListener::setInfoHandler
*/ */
void setDebugHandler(LoggingCallback debug_handler) { void
setDebugHandler (LoggingCallback debug_handler) {
this->debug = debug_handler; this->debug = debug_handler;
} }
@ -455,7 +575,8 @@ public:
* *
* \see serial::LoggingCallback, SerialListener::setInfoHandler * \see serial::LoggingCallback, SerialListener::setInfoHandler
*/ */
void setWarningHandler(LoggingCallback warning_handler) { void
setWarningHandler (LoggingCallback warning_handler) {
this->warn = warning_handler; this->warn = warning_handler;
} }
@ -491,12 +612,10 @@ public:
* \see SerialListener::setTokenizer, serial::TokenizerType * \see SerialListener::setTokenizer, serial::TokenizerType
*/ */
static TokenizerType static TokenizerType
delimeter_tokenizer (std::string delimeter); delimeter_tokenizer (std::string delimeter) {
return boost::bind(&SerialListener::_delimeter_tokenizer,
// delimeter tokenizer function _1, _2, delimeter);
static void }
_delimeter_tokenizer (std::string &data, std::vector<std::string> &tokens,
std::string delimeter);
/*! /*!
* This returns a comparator that matches only the exact string given. * This returns a comparator that matches only the exact string given.
@ -519,11 +638,9 @@ public:
* serial::ComparatorType * serial::ComparatorType
*/ */
static ComparatorType static ComparatorType
exactly (std::string exact_str); exactly (std::string exact_str) {
return boost::bind(&SerialListener::_exactly, _1, exact_str);
// exact comparator function }
static bool
_exactly (const std::string&, std::string);
/*! /*!
* This returns a comparator that looks for a given prefix. * This returns a comparator that looks for a given prefix.
@ -549,12 +666,6 @@ public:
return boost::bind(&SerialListener::_startsWith, _1, prefix); return boost::bind(&SerialListener::_startsWith, _1, prefix);
} }
// exact comparator function
static bool
_startsWith (const std::string& token, std::string prefix) {
return token.substr(0,prefix.length()) == prefix;
}
/*! /*!
* This returns a comparator that looks for a given postfix. * This returns a comparator that looks for a given postfix.
* *
@ -579,12 +690,6 @@ public:
return boost::bind(&SerialListener::_endsWith, _1, postfix); return boost::bind(&SerialListener::_endsWith, _1, postfix);
} }
// endswith comparator function
static bool
_endsWith (const std::string& token, std::string postfix) {
return token.substr(token.length()-postfix.length()) == postfix;
}
/*! /*!
* This returns a comparator that looks for a given substring in the token. * This returns a comparator that looks for a given substring in the token.
* *
@ -607,42 +712,55 @@ public:
*/ */
static ComparatorType static ComparatorType
contains (std::string substr) { contains (std::string substr) {
return boost::bind(&SerialListener::_contains, _1, substr); return boost::bind(_contains, _1, substr);
} }
private:
// 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) {
return token == exact_str;
}
// startswith comparator function
static bool
_startsWith (const std::string& token, std::string prefix) {
return token.substr(0,prefix.length()) == prefix;
}
// endswith comparator function
static bool
_endsWith (const std::string& token, std::string postfix) {
return token.substr(token.length()-postfix.length()) == postfix;
}
// contains comparator function // contains comparator function
static bool static bool
_contains (const std::string& token, std::string substr) { _contains (const std::string& token, std::string substr) {
return token.find(substr) != std::string::npos; return token.find(substr) != std::string::npos;
} }
private:
// Gets some data from the serial port // Gets some data from the serial port
void readSomeData (std::string&, size_t); void readSomeData (std::string&, size_t);
// Takes newly tokenized data and processes them
void addNewTokens(std::vector<std::string> &new_tokens,
std::vector<uuid_type> &new_uuids,
std::string &left_overs);
// Runs the new tokens through the filters // Runs the new tokens through the filters
void filterNewTokens (std::vector<uuid_type> new_uuids); void filterNewTokens (std::vector<TokenPtr> new_tokens);
// Runs a list of tokens through one filter // Given a filter_id and a list of tokens, return list of matched tokens
std::vector<std::pair<uuid_type,uuid_type> > void filter (FilterPtr filter, std::vector<TokenPtr> &tokens);
filter(uuid_type filter_uuid, std::vector<uuid_type> &token_uuids);
// Function that loops while listening is true // Function that loops while listening is true
void listen (); void listen ();
// Target of callback thread // Target of callback thread
void callback (); void callback ();
// Prune old tokens
void pruneTokens ();
// Erases one token
void eraseToken (uuid_type&);
// Erases several tokens
void eraseTokens (std::vector<uuid_type>&);
// Determines how much to read on each loop of listen // Determines how much to read on each loop of listen
size_t determineAmountToRead (); size_t determineAmountToRead ();
// Hanlder for listen for once
typedef boost::shared_ptr<boost::condition_variable> shared_cond_var_ptr_t;
void notifyListenForOnce (shared_cond_var_ptr_t cond_ptr);
// Tokenizer // Tokenizer
TokenizerType tokenize; TokenizerType tokenize;
@ -656,40 +774,181 @@ private:
ExceptionCallback handle_exc; ExceptionCallback handle_exc;
// Default handler // Default handler
DataCallback default_handler; FilterPtr default_filter;
DataCallback _default_handler;
ComparatorType default_comparator;
void default_handler(const std::string &token);
// Persistent listening variables // Persistent listening variables
bool listening; bool listening;
serial::Serial * serial_port; serial::Serial * serial_port;
boost::thread listen_thread; boost::thread listen_thread;
std::string data_buffer; std::string data_buffer;
boost::mutex token_mux; size_t chunk_size;
std::map<const uuid_type, std::string> tokens;
std::map<const uuid_type, boost::posix_time::ptime> ttls;
// Callback related variables // Callback related variables
// filter uuid, token uuid // filter id, token
ConcurrentQueue<std::pair<uuid_type,uuid_type> > callback_queue; // filter id == 0 is going to be default handled
ConcurrentQueue<std::pair<FilterPtr,TokenPtr> >
callback_queue;
boost::thread callback_thread; boost::thread callback_thread;
// For generating random uuids
boost::uuids::random_generator random_generator;
boost::uuids::nil_generator nil_generator;
// Setting for ttl on messages
boost::posix_time::time_duration ttl;
// map<uuid, filter type (blocking/non-blocking)>
std::vector<uuid_type> filters;
// map<uuid, comparator>
std::map<const uuid_type, ComparatorType> comparators;
// map<uuid, callback>
std::map<const uuid_type, DataCallback> callbacks;
// Mutex for locking use of filters // Mutex for locking use of filters
boost::mutex filter_mux; boost::mutex filter_mux;
boost::mutex callback_mux; // vector of filter ids
std::vector<FilterPtr> filters;
}; };
/*!
* This is the a filter that provides a wait function for blocking until a
* match is found.
*
* This should probably not be created manually, but instead should be
* constructed using SerialListener::createBlockingFilter(ComparatorType)
* function which returns a BlockingFilter instance.
*
* \see serial::SerialListener::ComparatorType,
* serial::SerialListener::createBlockingFilter
*/
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();
}
/*!
* Waits a given number of milliseconds or until a token is matched. If a
* token is matched it is returned, otherwise an empty string is returned.
*
* \param ms Time in milliseconds to wait on a new token.
*
* \return std::string token that was matched or "" if none were matched.
*/
std::string wait(size_t 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) {
this->cond.notify_all();
this->result = token;
}
private:
SerialListener * listener;
boost::condition_variable cond;
boost::mutex mutex;
std::string result;
};
/*!
* This is the a filter that provides a wait function for blocking until a
* match is found. It will also buffer up to a given buffer size of tokens so
* that they can be counted or accessed after they are matched by the filter.
*
* This should probably not be created manually, but instead should be
* constructed using SerialListener::createBufferedFilter(ComparatorType)
* function which returns a BufferedFilter instance.
*
* The internal buffer is a circular queue buffer, so when the buffer is full,
* the oldest token is dropped and the new one is added. Additionally, when
* wait is a called the oldest available token is returned.
*
* \see serial::SerialListener::ComparatorType,
* serial::SerialListener::createBufferedFilter
*/
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 = "";
}
/*!
* Waits a given number of milliseconds or until a matched token is
* available in the buffer. If a token is matched it is returned, otherwise
* an empty string is returned.
*
* \param ms Time in milliseconds to wait on a new token. If ms is set to 0
* then it will try to get a new token if one is available but will not
* block.
*
* \return std::string token that was matched or "" if none were matched.
*/
std::string wait(size_t 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;
}
/*!
* Clears the buffer of any tokens.
*/
void clear() {
queue.clear();
}
/*!
* Returns the number of tokens waiting in the buffer.
*/
size_t count() {
return queue.size();
}
/*!
* Returns the capacity of the buffer.
*/
size_t capacity() {
return buffer_size;
}
FilterPtr filter_ptr;
void callback(const std::string &token) {
std::string throw_away;
if (this->queue.size() == 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 serial
#endif // SERIAL_LISTENER_H #endif // SERIAL_LISTENER_H

16
serial.cmake
View File

@ -13,7 +13,10 @@ project(Serial)
# Use clang if available # Use clang if available
IF(EXISTS /usr/bin/clang) IF(EXISTS /usr/bin/clang)
set(CMAKE_CXX_COMPILER /usr/bin/clang++) set(CMAKE_CXX_COMPILER /usr/bin/clang++)
set(CMAKE_CXX_FLAGS "-ferror-limit=5 -std=c++0x -stdlib=libc++") set(CMAKE_OSX_DEPLOYMENT_TARGET "")
# set(CMAKE_CXX_FLAGS "-ferror-limit=5 -std=c++0x -stdlib=libc++")
set(CMAKE_CXX_FLAGS "-ferror-limit=5")
set(CMAKE_BUILD_TYPE Debug)
ENDIF(EXISTS /usr/bin/clang) ENDIF(EXISTS /usr/bin/clang)
option(SERIAL_BUILD_TESTS "Build all of the Serial tests." OFF) option(SERIAL_BUILD_TESTS "Build all of the Serial tests." OFF)
@ -39,10 +42,17 @@ ENDIF(NOT DEFINED(LIBRARY_OUTPUT_PATH))
include_directories(${PROJECT_SOURCE_DIR}/include) include_directories(${PROJECT_SOURCE_DIR}/include)
# Add default source files # Add default source files
set(SERIAL_SRCS src/serial.cc src/impl/unix.cc) # src/serial_listener.cc) set(SERIAL_SRCS src/serial.cc src/impl/unix.cc src/serial_listener.cc)
# Add default header files # Add default header files
set(SERIAL_HEADERS include/serial/serial.h include/serial/serial_listener.h) set(SERIAL_HEADERS include/serial/serial.h include/serial/serial_listener.h)
IF(UNIX)
list(APPEND SERIAL_SRCS src/impl/unix.cc)
list(APPEND SERIAL_HEADERS include/serial/impl/unix.h)
ELSE(UNIX)
ENDIF(UNIX)
# Find Boost, if it hasn't already been found # Find Boost, if it hasn't already been found
IF(NOT Boost_FOUND OR NOT Boost_SYSTEM_FOUND OR NOT Boost_FILESYSTEM_FOUND OR NOT Boost_THREAD_FOUND) IF(NOT Boost_FOUND OR NOT Boost_SYSTEM_FOUND OR NOT Boost_FILESYSTEM_FOUND OR NOT Boost_THREAD_FOUND)
find_package(Boost COMPONENTS system filesystem thread REQUIRED) find_package(Boost COMPONENTS system filesystem thread REQUIRED)
@ -99,6 +109,8 @@ IF(SERIAL_BUILD_TESTS)
# Link the Test program to the serial library # Link the Test program to the serial library
target_link_libraries(serial_listener_tests ${GTEST_BOTH_LIBRARIES} target_link_libraries(serial_listener_tests ${GTEST_BOTH_LIBRARIES}
serial) serial)
# # See: http://code.google.com/p/googlemock/issues/detail?id=146
# add_definitions(-DGTEST_USE_OWN_TR1_TUPLE=1)
add_test(AllTestsIntest_serial serial_listener_tests) add_test(AllTestsIntest_serial serial_listener_tests)
ENDIF(SERIAL_BUILD_TESTS) ENDIF(SERIAL_BUILD_TESTS)

51
src/impl/unix.cc
View File

@ -22,6 +22,20 @@
#endif #endif
#endif #endif
class UnhandledException : public std::exception {
const char * e_what;
public:
UnhandledException(const char * e_what) {this->e_what = e_what;}
virtual const char* what() const throw() {
std::stringstream ss;
ss << "Unhandled Exception: " << this->e_what;
return ss.str().c_str();
}
};
typedef UnhandledException e;
using ::serial::Serial; using ::serial::Serial;
using std::string; using std::string;
@ -29,14 +43,11 @@ Serial::SerialImpl::SerialImpl (const string &port, int baudrate,
long timeout, bytesize_t bytesize, long timeout, bytesize_t bytesize,
parity_t parity, stopbits_t stopbits, parity_t parity, stopbits_t stopbits,
flowcontrol_t flowcontrol) flowcontrol_t flowcontrol)
: fd_(-1), isOpen_(false), interCharTimeout_(-1), port_(port), baudrate_(baudrate), : fd_(-1), isOpen_(false), interCharTimeout_(-1), port_(port),
timeout_(timeout), bytesize_(bytesize), parity_(parity), stopbits_(stopbits), baudrate_(baudrate), timeout_(timeout), bytesize_(bytesize),
flowcontrol_(flowcontrol) parity_(parity), stopbits_(stopbits), flowcontrol_(flowcontrol)
{ {
printf("Got port: %s\n", port.c_str()); if (port_.empty() == false) this->open();
if (!port_.empty()) {
this->open();
}
} }
Serial::SerialImpl::~SerialImpl () { Serial::SerialImpl::~SerialImpl () {
@ -47,11 +58,11 @@ void
Serial::SerialImpl::open () { Serial::SerialImpl::open () {
if (port_.empty()) { if (port_.empty()) {
printf("Port was empty\n"); printf("Port was empty\n");
throw "error"; throw e("error");
} }
if (isOpen_ == true) { if (isOpen_ == true) {
printf("Port already opened\n"); printf("Port already opened\n");
throw "error"; throw e("error");
} }
fd_ = ::open (port_.c_str(), O_RDWR | O_NOCTTY | O_NONBLOCK); fd_ = ::open (port_.c_str(), O_RDWR | O_NOCTTY | O_NONBLOCK);
@ -59,7 +70,7 @@ Serial::SerialImpl::open () {
if (fd_ == -1) { if (fd_ == -1) {
printf("Error opening serial port %s - %s(%d).\n", printf("Error opening serial port %s - %s(%d).\n",
port_.c_str(), strerror(errno), errno); port_.c_str(), strerror(errno), errno);
throw "Error"; // Error throw e("error");
} }
reconfigurePort(); reconfigurePort();
@ -69,7 +80,7 @@ Serial::SerialImpl::open () {
void void
Serial::SerialImpl::reconfigurePort () { Serial::SerialImpl::reconfigurePort () {
if (fd_ == -1) { if (fd_ == -1) {
throw "Error"; // Can only operate on a valid file descriptor throw e("Error"); // Can only operate on a valid file descriptor
} }
struct termios options; // The current options for the file descriptor struct termios options; // The current options for the file descriptor
@ -82,7 +93,7 @@ Serial::SerialImpl::reconfigurePort () {
} }
if (tcgetattr(fd_, &originalTTYAttrs) == -1) { if (tcgetattr(fd_, &originalTTYAttrs) == -1) {
throw "Error"; throw e("Error");
} }
options = originalTTYAttrs; options = originalTTYAttrs;
@ -116,7 +127,7 @@ Serial::SerialImpl::reconfigurePort () {
else if (bytesize_ == FIVEBITS) else if (bytesize_ == FIVEBITS)
options.c_cflag |= CS5; options.c_cflag |= CS5;
else else
throw "ValueError(Invalid char len: %%r)"; throw e("ValueError(Invalid char len: %%r)");
// setup stopbits // setup stopbits
if (stopbits_ == STOPBITS_ONE) if (stopbits_ == STOPBITS_ONE)
options.c_cflag &= ~(CSTOPB); options.c_cflag &= ~(CSTOPB);
@ -125,7 +136,7 @@ Serial::SerialImpl::reconfigurePort () {
else if (stopbits_ == STOPBITS_TWO) else if (stopbits_ == STOPBITS_TWO)
options.c_cflag |= (CSTOPB); options.c_cflag |= (CSTOPB);
else else
throw "ValueError(Invalid stop bit specification:)"; throw e("ValueError(Invalid stop bit specification:)");
// setup parity // setup parity
options.c_iflag &= ~(INPCK|ISTRIP); options.c_iflag &= ~(INPCK|ISTRIP);
if (parity_ == PARITY_NONE) { if (parity_ == PARITY_NONE) {
@ -139,7 +150,7 @@ Serial::SerialImpl::reconfigurePort () {
options.c_cflag |= (PARENB|PARODD); options.c_cflag |= (PARENB|PARODD);
} }
else { else {
throw "ValueError(Invalid parity:"; throw e("ValueError(Invalid parity:");
} }
// setup flow control // setup flow control
// xonxoff // xonxoff
@ -206,14 +217,14 @@ Serial::SerialImpl::available () {
return count; return count;
} }
else { else {
throw "Error"; throw e("Error");
} }
} }
string string
Serial::SerialImpl::read (size_t size) { Serial::SerialImpl::read (size_t size) {
if (!isOpen_) { if (!isOpen_) {
throw "PortNotOpenError()"; // throw e("PortNotOpenError()"); //
} }
string message = ""; string message = "";
char buf[1024]; // TODO(ash_gti): Should this be 1024? or...? char buf[1024]; // TODO(ash_gti): Should this be 1024? or...?
@ -243,7 +254,7 @@ Serial::SerialImpl::read (size_t size) {
// Disconnected devices, at least on Linux, show the // Disconnected devices, at least on Linux, show the
// behavior that they are always ready to read immediately // behavior that they are always ready to read immediately
// but reading returns nothing. // but reading returns nothing.
throw "SerialException('device reports readiness to read but returned no data (device disconnected?)')"; throw e("SerialException('device reports readiness to read but returned no data (device disconnected?)')");
} }
message.append(buf, bytes_read); message.append(buf, bytes_read);
} }
@ -257,12 +268,12 @@ Serial::SerialImpl::read (size_t size) {
size_t size_t
Serial::SerialImpl::write (const string &data) { Serial::SerialImpl::write (const string &data) {
if (isOpen_ == false) { if (isOpen_ == false) {
throw "portNotOpenError"; throw e("portNotOpenError");
} }
size_t t = data.length(); size_t t = data.length();
size_t n = ::write(fd_, data.c_str(), data.length()); size_t n = ::write(fd_, data.c_str(), data.length());
if (n == -1) { if (n == -1) {
throw "Write error"; throw e("Write error");
} }
return n; return n;
} }

19
src/serial.cc
View File

@ -16,13 +16,12 @@ using std::vector;
using std::numeric_limits; using std::numeric_limits;
using std::size_t; using std::size_t;
Serial::Serial (const string &port, int baudrate, Serial::Serial (const string &port, int baudrate, long timeout,
long timeout, bytesize_t bytesize, bytesize_t bytesize, parity_t parity, stopbits_t stopbits,
parity_t parity, stopbits_t stopbits,
flowcontrol_t flowcontrol) flowcontrol_t flowcontrol)
: pimpl(new SerialImpl(port,baudrate,timeout,bytesize,parity,stopbits,
flowcontrol))
{ {
pimpl = new SerialImpl(port, baudrate, timeout, bytesize, parity,
stopbits, flowcontrol);
} }
Serial::~Serial () { Serial::~Serial () {
@ -48,11 +47,6 @@ Serial::available () {
return this->pimpl->available(); return this->pimpl->available();
} }
//size_t
//Serial::read (unsigned char* buffer, size_t size) {
// return this->pimpl->read (buffer, size);
//}
string string
Serial::read (size_t size) { Serial::read (size_t size) {
return this->pimpl->read (size); return this->pimpl->read (size);
@ -105,11 +99,6 @@ Serial::readlines(string eol) {
return lines; return lines;
} }
//size_t
//Serial::write (unsigned char* data, size_t length) {
// return this->pimpl->write (data, length);
//}
size_t size_t
Serial::write (const string &data) { Serial::write (const string &data) {
return this->pimpl->write (data); return this->pimpl->write (data);

387
src/serial_listener.cc
View File

@ -20,23 +20,35 @@ inline void defaultExceptionCallback(const std::exception &error) {
throw(error); throw(error);
} }
inline bool defaultComparator(const std::string &token) {
return true;
}
using namespace serial; using namespace serial;
/***** Listener Class Functions *****/ /***** Listener Class Functions *****/
SerialListener::SerialListener() : listening(false) { void
SerialListener::default_handler(const std::string &token) {
if (this->_default_handler)
this->_default_handler(token);
}
SerialListener::SerialListener() : listening(false), chunk_size(5) {
// Set default callbacks // Set default callbacks
this->handle_exc = defaultExceptionCallback; this->handle_exc = defaultExceptionCallback;
this->info = defaultInfoCallback; this->info = defaultInfoCallback;
this->debug = defaultDebugCallback; this->debug = defaultDebugCallback;
this->warn = defaultWarningCallback; this->warn = defaultWarningCallback;
this->default_handler = NULL;
// Default handler stuff
this->_default_handler = NULL;
this->default_comparator = defaultComparator;
DataCallback tmp = boost::bind(&SerialListener::default_handler, this, _1);
this->default_filter = FilterPtr(new Filter(default_comparator, tmp));
// Set default tokenizer // Set default tokenizer
this->setTokenizer(delimeter_tokenizer("\r")); this->setTokenizer(delimeter_tokenizer("\r"));
// Set default ttl
this->setTimeToLive();
} }
SerialListener::~SerialListener() { SerialListener::~SerialListener() {
@ -48,37 +60,19 @@ SerialListener::~SerialListener() {
void void
SerialListener::callback() { SerialListener::callback() {
try { try {
// <filter uuid, token uuid> // <filter id, token>
std::pair<uuid_type,uuid_type> pair; std::pair<FilterPtr,TokenPtr> pair;
DataCallback _callback;
while (this->listening) { while (this->listening) {
if (this->callback_queue.timed_wait_and_pop(pair, 10)) { if (this->callback_queue.timed_wait_and_pop(pair, 10)) {
std::cout << "Got something off the callback queue: ";
std::cout << (*pair.second) << std::endl;
if (this->listening) { if (this->listening) {
try { try {
// If default handler pair.first->callback((*pair.second));
if (pair.first.is_nil()) {
if (this->default_handler)
this->default_handler(this->tokens[pair.second]);
// Else use provided callback
} else {
bool go = false;
// Grab the callback as to not hold the mutex while executing
{
boost::mutex::scoped_lock l(callback_mux);
// Make sure the filter hasn't been removed
if ((go = (this->callbacks.count(pair.first) > 0)))
_callback = this->callbacks[pair.first];
}
// Execute callback
if (go)
_callback(this->tokens[pair.second]);
}
} catch (std::exception &e) { } catch (std::exception &e) {
this->handle_exc(e); this->handle_exc(e);
}// try callback }// try callback
} // if listening } // if listening
// Erase the used and executed callback
this->eraseToken(pair.second);
} // if popped } // if popped
} // while (this->listening) } // while (this->listening)
} catch (std::exception &e) { } catch (std::exception &e) {
@ -86,12 +80,6 @@ SerialListener::callback() {
} }
} }
void
SerialListener::setTimeToLive(size_t ms) {
using namespace boost::posix_time;
this->ttl = time_duration(milliseconds(ms));
}
void void
SerialListener::startListening(Serial &serial_port) { SerialListener::startListening(Serial &serial_port) {
if (this->listening) { if (this->listening) {
@ -121,32 +109,19 @@ SerialListener::stopListening() {
listen_thread.join(); listen_thread.join();
callback_thread.join(); callback_thread.join();
callback_queue.clear();
this->data_buffer = ""; this->data_buffer = "";
this->tokens.clear();
this->ttls.clear();
this->serial_port = NULL; this->serial_port = NULL;
// Delete all the filters // Delete all the filters
this->stopListeningForAll(); this->removeAllFilters();
}
void
SerialListener::stopListeningForAll() {
boost::mutex::scoped_lock l(filter_mux);
filters.clear();
comparators.clear();
boost::mutex::scoped_lock l2(callback_mux);
callbacks.clear();
callback_queue.clear();
} }
size_t size_t
SerialListener::determineAmountToRead() { SerialListener::determineAmountToRead() {
// TODO: Make a more intelligent method based on the length of the things // 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' // filters are looking for. e.g.: if the filter is looking for 'V=XX\r'
// make the read amount at least 5. // make the read amount at least 5.
return 5; return this->chunk_size;
} }
void void
@ -160,134 +135,31 @@ SerialListener::readSomeData(std::string &temp, size_t this_many) {
this->handle_exc(SerialListenerException("Serial port not open.")); this->handle_exc(SerialListenerException("Serial port not open."));
} }
temp = this->serial_port->read(this_many); temp = this->serial_port->read(this_many);
std::cout << "Read(" << temp.length() << "): " << temp << std::endl;
} }
void void
SerialListener::addNewTokens(std::vector<std::string> &new_tokens, SerialListener::filterNewTokens (std::vector<TokenPtr> new_tokens) {
std::vector<uuid_type> &new_uuids,
std::string &left_overs)
{
std::vector<std::string>::iterator it_new;
for (it_new=new_tokens.begin(); it_new != new_tokens.end(); it_new++) {
// The last token needs to be put back in the buffer always:
// (in the case that the delimeter is \r)...
// 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_new == new_tokens.end()-1) {
left_overs = (*it_new);
continue;
}
// If the token is empty ignore it
if ((*it_new).length() == 0)
continue;
// Create a new uuid
uuid_type uuid = random_generator();
// Put the new uuid in the list of new uuids
new_uuids.push_back(uuid);
// Create a uuid, token pair
std::pair<const uuid_type,std::string> token_pair(uuid,(*it_new));
// Create a uuid, ttl pair
using namespace boost::posix_time;
std::pair<const uuid_type,ptime>
ttl_pair(uuid,ptime(microsec_clock::local_time()));
// Insert the new pairs
{
boost::mutex::scoped_lock l(token_mux);
this->tokens.insert(token_pair);
ttls.insert(ttl_pair);
}
} // for (it_new=new_tokens.begin(); it_new != new_tokens.end(); it_new++)
}
void
SerialListener::eraseToken(uuid_type &uuid) {
boost::mutex::scoped_lock l(token_mux);
this->tokens.erase(uuid);
this->ttls.erase(uuid);
}
void
SerialListener::eraseTokens(std::vector<uuid_type> &uuids) {
std::vector<uuid_type>::iterator it;
for (it=uuids.begin(); it != uuids.end(); it++) {
this->eraseToken((*it));
}
}
// TODO: Investigate possible race condition where the filter processing takes
// longer than the ttl
void
SerialListener::filterNewTokens(std::vector<uuid_type> new_uuids) {
// Iterate through the filters, checking each against new tokens // Iterate through the filters, checking each against new tokens
std::vector<std::pair<uuid_type,uuid_type> > tbd; boost::mutex::scoped_lock lock(filter_mux);
boost::mutex::scoped_lock l(filter_mux); std::vector<FilterPtr>::iterator it;
std::vector<uuid_type>::iterator it;
for (it=filters.begin(); it!=filters.end(); it++) { for (it=filters.begin(); it!=filters.end(); it++) {
std::vector<std::pair<uuid_type,uuid_type> > temp = this->filter((*it), new_tokens);
this->filter((*it), new_uuids);
if (temp.size() > 0)
tbd.insert(tbd.end(), temp.begin(), temp.end());
} // for (it=filters.begin(); it!=filters.end(); it++) } // for (it=filters.begin(); it!=filters.end(); it++)
// Dispatch
std::vector<std::pair<uuid_type,uuid_type> >::iterator it_tbd;
for (it_tbd = tbd.begin(); it_tbd != tbd.end(); it_tbd++) {
callback_queue.push((*it_tbd));
}
}
// <filter,token>
std::vector<std::pair<uuid_type,uuid_type> >
SerialListener::filter(uuid_type filter_uuid,
std::vector<uuid_type> &token_uuids)
{
std::vector<uuid_type> to_be_erased;
std::vector<std::pair<uuid_type,uuid_type> > to_be_dispatched;
// Iterate through the token uuids and run each against the filter
std::vector<uuid_type>::iterator it;
for (it=token_uuids.begin(); it!=token_uuids.end(); it++) {
bool matched = false;
uuid_type token_uuid = (*it);
if (this->comparators[filter_uuid](this->tokens[token_uuid])) {
matched = true;
to_be_dispatched.push_back(std::make_pair(filter_uuid,token_uuid));
}
}
return to_be_dispatched;
} }
// <filter_ptr,token_ptr>
void void
SerialListener::pruneTokens() { SerialListener::filter (FilterPtr filter, std::vector<TokenPtr> &tokens)
// Iterate through the buffered tokens
std::vector<uuid_type> to_be_erased;
std::map<const uuid_type,std::string>::iterator it;
{ {
boost::mutex::scoped_lock l(token_mux); // Iterate through the token uuids and run each against the filter
for (it = this->tokens.begin(); it != this->tokens.end(); it++) { std::vector<TokenPtr>::iterator it;
uuid_type uuid = (*it).first; for (it=tokens.begin(); it!=tokens.end(); it++) {
using namespace boost::posix_time; TokenPtr token = (*it);
// If the current time - the creation time is greater than the ttl, if (filter->comparator((*token)))
// then prune it callback_queue.push(std::make_pair(filter,token));
if (ptime(microsec_clock::local_time())-this->ttls[uuid] > this->ttl) {
// If there is a default handler pass it on
if (this->default_handler) {
boost::mutex::scoped_lock l(callback_mux);
callback_queue.push(std::make_pair(nil_generator(),uuid));
} else {
// Otherwise delete it
to_be_erased.push_back(uuid);
} }
} }
}
}
// Remove any lines that need to be erased
// (this must be done outside the iterator to prevent problems incrementing
// the iterator)
this->eraseTokens(to_be_erased);
}
void void
SerialListener::listen() { SerialListener::listen() {
@ -302,17 +174,11 @@ SerialListener::listen() {
// Add the new data to the buffer // Add the new data to the buffer
this->data_buffer += temp; this->data_buffer += temp;
// Call the tokenizer on the updated buffer // Call the tokenizer on the updated buffer
std::vector<std::string> new_tokens; std::vector<TokenPtr> new_tokens;
this->tokenize(this->data_buffer, new_tokens); this->tokenize(this->data_buffer, new_tokens);
// Add the new tokens to the new token buffer, get a list of new uuids
// to filter once, and put left_overs in the data buffer.
std::vector<uuid_type> new_uuids;
this->addNewTokens(new_tokens, new_uuids, this->data_buffer);
// Run the new tokens through existing filters // Run the new tokens through existing filters
this->filterNewTokens(new_uuids); this->filterNewTokens(new_tokens);
} }
// Look for old data and pass to the default handler or delete
this->pruneTokens();
// Done parsing lines and buffer should now be set to the left overs // Done parsing lines and buffer should now be set to the left overs
} // while (this->listening) } // while (this->listening)
} catch (std::exception &e) { } catch (std::exception &e) {
@ -320,154 +186,53 @@ SerialListener::listen() {
} }
} }
bool /***** Filter Functions *****/
SerialListener::listenForStringOnce(std::string token, size_t milliseconds) {
return this->listenForOnce(exactly(token), milliseconds); FilterPtr
SerialListener::createFilter(ComparatorType comparator, DataCallback callback)
{
FilterPtr filter_ptr(new Filter(comparator, callback));
boost::mutex::scoped_lock l(filter_mux);
this->filters.push_back(filter_ptr);
return filter_ptr;
}
BlockingFilterPtr
SerialListener::createBlockingFilter(ComparatorType comparator) {
return BlockingFilterPtr(
new BlockingFilter(comparator, (*this)));
}
BufferedFilterPtr
SerialListener::createBufferedFilter(ComparatorType comparator,
size_t buffer_size)
{
return BufferedFilterPtr(
new BufferedFilter(comparator, buffer_size, (*this)));
} }
void void
SerialListener::notifyListenForOnce(shared_cond_var_ptr_t cond_ptr) { SerialListener::removeFilter(FilterPtr filter_ptr) {
cond_ptr->notify_all();
}
bool
SerialListener::listenForOnce(ComparatorType comparator, size_t ms)
{
shared_cond_var_ptr_t cond_ptr(new boost::condition_variable());
boost::mutex mut;
// Create blocking filter
const uuid_type uuid = random_generator();
{
boost::mutex::scoped_lock l(filter_mux); boost::mutex::scoped_lock l(filter_mux);
filters.push_back(uuid); filters.erase(std::find(filters.begin(),filters.end(),filter_ptr));
comparators.insert(std::make_pair(uuid,comparator));
}
{
boost::mutex::scoped_lock l(callback_mux);
callbacks.insert(std::make_pair(uuid,
boost::bind(&SerialListener::notifyListenForOnce, this, cond_ptr)));
}
// Run this filter through all tokens onces
std::vector<uuid_type> token_uuids;
std::map<const uuid_type,std::string>::iterator it;
for (it = tokens.begin(); it != tokens.end(); it++)
token_uuids.push_back((*it).first);
std::vector<std::pair<uuid_type,uuid_type> > pairs =
this->filter(uuid, token_uuids);
// If there is at least one
if (pairs.size() > 0) {
// If there is more than one find the oldest
size_t index = 0;
if (pairs.size() > 1) {
using namespace boost::posix_time;
ptime oldest_time = ttls[pairs[index].second];
size_t i = 0;
std::vector<std::pair<uuid_type,uuid_type> >::iterator it;
for (it = pairs.begin(); it != pairs.end(); it++) {
if (ttls[(*it).second] < oldest_time) {
oldest_time = ttls[(*it).second];
index = i;
}
i++;
}
}
// Either way put the final index into the callback queue
callback_queue.push(pairs[index]);
}
bool result = false;
// Wait
boost::unique_lock<boost::mutex> lock(mut);
using namespace boost::posix_time;
if (cond_ptr->timed_wait(lock, milliseconds(ms)))
result = true;
// Destroy the filter
{
boost::mutex::scoped_lock l(filter_mux);
filters.erase(std::find(filters.begin(),filters.end(),uuid));
comparators.erase(uuid);
}
{
boost::mutex::scoped_lock l(callback_mux);
callbacks.erase(uuid);
}
return result;
}
uuid_type
SerialListener::listenFor(ComparatorType comparator, DataCallback callback)
{
// Create Filter
uuid_type uuid = random_generator();
std::pair<const uuid_type,ComparatorType>
comparator_pair(uuid, comparator);
std::pair<const uuid_type,DataCallback>
callback_pair(uuid, callback);
{
boost::mutex::scoped_lock l(filter_mux);
filters.push_back(uuid);
comparators.insert(comparator_pair);
}
{
boost::mutex::scoped_lock l(callback_mux);
callbacks.insert(callback_pair);
}
// Run this filter through all tokens onces
std::vector<uuid_type> token_uuids;
std::map<const uuid_type,std::string>::iterator it;
for (it = tokens.begin(); it != tokens.end(); it++)
token_uuids.push_back((*it).first);
std::vector<std::pair<uuid_type,uuid_type> > pairs =
this->filter(uuid, token_uuids);
// Dispatch
std::vector<std::pair<uuid_type,uuid_type> >::iterator it_cb;
for (it_cb = pairs.begin(); it_cb != pairs.end(); it_cb++) {
callback_queue.push((*it_cb));
}
return uuid;
} }
void void
SerialListener::stopListeningFor(uuid_type filter_uuid) { SerialListener::removeFilter(BlockingFilterPtr blocking_filter) {
// Delete filter this->removeFilter(blocking_filter->filter_ptr);
boost::mutex::scoped_lock l(filter_mux);
filters.erase(std::find(filters.begin(),filters.end(),filter_uuid));
comparators.erase(filter_uuid);
boost::mutex::scoped_lock l2(callback_mux);
callbacks.erase(filter_uuid);
}
TokenizerType
SerialListener::delimeter_tokenizer (std::string delimeter) {
return boost::bind(&SerialListener::_delimeter_tokenizer,
_1, _2, delimeter);
} }
void void
SerialListener::_delimeter_tokenizer (std::string &data, SerialListener::removeFilter(BufferedFilterPtr buffered_filter) {
std::vector<std::string> &tokens, this->removeFilter(buffered_filter->filter_ptr);
std::string delimeter)
{
boost::split(tokens, data, boost::is_any_of(delimeter));
} }
ComparatorType void
SerialListener::exactly(std::string exact_str) { SerialListener::removeAllFilters() {
return boost::bind(&SerialListener::_exactly, _1, exact_str); boost::mutex::scoped_lock l(filter_mux);
} filters.clear();
callback_queue.clear();
bool
SerialListener::_exactly(const std::string &token, std::string exact_str) {
return token == exact_str;
} }

255
tests/serial_listener_tests.cc
View File

@ -18,136 +18,131 @@ void default_handler(std::string line) {
namespace { namespace {
class SerialListenerTests : public ::testing::Test { // class SerialListenerTests : public ::testing::Test {
protected: // protected:
virtual void SetUp() { // virtual void SetUp() {
listener.listening = true; // listener.listening = true;
listener.setTimeToLive(10); // listener.setDefaultHandler(default_handler);
listener.default_handler = default_handler; // listener.callback_thread =
listener.callback_thread = // boost::thread(boost::bind(&SerialListener::callback, &listener));
boost::thread(boost::bind(&SerialListener::callback, &listener)); // }
} //
// virtual void TearDown() {
virtual void TearDown() { // listener.listening = false;
listener.listening = false; // listener.callback_thread.join();
listener.callback_thread.join(); // }
} //
// void stopCallbackThread() {
void stopCallbackThread() { // while (true) {
while (true) { // boost::this_thread::sleep(boost::posix_time::milliseconds(1));
boost::this_thread::sleep(boost::posix_time::milliseconds(1)); // boost::mutex::scoped_lock lock(listener.callback_queue.the_mutex);
boost::mutex::scoped_lock lock(listener.callback_queue.the_mutex); // if (listener.callback_queue.the_queue.empty())
if (listener.callback_queue.the_queue.empty()) // break;
break; // }
} // listener.listening = false;
listener.listening = false; // listener.callback_thread.join();
listener.callback_thread.join(); // }
} //
// void execute_listenForStringOnce() {
void execute_listenForStringOnce() { // listener.listenForStringOnce("?$1E", 50);
listener.listenForStringOnce("?$1E", 50); // }
} //
// void simulate_loop(std::string input_str) {
void simulate_loop(std::string input_str) { // std::vector<TokenPtr> new_tokens;
std::vector<std::string> new_tokens; // listener.tokenize(input_str, new_tokens);
listener.tokenize(input_str, new_tokens); // listener.filterNewTokens(new_tokens);
std::vector<uuid_type> new_uuids; // }
listener.addNewTokens(new_tokens, new_uuids, listener.data_buffer); //
listener.filterNewTokens(new_uuids); // SerialListener listener;
boost::this_thread::sleep(boost::posix_time::milliseconds(11)); //
listener.pruneTokens(); // };
} //
// TEST_F(SerialListenerTests, handlesPartialMessage) {
SerialListener listener; // global_count = 0;
// std::string input_str = "?$1E\r$1E=Robo";
}; //
// simulate_loop(input_str);
TEST_F(SerialListenerTests, handlesPartialMessage) { //
global_count = 0; // // give some time for the callback thread to finish
std::string input_str = "?$1E\r$1E=Robo"; // stopCallbackThread();
//
simulate_loop(input_str); // ASSERT_EQ(global_count, 1);
// }
// give some time for the callback thread to finish //
stopCallbackThread(); // TEST_F(SerialListenerTests, listenForOnceWorks) {
// global_count = 0;
ASSERT_EQ(global_count, 1); //
} // boost::thread t(
// boost::bind(&SerialListenerTests::execute_listenForStringOnce, this));
TEST_F(SerialListenerTests, listenForOnceWorks) { //
global_count = 0; // boost::this_thread::sleep(boost::posix_time::milliseconds(5));
//
boost::thread t( // simulate_loop("\r+\r?$1E\r$1E=Robo");
boost::bind(&SerialListenerTests::execute_listenForStringOnce, this)); //
// ASSERT_TRUE(t.timed_join(boost::posix_time::milliseconds(60)));
boost::this_thread::sleep(boost::posix_time::milliseconds(5)); //
// // Make sure the filters are getting deleted
simulate_loop("\r+\r?$1E\r$1E=Robo"); // ASSERT_EQ(listener.filters.size(), 0);
//
ASSERT_TRUE(t.timed_join(boost::posix_time::milliseconds(60))); // // give some time for the callback thread to finish
// stopCallbackThread();
// Make sure the filters are getting deleted //
ASSERT_EQ(listener.filters.size(), 0); // ASSERT_EQ(global_count, 1);
// }
// give some time for the callback thread to finish //
stopCallbackThread(); // // lookForOnce should not find it, but timeout after 1000ms, so it should
// // still join.
ASSERT_EQ(global_count, 1); // TEST_F(SerialListenerTests, listenForOnceTimesout) {
} // global_count = 0;
//
// lookForOnce should not find it, but timeout after 1000ms, so it should // boost::thread t(
// still join. // boost::bind(&SerialListenerTests::execute_listenForStringOnce, this));
TEST_F(SerialListenerTests, listenForOnceTimesout) { //
global_count = 0; // boost::this_thread::sleep(boost::posix_time::milliseconds(55));
//
boost::thread t( // simulate_loop("\r+\r?$1ENOTRIGHT\r$1E=Robo");
boost::bind(&SerialListenerTests::execute_listenForStringOnce, this)); //
// ASSERT_TRUE(t.timed_join(boost::posix_time::milliseconds(60)));
boost::this_thread::sleep(boost::posix_time::milliseconds(55)); //
// // give some time for the callback thread to finish
simulate_loop("\r+\r?$1ENOTRIGHT\r$1E=Robo"); // stopCallbackThread();
//
ASSERT_TRUE(t.timed_join(boost::posix_time::milliseconds(60))); // ASSERT_EQ(global_count, 2);
// }
// give some time for the callback thread to finish //
stopCallbackThread(); // bool listenForComparator(std::string line) {
// // std::cout << "In listenForComparator(" << line << ")" << std::endl;
ASSERT_EQ(global_count, 2); // if (line.substr(0,2) == "V=") {
} // return true;
// }
bool listenForComparator(std::string line) { // return false;
// std::cout << "In listenForComparator(" << line << ")" << std::endl; // }
if (line.substr(0,2) == "V=") { //
return true; // void listenForCallback(std::string line) {
} // // std::cout << "In listenForCallback(" << line << ")" << std::endl;
return false; // global_listen_count++;
} // }
//
void listenForCallback(std::string line) { // TEST_F(SerialListenerTests, listenForWorks) {
// std::cout << "In listenForCallback(" << line << ")" << std::endl; // global_count = 0;
global_listen_count++; // global_listen_count = 0;
} //
// FilterPtr filt_uuid =
TEST_F(SerialListenerTests, listenForWorks) { // listener.listenFor(listenForComparator, listenForCallback);
global_count = 0; //
global_listen_count = 0; // simulate_loop("\r+\rV=05:06\r?$1E\rV=06:05\r$1E=Robo");
//
boost::uuids::uuid filt_uuid = // // give some time for the callback thread to finish
listener.listenFor(listenForComparator, listenForCallback); // stopCallbackThread();
//
simulate_loop("\r+\rV=05:06\r?$1E\rV=06:05\r$1E=Robo"); // ASSERT_EQ(global_count, 2);
// ASSERT_EQ(global_listen_count, 2);
// give some time for the callback thread to finish //
stopCallbackThread(); // listener.stopListeningFor(filt_uuid);
//
ASSERT_EQ(global_count, 2); // ASSERT_EQ(listener.filters.size(), 0);
ASSERT_EQ(global_listen_count, 2); //
// }
listener.stopListeningFor(filt_uuid);
ASSERT_EQ(listener.filters.size(), 0);
}
} // namespace } // namespace