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 #include <thread>
  25
  26 #include <boost/asio.hpp>
  27 #include <boost/array.hpp>
  28 #include <boost/bind.hpp>
  29
  30 #include "branch_hints.hpp"
  31 #include "osc_dispatcher.hpp"
  32
  33 #include "static_allocator.hpp"
  34 #include "nova-tt/semaphore.hpp"
  35 #include "nova-tt/thread_priority.hpp"
  36 #include "nova-tt/name_thread.hpp"
  37
  38 namespace nova {
  39
  40 namespace detail {
  41 using namespace boost::asio;
  42 using namespace boost::asio::ip;
  43
  44 class network_thread
  45 {
  46 public:
  47     void start_receive(void)
  48     {
  49         thread_ = std::move(std::thread(network_thread::start, this));
  50         sem.wait();
  51     }
  52
  53     ~network_thread(void)
  54     {
  55         if (!thread_.joinable())
  56             return;
  57         io_service_.stop();
  58         thread_.join();
  59     }
  60
  61     io_service & get_io_service(void)
  62     {
  63         return io_service_;
  64     }
  65
  66     void send_udp(const char * data, unsigned int size, udp::endpoint const & receiver)
  67     {
  68         udp::socket socket(io_service_);
  69         socket.open(udp::v4());
  70         socket.send_to(boost::asio::buffer(data, size), receiver);
  71     }
  72
  73 private:
  74     static void start(network_thread * self)
  75     {
  76 #ifdef NOVA_TT_PRIORITY_RT
  77         thread_set_priority_rt(thread_priority_interval_rt().first);
  78 #endif
  79         name_thread("Network Receive");
  80         self->run();
  81     }
  82
  83     void run(void)
  84     {
  85         sem.post();
  86
  87         io_service::work work(io_service_);
  88         io_service_.run();
  89     }
  90
  91 protected:
  92     io_service io_service_;
  93
  94 private:
  95     semaphore sem;
  96     std::thread thread_;
  97 };
  98
  99 class osc_server:
 100     private network_thread,
 101     private osc_dispatcher
 102 {
 103 public:
 104     osc_server(unsigned int port):
 105         socket_(network_thread::io_service_, udp::endpoint(udp::v4(), port))
 106     {
 107         network_thread::start_receive();
 108         start_receive();
 109     }
 110
 111     void add_responder(std::string const & address, osc_responder * resp)
 112     {
 113         osc_dispatcher::add_responder(address, resp);
 114     }
 115
 116     void remove_responder(std::string const & address, osc_responder * resp)
 117     {
 118         osc_dispatcher::remove_responder(address, resp);
 119     }
 120
 121 private:
 122     void start_receive(void)
 123     {
 124         socket_.async_receive_from(
 125             buffer(recv_buffer_), remote_endpoint_,
 126             boost::bind(&osc_server::handle_receive, this,
 127                         placeholders::error, placeholders::bytes_transferred));
 128     }
 129
 130     void handle_receive(const boost::system::error_code& error,
 131                         std::size_t bytes_transferred)
 132     {
 133         if (unlikely(error == error::operation_aborted))
 134             return;    /* we're done */
 135
 136         if (error == error::message_size)
 137         {
 138             overflow_vector.insert(overflow_vector.end(),
 139                                    recv_buffer_.begin(), recv_buffer_.end());
 140             return;
 141         }
 142
 143         if (error)
 144         {
 145             std::cout << "osc_server received error code " << error << std::endl;
 146             start_receive();
 147             return;
 148         }
 149
 150         if (overflow_vector.empty())
 151             osc_dispatcher::handle_packet(recv_buffer_.begin(), bytes_transferred);
 152         else
 153         {
 154             overflow_vector.insert(overflow_vector.end(),
 155                                    recv_buffer_.begin(), recv_buffer_.end());
 156
 157             osc_dispatcher::handle_packet(overflow_vector.data(), overflow_vector.size());
 158
 159             overflow_vector.clear();
 160         }
 161
 162         start_receive();
 163         return;
 164     }
 165
 166     udp::socket socket_;
 167     boost::array<char, 1<<13 > recv_buffer_;
 168     udp::endpoint remote_endpoint_;
 169
 170     typedef static_allocator<char, 1<<20> string_allocator; /* 1 MB memory pool ...
 171                                                              *
 172                                                              * not thread-safe, but that is
 173                                                              * not required here
 174                                                              *
 175                                                              * */
 176 /*     typedef std::basic_string<char, std::char_traits<char>, string_allocator> string; */
 177     typedef std::string string;
 178     string overflow_vector;
 179 };
 180
 181 } /* namespace */
 182
 183 using detail::network_thread;
 184 using detail::osc_server;
 185
 186 } /* namespace nova */
 187
 188 #endif /* OSC_SERVER_HPP */