AsyncSerial.cpp

/*
 * File:   AsyncSerial.cpp
 * Author: Terraneo Federico
 * Distributed under the Boost Software License, Version 1.0.
 * Created on September 7, 2009, 10:46 AM
 *
 * v1.02: Fixed a bug in BufferedAsyncSerial: Using the default constructor
 * the callback was not set up and reading didn't work.
 *
 * v1.01: Fixed a bug that did not allow to reopen a closed serial port.
 *
 * v1.00: First release.
 *
 * IMPORTANT:
 * On Mac OS X boost asio's serial ports have bugs, and the usual implementation
 * of this class does not work. So a workaround class was written temporarily,
 * until asio (hopefully) will fix Mac compatibility for serial ports.
 *
 * Please note that unlike said in the documentation on OS X until asio will
 * be fixed serial port *writes* are *not* asynchronous, but at least
 * asynchronous *read* works.
 * In addition the serial port open ignores the following options: parity,
 * character size, flow, stop bits, and defaults to 8N1 format.
 * I know it is bad but at least it's better than nothing.
 *
 */
 
#include "AsyncSerial.h"
 
#include <string>
#include <algorithm>
#include <iostream>
#include <boost/bind/bind.hpp>
 
using namespace std;
 
//
//Class AsyncSerial
//
class AsyncSerialImpl : private boost::noncopyable
{
public:
  AsyncSerialImpl() : io(), port(io), backgroundThread(), open(false),
    error(false), quit(false) {}
 
  boost::asio::io_service io; 
  boost::asio::serial_port port; 
  boost::thread backgroundThread; 
  bool open; 
  bool error; 
  bool quit; 
 
  mutable boost::mutex errorMutex; 
  mutable boost::mutex quitMutex; 
 
  std::vector<char> writeQueue;
  boost::shared_array<char> writeBuffer; 
  size_t writeBufferSize; 
  boost::mutex writeQueueMutex; 
  char readBuffer[AsyncSerial::readBufferSize]; 
 
  boost::function<void(const char*, size_t)> callback;
};
 
AsyncSerial::AsyncSerial() : pimpl(new AsyncSerialImpl)
{
 
}
 
bool AsyncSerial::isPortAvailable(std::string portName)
{
  try
  {
    boost::asio::io_service service;
    boost::asio::serial_port sp(service, portName);
 
    if (sp.is_open())
    {
      sp.close();
      return true;
    }
  }
  catch (std::exception&)
  {
    return false;
  }
 
  return false;
}
 
AsyncSerial::AsyncSerial(const std::string& devname, unsigned int baud_rate,
  boost::asio::serial_port_base::parity opt_parity,
  boost::asio::serial_port_base::character_size opt_csize,
  boost::asio::serial_port_base::flow_control opt_flow,
  boost::asio::serial_port_base::stop_bits opt_stop)
  : pimpl(new AsyncSerialImpl)
{
  mDevname = devname;
  mBaudrate = boost::asio::serial_port_base::baud_rate(baud_rate);
  mParity = opt_parity;
  mCsize = opt_csize;
  mFlow = opt_flow;
  mStopbit = opt_stop;
 
  pimpl->backgroundThread = boost::thread(boost::bind(&boost::asio::io_service::run, &pimpl->io));
  pimpl->io.post(boost::bind(&AsyncSerial::retry, this));
}
 
void AsyncSerial::open(const std::string& devname, unsigned int baud_rate,
  boost::asio::serial_port_base::parity opt_parity,
  boost::asio::serial_port_base::character_size opt_csize,
  boost::asio::serial_port_base::flow_control opt_flow,
  boost::asio::serial_port_base::stop_bits opt_stop)
{
  if (isOpen())
    doClose();
 
  setErrorStatus(true);//If an exception is thrown, error_ remains true
  pimpl->port.open(devname);
  pimpl->port.set_option(mBaudrate);
  pimpl->port.set_option(mParity);
  pimpl->port.set_option(mCsize);
  pimpl->port.set_option(mFlow);
  pimpl->port.set_option(mStopbit);
 
  //This gives some work to the io_service before it is started
  pimpl->io.post(boost::bind(&AsyncSerial::doRead, this));
 
  setErrorStatus(false);//If we get here, no error
  pimpl->open = true; //Port is now open
}
 
