server/supernova/utilities/osc_server.hpp

   1 //  osc responder class
   2 //  Copyright (C) 2008 Tim Blechmann
   3 //
   4 //  This program is free software; you can redistribute it and/or modify
   5 //  it under the terms of the GNU General Public License as published by
   6 //  the Free Software Foundation; either version 2 of the License, or
   7 //  (at your option) any later version.
   8 //
   9 //  This program is distributed in the hope that it will be useful,
  10 //  but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 //  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 //  GNU General Public License for more details.
  13 //
  14 //  You should have received a copy of the GNU General Public License
  15 //  along with this program; see the file COPYING.  If not, write to
  16 //  the Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  17 //  Boston, MA 02111-1307, USA.
  18
  19 #ifndef OSC_SERVER_HPP
  20 #define OSC_SERVER_HPP
  21
  22 #include <string>
  23 #include <iostream>
  24
  25 #include <boost/asio.hpp>
  26 #include <boost/thread.hpp>
  27 #include <boost/bind.hpp>
  28
  29 #include "branch_hints.hpp"
  30 #include "osc_dispatcher.hpp"
  31
  32 #include "static_allocator.hpp"
  33 #include "nova-tt/semaphore.hpp"
  34 #include "nova-tt/thread_priority.hpp"
  35 #include "nova-tt/name_thread.hpp"
  36
  37 namespace nova
  38 {
  39
  40 namespace detail
  41 {
  42 using namespace boost::asio;
  43 using namespace boost::asio::ip;
  44
  45 class network_thread
  46 {
  47 public:
  48     void start_receive(void)
  49     {
  50         thread_ = boost::thread(network_thread::start, this);
  51         sem.wait();
  52     }
  53
  54     ~network_thread(void)
  55     {
  56         if (!thread_.joinable())
  57             return;
  58         io_service_.stop();
  59         thread_.join();
  60     }
  61
  62     io_service & get_io_service(void)
  63     {
  64         return io_service_;
  65     }
  66
  67     void send_udp(const char * data, unsigned int size, udp::endpoint const & receiver)
  68     {
  69         udp::socket socket(io_service_);
  70         socket.open(udp::v4());
  71         socket.send_to(boost::asio::buffer(data, size), receiver);
  72     }
  73
  74 private:
  75     static void start(network_thread * self)
  76     {
  77 #ifdef NOVA_TT_PRIORITY_RT
  78         thread_set_priority_rt(thread_priority_interval_rt().first);
  79 #endif
  80         name_thread("Network Receive");
  81         self->run();
  82     }
  83
  84     void run(void)
  85     {
  86         sem.post();
  87
  88         io_service::work work(io_service_);
  89         io_service_.run();
  90     }
  91
  92 protected:
  93     io_service io_service_;
  94
  95 private:
  96     semaphore sem;
  97     boost::thread thread_;
  98 };
  99
 100 class osc_server:
 101     private network_thread,
 102     private osc_dispatcher
 103 {
 104 public:
 105     osc_server(unsigned int port):
 106         socket_(network_thread::io_service_, udp::endpoint(udp::v4(), port))
 107     {
 108         network_thread::start_receive();
 109         start_receive();
 110     }
 111
 112     void add_responder(std::string const & address, osc_responder * resp)
 113     {
 114         osc_dispatcher::add_responder(address, resp);
 115     }
 116
 117     void remove_responder(std::string const & address, osc_responder * resp)
 118     {
 119         osc_dispatcher::remove_responder(address, resp);
 120     }
 121
 122 private:
 123     void start_receive(void)
 124     {
 125         socket_.async_receive_from(
 126             buffer(recv_buffer_), remote_endpoint_,
 127             boost::bind(&osc_server::handle_receive, this,
 128                         placeholders::error, placeholders::bytes_transferred));
 129     }
 130
 131     void handle_receive(const boost::system::error_code& error,
 132                         std::size_t bytes_transferred)
 133     {
 134         if (unlikely(error == error::operation_aborted))
 135             return;    /* we're done */
 136
 137         if (error == error::message_size)
 138         {
 139             overflow_vector.insert(overflow_vector.end(),
 140                                    recv_buffer_.begin(), recv_buffer_.end());
 141             return;
 142         }
 143
 144         if (error)
 145         {
 146             std::cout << "osc_server received error code " << error << std::endl;
 147             start_receive();
 148             return;
 149         }
 150
 151         if (overflow_vector.empty())
 152             osc_dispatcher::handle_packet(recv_buffer_.begin(), bytes_transferred);
 153         else
 154         {
 155             overflow_vector.insert(overflow_vector.end(),
 156                                    recv_buffer_.begin(), recv_buffer_.end());
 157
 158             osc_dispatcher::handle_packet(overflow_vector.data(), overflow_vector.size());
 159
 160             overflow_vector.clear();
 161         }
 162
 163         start_receive();
 164         return;
 165     }
 166
 167     udp::socket socket_;
 168     boost::array<char, 1<<13 > recv_buffer_;
 169     udp::endpoint remote_endpoint_;
 170
 171     typedef static_allocator<char, 1<<20> string_allocator; /* 1 MB memory pool ...
 172                                                              *
 173                                                              * not thread-safe, but that is
 174                                                              * not required here
 175                                                              *
 176                                                              * */
 177 /*     typedef std::basic_string<char, std::char_traits<char>, string_allocator> string; */
 178     typedef std::string string;
 179     string overflow_vector;
 180 };
 181
 182 } /* namespace */
 183
 184 using detail::network_thread;
 185 using detail::osc_server;
 186
 187 } /* namespace nova */
 188
 189 #endif /* OSC_SERVER_HPP */