connection.cc

/* vim: set et sw=3 tw=0 fo=croqlaw cino=t0:
 * 
 * Astxx, the Asterisk C++ API and Utility Library.
 * Copyright (C) 2005-2007  Matthew A. Nicholson
 * Copyright (C) 2005-2007  Digium, Inc.
 * 
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License version 2.1 as published by the Free Software Foundation.
 * 
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */

#include <astxx/manager/connection.h>
#include <astxx/manager/message.h>
#include <astxx/manager/error.h>
#include <boost/asio.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/bind.hpp>
#include <boost/ref.hpp>

namespace astxx {
   namespace manager {
      using boost::asio::ip::tcp;
      using boost::lexical_cast;

      class response_waiter {
         public:
            response_waiter(manager::connection& connection)
               : connection(connection), response("") {
            }

            void operator()(message::response r) {
               response = r;
            }

            message::response wait() const {
               while (response == "") {
                  connection.wait_response();
                  connection.process_responses();
               }
               return response;
            }
         private:
            manager::connection& connection;
            message::response response;
      };

      connection::connection(const std::string& host, unsigned short port) : socket(io_service) {
         connect(host, port);
      }

      void connection::connect(const std::string& host, unsigned short port) {
         // update the internal host and port if they are not the default 
         // values
         if (not host.empty()) {
            m_host = host;
            if (port) {
               m_port = lexical_cast<std::string>(port);
            }
         }

         tcp::resolver resolver(io_service);
         tcp::resolver::query query(m_host, lexical_cast<std::string>(m_port));

         tcp::resolver::iterator endpoint_iterator = resolver.resolve(query);
         tcp::resolver::iterator end;

         // here we loop through all possible endpoints trying to connect to 
         // each as long as we failed to connect to the previous one
         boost::system::error_code error = boost::asio::error::host_not_found;
         while (error && endpoint_iterator != end) {
            socket.close();
            socket.connect(*endpoint_iterator++, error);
         }

         if (error)
            throw boost::system::system_error(error);

         std::string greeting_line = read_line();
         
         // split the version and the name
         std::string::size_type i = greeting_line.find_last_of('/');
         if (i != std::string::npos) {
            m_name.assign(greeting_line, 0, i);
            if (++i != std::string::npos) {
               m_version.assign(greeting_line, i, std::string::npos);
            }
         }
      }

      void connection::disconnect() {
         socket.shutdown(boost::asio::ip::tcp::socket::shutdown_both);
         socket.close();
      }

      bool connection::is_connected() const {
         return socket.is_open();
      }

      message::response connection::send_action(const basic_action& command) {
         response_waiter rw(*this);
         send_action_async(command, boost::ref(rw));
         return rw.wait();
      }

      void connection::send_action_async(const basic_action& command, response_handler_t handler) {

         message::action action = command.action();
         if (not command.action_id().empty()) {
            action["ActionID"] = command.action_id();
         }

         boost::asio::write(socket, boost::asio::buffer(action.format()));
         response_handlers.push(handler);
      }
      
      message::response connection::operator()(const basic_action& command) {
         return send_action(command);
      }
      
      void connection::operator()(const basic_action& command, response_handler_t handler) {
         send_action_async(command, handler);
      }
      
      std::pair<std::string, std::string> connection::parse_header(const std::string& header) {
         std::string key;
         std::string value;

         if (header.empty()) {
            throw manager::empty_header();
         }

         std::string::size_type i = header.find_first_of(':');
         if (i == std::string::npos) {
            throw manager::parse_error("missing ':' in header: " + header);
         }

         key.assign(header, 0, i++);
         
         if (i != std::string::npos) {
            if (header[i] == ' ') {
               ++i;
            }
            if (i != std::string::npos) {
               value.assign(header, i, std::string::npos);
            }
         }
         return std::make_pair(key, value);
      }

      std::string connection::read_line() {
         std::string line;

         // loop reading one character at a time until we find a '\n'
         for (char c = 0, cc = 0; !(c == '\n' && cc == '\r'); line += c) {
            cc = c;
            boost::asio::read(socket, boost::asio::buffer(&c, 1), boost::asio::transfer_at_least(1));
         }

         // strip the \r\n
         std::string::size_type rn = line.find("\r\n");
         if (rn != std::string::npos) {
            line.erase(rn);
         }

         return line;
      }

      void connection::read_message() {
         std::pair<std::string, std::string> pair = parse_header(read_line());
         if (pair.first == "Event") {
            message::event event(pair.second);
            for (;;) {
               std::string line = read_line();

               // stop if we get a blank line
               if (line.empty()) {
                  break;
               }
               event.insert(parse_header(line));
            }
            // put the event in the queue
            events.push(event);
         }
         else if (pair.first == "Response") {
            message::response response(pair.second);
            for (;;) {
               std::string line = read_line();

               // stop if we get a blank line
               if (line.empty()) {
                  break;
               }

               // check for '--END COMMAND--' if necessary
               if (pair.second == "Follows") {
                  std::string::size_type ec = line.find("--END COMMAND--");

                  // strip the '--END COMMAND--', if found, otherwise this is a 
                  // normal header
                  if (ec != std::string::npos) {
                     line.erase(ec);
                     response.data = line;
                  }
                  else {
                     response.insert(parse_header(line));
                  }

               } else {
                  response.insert(parse_header(line));
               }
            }
            // put the resposne in the queue
            responses.push(response);
         }
         else {
            throw manager::unknown_message(pair.first);
         }
      }
      
      message::response connection::read_response() {
         wait_response();
         message::response response = responses.front();
         responses.pop();
         return response;
      }
      
      message::event connection::read_event() {
         wait_event();
         message::event event = events.front();
         events.pop();
         return event;
      }

      void connection::process_events() {
         /* Here we pop events off of the queue one by one because it is 
          * possible for our handlers to add events to the queue (by executing 
          * an action which adds events while waiting for a response).  By 
          * popping events off one by one, we avoid worring about our iterators 
          * becomming invalid.
          */
         while (not events.empty()) {
            message::event e = events.front();
            events.pop();

            event_handlers_t::iterator i = event_handlers.find(e.main_header());
            if (i != event_handlers.end()) {
               (*i->second)(e);
            }

            // execute the catch all
            i = event_handlers.find("");
            if (i != event_handlers.end()) {
               (*i->second)(e);
            }
         }
      }

      void connection::wait_event() {
         while (events.empty()) {
            read_message();
         }
      }

      void connection::process_responses() {
         /* Here we pop responses off of the queue one by one because it is 
          * possible for our handlers to add responses to the queue (by 
          * executing an action synchronously).  By popping events off one by 
          * one, we avoid worring about our iterators becomming invalid.
          */
         while (not response_handlers.empty() and not responses.empty()) {
            response_handler_t f = response_handlers.front();
            response_handlers.pop();

            f(read_response());
         }
      }

      void connection::wait_response() {
         while (responses.empty()) {
            read_message();
         }
      }
      
      void connection::pump_messages() {
         while (socket.available()) {
            read_message();
         }
      }

      boost::signals::connection connection::register_event(const std::string& e, boost::function<void (message::event)> f) {
         std::pair<event_handlers_t::iterator, bool> ii = event_handlers.insert(std::make_pair(e, boost::shared_ptr<boost::signal<void (message::event)> >(new boost::signal<void (message::event)>())));
         return ii.first->second->connect(f);
      }

   }
}