void AsyncSerial::retry()
{
  if (isOpen())
    return;
 
  try
  {
    //look for port name
    if (mDevname.length() <= 3)
    {
#ifdef _WIN32
      mDevname = "COM" + mDevname;
#else
      std::string testport = "/dev/ttyS" + mDevname;
      if (isPortAvailable(testport))
      {
        mDevname = testport;
      }
      else
      {
        testport = "/dev/ttyUSB" + mDevname;
        if (isPortAvailable(testport))
        {
          mDevname = testport;
        }
        else
        {
          testport = "/dev/ttyACM" + mDevname;
          if (isPortAvailable(testport))
          {
            mDevname = testport;
          }
          else
          {
            testport = "/dev/rfcomm" + mDevname;
            if (isPortAvailable(testport))
            {
              mDevname = testport;
            }
          }
        }
      }
#endif
    }
 
    setErrorStatus(true);//If an exception is thrown, error_ remains true
    pimpl->port.open(mDevname);
    pimpl->port.set_option(boost::asio::serial_port_base::baud_rate(mBaudrate));
    pimpl->port.set_option(mParity);
    pimpl->port.set_option(mCsize);
    pimpl->port.set_option(mFlow);
    pimpl->port.set_option(mStopbit);
    pimpl->writeQueue.clear();
    pimpl->writeBuffer.reset();
    pimpl->writeBufferSize = 0;
 
    //This gives some work to the io_service before it is started
    pimpl->io.post(boost::bind(&AsyncSerial::doRead, this));
 
    setErrorStatus(false);//If we get here, no error
    pimpl->open = true; //Port is now open
  }
  catch (std::exception&)
  {
    if (!quitStatus())
      pimpl->io.post(boost::bind(&AsyncSerial::retry, this));
  }
}
 
bool AsyncSerial::isOpen() const
{
  return pimpl->open;
}
 
bool AsyncSerial::errorStatus() const
{
  boost::lock_guard<boost::mutex> l(pimpl->errorMutex);
  return pimpl->error;
}
 
void AsyncSerial::setQuitStatus(bool e)
{
  boost::lock_guard<boost::mutex> l(pimpl->quitMutex);
  pimpl->quit = e;
}
 
bool AsyncSerial::quitStatus() const
{
  boost::lock_guard<boost::mutex> l(pimpl->quitMutex);
  return pimpl->quit;
}
 
void AsyncSerial::close()
{
  setQuitStatus(true);
  pimpl->io.stop();
  pimpl->backgroundThread.join();
 
  if (isOpen())
  {
    pimpl->open = false;
    doClose();
 
    if (errorStatus())
    {
      throw(boost::system::system_error(boost::system::error_code(),
        "Error while closing the device"));
    }
  }
}
 
void AsyncSerial::write(const char *data, size_t size)
{
  if (isOpen())
  {
    {
      boost::lock_guard<boost::mutex> l(pimpl->writeQueueMutex);
      pimpl->writeQueue.insert(pimpl->writeQueue.end(), data, data + size);
    }
    pimpl->io.post(boost::bind(&AsyncSerial::doWrite, this));
  }
}
 
void AsyncSerial::write(const std::vector<char>& data)
{
  if (isOpen())
  {
    {
      boost::lock_guard<boost::mutex> l(pimpl->writeQueueMutex);
      pimpl->writeQueue.insert(pimpl->writeQueue.end(), data.begin(),
        data.end());
    }
    pimpl->io.post(boost::bind(&AsyncSerial::doWrite, this));
  }
}
 
void AsyncSerial::writeString(const std::string& s)
{
  if (isOpen())
  {
    {
      boost::lock_guard<boost::mutex> l(pimpl->writeQueueMutex);
      pimpl->writeQueue.insert(pimpl->writeQueue.end(), s.begin(), s.end());
    }
    pimpl->io.post(boost::bind(&AsyncSerial::doWrite, this));
  }
}
 
AsyncSerial::~AsyncSerial()
{
  try
  {
    close();
  }
  catch (...)
  {
    //Don't throw from a destructor
  }
}
 
void AsyncSerial::doRead()
{
  pimpl->port.async_read_some(boost::asio::buffer(pimpl->readBuffer, readBufferSize),
    boost::bind(&AsyncSerial::readEnd,
    this,
    boost::asio::placeholders::error,
    boost::asio::placeholders::bytes_transferred));
}
 
void AsyncSerial::readEnd(const boost::system::error_code& error,
  size_t bytes_transferred)
{
  if (error)
  {
    //error can be true even because the serial port was closed.
    //In this case it is not a real error, so ignore
    if (isOpen())
    {
      doClose();
      setErrorStatus(true);
    }
  }
  else
  {
    if (pimpl->callback) pimpl->callback(pimpl->readBuffer, bytes_transferred);
    doRead();
  }
}
 
void AsyncSerial::doWrite()
{
  //If a write operation is already in progress, do nothing
  if (pimpl->writeBuffer == 0)
  {
    boost::lock_guard<boost::mutex> l(pimpl->writeQueueMutex);
    pimpl->writeBufferSize = pimpl->writeQueue.size();
    pimpl->writeBuffer.reset(new char[pimpl->writeQueue.size()]);
    copy(pimpl->writeQueue.begin(), pimpl->writeQueue.end(), pimpl->writeBuffer.get());
    pimpl->writeQueue.clear();
    async_write(pimpl->port, boost::asio::buffer(pimpl->writeBuffer.get(), pimpl->writeBufferSize), boost::bind(&AsyncSerial::writeEnd, this, boost::asio::placeholders::error));
  }
}
 
void AsyncSerial::writeEnd(const boost::system::error_code& error)
{
  if (!error)
  {
    boost::lock_guard<boost::mutex> l(pimpl->writeQueueMutex);
    if (pimpl->writeQueue.empty())
    {
      pimpl->writeBuffer.reset();
      pimpl->writeBufferSize = 0;
 
      return;
    }
    pimpl->writeBufferSize = pimpl->writeQueue.size();
    pimpl->writeBuffer.reset(new char[pimpl->writeQueue.size()]);
    copy(pimpl->writeQueue.begin(), pimpl->writeQueue.end(), pimpl->writeBuffer.get());
    pimpl->writeQueue.clear();
    async_write(pimpl->port, boost::asio::buffer(pimpl->writeBuffer.get(), pimpl->writeBufferSize), boost::bind(&AsyncSerial::writeEnd, this, boost::asio::placeholders::error));
  }
  else
  {
    setErrorStatus(true);
    doClose();
  }
}
 
void AsyncSerial::doClose()
{
  boost::system::error_code ec;
  pimpl->port.cancel(ec);
 
  if (ec)
    setErrorStatus(true);
 
  pimpl->port.close(ec);
  pimpl->open = false;
  pimpl->writeQueue.clear();
 
  if (ec)
    setErrorStatus(true);
 
  if (!quitStatus())
  {
    retry();
  }
}
 
void AsyncSerial::setErrorStatus(bool e)
{
  boost::lock_guard<boost::mutex> l(pimpl->errorMutex);
  pimpl->error = e;
}
 
void AsyncSerial::setReadCallback(const boost::function<void(const char*, size_t)>& callback)
{
  pimpl->callback = callback;
}
 
void AsyncSerial::clearReadCallback()
{
  pimpl->callback.clear();
}
 
//
//Class CallbackAsyncSerial
//
 
CallbackAsyncSerial::CallbackAsyncSerial() : AsyncSerial()
{
 
}
 
CallbackAsyncSerial::CallbackAsyncSerial(const std::string& devname,
  unsigned int baud_rate,
  boost::asio::serial_port_base::parity opt_parity,
  boost::asio::serial_port_base::character_size opt_csize,
  boost::asio::serial_port_base::flow_control opt_flow,
  boost::asio::serial_port_base::stop_bits opt_stop)
  : AsyncSerial(devname, baud_rate, opt_parity, opt_csize, opt_flow, opt_stop)
{
 
}
 
void CallbackAsyncSerial::setCallback(const
  boost::function<void(const char*, size_t)>& callback)
{
  setReadCallback(callback);
}
 
void CallbackAsyncSerial::clearCallback()
{
  clearReadCallback();
}
 
CallbackAsyncSerial::~CallbackAsyncSerial()
{
  clearReadCallback();
}