server/supernova/sc/sc_osc_handler.hpp

   1 //  osc handler for supercollider-style communication
   2 //  Copyright (C) 2009, 2010 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 SERVER_SC_OSC_HANDLER_HPP
  20 #define SERVER_SC_OSC_HANDLER_HPP
  21
  22 #include <vector>
  23 #include <algorithm>
  24
  25 #include <boost/date_time/microsec_time_clock.hpp>
  26 #include <boost/intrusive/treap_set.hpp>
  27
  28 #include "osc/OscReceivedElements.h"
  29
  30 #include "../server/dynamic_endpoint.hpp"
  31 #include "../server/memory_pool.hpp"
  32 #include "../server/server_args.hpp"
  33 #include "../server/server_scheduler.hpp"
  34 #include "../utilities/osc_server.hpp"
  35 #include "../utilities/sized_array.hpp"
  36 #include "../utilities/static_pool.hpp"
  37 #include "../utilities/time_tag.hpp"
  38
  39 namespace nova
  40 {
  41 typedef bool (*AsyncStageFn)(World *inWorld, void* cmdData);
  42 typedef void (*AsyncFreeFn)(World *inWorld, void* cmdData);
  43
  44 namespace detail
  45 {
  46 using namespace boost::asio;
  47 using namespace boost::asio::ip;
  48
  49
  50 /**
  51  * observer to receive osc notifications
  52  *
  53  * \todo shall we use a separate thread for observer notifications?
  54  * */
  55 class sc_notify_observers
  56 {
  57     typedef std::vector<nova_endpoint> observer_vector;
  58
  59 public:
  60     sc_notify_observers(boost::asio::io_service & io_service):
  61         udp_socket(io_service), tcp_socket(io_service)
  62     {}
  63
  64     void add_observer(nova_endpoint const & ep)
  65     {
  66         observers.push_back(ep);
  67     }
  68
  69     void remove_observer(nova_endpoint const & ep)
  70     {
  71         observer_vector::iterator it = std::find(observers.begin(),
  72                                                  observers.end(), ep);
  73         assert (it != observers.end());
  74
  75         observers.erase(it);
  76     }
  77
  78
  79     /* @{ */
  80     /** notifications, should be called from the real-time thread */
  81     void notification_node_started(const server_node * node)
  82     {
  83         notify("/n_go", node);
  84     }
  85
  86     void notification_node_ended(const server_node * node)
  87     {
  88         notify("/n_end", node);
  89     }
  90
  91     void notification_node_turned_off(const server_node * node)
  92     {
  93         notify("/n_off", node);
  94     }
  95
  96     void notification_node_turned_on(const server_node * node)
  97     {
  98         notify("/n_on", node);
  99     }
 100
 101     void notification_node_moved(const server_node * node)
 102     {
 103         notify("/n_move", node);
 104     }
 105
 106     void send_trigger(int32_t node_id, int32_t trigger_id, float value);
 107
 108     void send_node_reply(int32_t node_id, int reply_id, const char* command_name, int argument_count, const float* values);
 109     /* @} */
 110
 111     /** send notifications, should not be called from the real-time thread */
 112     void send_notification(const char * data, size_t length);
 113
 114     /* @{ */
 115     /** sending functions */
 116     void send(const char * data, size_t size, nova_endpoint const & endpoint)
 117     {
 118         nova_protocol prot = endpoint.protocol();
 119         if (prot.family() == AF_INET && prot.type() == SOCK_DGRAM)
 120         {
 121             udp::endpoint ep(endpoint.address(), endpoint.port());
 122             send_udp(data, size, ep);
 123         }
 124         else if (prot.family() == AF_INET && prot.type() == SOCK_STREAM)
 125         {
 126             tcp::endpoint ep(endpoint.address(), endpoint.port());
 127             send_tcp(data, size, ep);
 128         }
 129     }
 130
 131     void send_udp(const char * data, unsigned int size, udp::endpoint const & receiver)
 132     {
 133         boost::mutex::scoped_lock lock(udp_mutex);
 134         sc_notify_observers::udp_socket.send_to(boost::asio::buffer(data, size), receiver);
 135     }
 136
 137     void send_tcp(const char * data, unsigned int size, tcp::endpoint const & receiver)
 138     {
 139         boost::mutex::scoped_lock lock(tcp_mutex);
 140         tcp_socket.connect(receiver);
 141         boost::asio::write(tcp_socket, boost::asio::buffer(data, size));
 142     }
 143     /* @} */
 144
 145 private:
 146     void notify(const char * address_pattern, const server_node * node);
 147     void send_notification(const char * data, size_t length, nova_endpoint const & endpoint);
 148
 149     observer_vector observers;
 150
 151 protected:
 152     udp::socket udp_socket;
 153     tcp::socket tcp_socket;
 154     boost::mutex udp_mutex, tcp_mutex;
 155 };
 156
 157 class sc_scheduled_bundles
 158 {
 159 public:
 160     struct bundle_node:
 161         public boost::intrusive::bs_set_base_hook<>
 162     {
 163         bundle_node(time_tag const & timeout, const char * data, nova_endpoint const & endpoint):
 164             timeout_(timeout), data_(data), endpoint_(endpoint)
 165         {}
 166
 167         void run(void);
 168
 169         const time_tag timeout_;
 170         const char * const data_;
 171         const nova_endpoint endpoint_;
 172
 173         friend bool operator< (const bundle_node & lhs, const bundle_node & rhs)
 174         {
 175             return priority_order(lhs, rhs);
 176         }
 177
 178         friend bool priority_order (const bundle_node & lhs, const bundle_node & rhs)
 179         {
 180             return lhs.timeout_ < rhs.timeout_; // lower value, higher priority
 181         }
 182     };
 183
 184     typedef boost::intrusive::treap_multiset<bundle_node> bundle_queue_t;
 185
 186     void insert_bundle(time_tag const & timeout, const char * data, size_t length,
 187                        nova_endpoint const & endpoint);
 188
 189     void execute_bundles(time_tag const & last, time_tag const & now);
 190
 191     void clear_bundles(void)
 192     {
 193         bundle_q.clear_and_dispose(dispose_bundle);
 194     }
 195
 196     static void dispose_bundle(bundle_node * node)
 197     {
 198         node->~bundle_node();
 199         rt_pool.free(node);
 200     }
 201
 202 private:
 203     bundle_queue_t bundle_q;
 204 };
 205
 206 class sc_osc_handler:
 207     private detail::network_thread,
 208     public sc_notify_observers
 209 {
 210     /* @{ */
 211     /** constructor helpers */
 212     void open_tcp_acceptor(tcp const & protocol, unsigned int port);
 213     void open_udp_socket(udp const & protocol, unsigned int port);
 214     bool open_socket(int family, int type, int protocol, unsigned int port);
 215     /* @} */
 216
 217 public:
 218     sc_osc_handler(server_arguments const & args):
 219         sc_notify_observers(detail::network_thread::io_service_),
 220         dump_osc_packets(0), error_posting(1), quit_received(false),
 221         tcp_acceptor_(detail::network_thread::io_service_),
 222         tcp_password_(args.server_password.c_str())
 223     {
 224         if (args.tcp_port && !open_socket(AF_INET, SOCK_STREAM, IPPROTO_TCP, args.tcp_port))
 225             throw std::runtime_error("cannot open socket");
 226         if (args.udp_port && !open_socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP, args.udp_port))
 227             throw std::runtime_error("cannot open socket");
 228     }
 229
 230     void start_receive_thread(void)
 231     {
 232         detail::network_thread::start_receive();
 233     }
 234
 235     ~sc_osc_handler(void)
 236     {}
 237
 238     typedef osc::ReceivedPacket osc_received_packet;
 239     typedef osc::ReceivedBundle received_bundle;
 240     typedef osc::ReceivedMessage received_message;
 241
 242     struct received_packet:
 243         public audio_sync_callback
 244     {
 245         received_packet(const char * dat, size_t length, nova_endpoint const & endpoint):
 246             data(dat), length(length), endpoint_(endpoint)
 247         {}
 248
 249         void * operator new(std::size_t size, void* ptr)
 250         {
 251             return ::operator new(size, ptr);
 252         }
 253
 254         static received_packet * alloc_packet(const char * data, size_t length,
 255                                               nova_endpoint const & remote_endpoint);
 256
 257         void run(void);
 258
 259         const char * const data;
 260         const size_t length;
 261         const nova_endpoint endpoint_;
 262     };
 263
 264 private:
 265     /* @{ */
 266     /** udp socket handling */
 267     void start_receive_udp(void)
 268     {
 269         sc_notify_observers::udp_socket.async_receive_from(
 270             buffer(recv_buffer_), udp_remote_endpoint_,
 271             boost::bind(&sc_osc_handler::handle_receive_udp, this,
 272                         placeholders::error, placeholders::bytes_transferred));
 273     }
 274
 275     void handle_receive_udp(const boost::system::error_code& error,
 276                             std::size_t bytes_transferred);
 277     /* @} */
 278
 279     /* @{ */
 280     /** tcp connection handling */
 281     class tcp_connection:
 282         public boost::enable_shared_from_this<tcp_connection>
 283     {
 284     public:
 285         typedef boost::shared_ptr<tcp_connection> pointer;
 286
 287         static pointer create(boost::asio::io_service& io_service)
 288         {
 289             return pointer(new tcp_connection(io_service));
 290         }
 291
 292         tcp::socket& socket()
 293         {
 294             return socket_;
 295         }
 296
 297         void start(sc_osc_handler * self);
 298
 299     private:
 300         tcp_connection(boost::asio::io_service& io_service)
 301             : socket_(io_service)
 302         {}
 303
 304         tcp::socket socket_;
 305     };
 306
 307     void start_accept(void)
 308     {
 309         tcp_connection::pointer new_connection = tcp_connection::create(tcp_acceptor_.get_io_service());
 310
 311         tcp_acceptor_.async_accept(new_connection->socket(),
 312             boost::bind(&sc_osc_handler::handle_accept, this, new_connection,
 313             boost::asio::placeholders::error));
 314     }
 315
 316     void handle_accept(tcp_connection::pointer new_connection,
 317                        const boost::system::error_code& error)
 318     {
 319         if (!error)
 320         {
 321             new_connection->start(this);
 322             start_accept();
 323         }
 324     }
 325     /* @} */
 326
 327 public:
 328     void dumpOSC(int i)
 329     {
 330         dump_osc_packets = i;
 331     }
 332
 333 private:
 334     int dump_osc_packets;
 335
 336     /* @{ */
 337 public:
 338     /** \todo how to handle temporary message error suppression? */
 339
 340     void set_error_posting(int val)
 341     {
 342         error_posting = val;
 343     }
 344
 345 private:
 346     int error_posting;
 347     /* @} */
 348
 349     /* @{ */
 350     /** packet handling */
 351 public:
 352     void handle_packet_async(const char* data, size_t length, nova::nova_endpoint const & endpoint);
 353     void handle_packet(const char* data, size_t length, nova::nova_endpoint const & endpoint);
 354     time_tag handle_bundle_nrt(const char * data_, std::size_t length);
 355
 356 private:
 357     template <bool realtime>
 358     void handle_bundle(received_bundle const & bundle, nova_endpoint const & endpoint);
 359     template <bool realtime>
 360     void handle_message(received_message const & message, size_t msg_size, nova_endpoint const & endpoint);
 361     template <bool realtime>
 362     void handle_message_int_address(received_message const & message, size_t msg_size, nova_endpoint const & endpoint);
 363     template <bool realtime>
 364     void handle_message_sym_address(received_message const & message, size_t msg_size, nova_endpoint const & endpoint);
 365
 366     friend class sc_scheduled_bundles::bundle_node;
 367     /* @} */
 368
 369     /* @{ */
 370     /** bundle scheduling */
 371 public:
 372     void clear_scheduled_bundles(void)
 373     {
 374         scheduled_bundles.clear_bundles();
 375     }
 376
 377     void execute_scheduled_bundles(void)
 378     {
 379         scheduled_bundles.execute_bundles(last, now);
 380     }
 381
 382     void increment_logical_time(time_tag const & diff)
 383     {
 384         last = now;
 385         now += diff;
 386     }
 387
 388     void update_time_from_system(void)
 389     {
 390         now = time_tag::from_ptime(boost::date_time::microsec_clock<boost::posix_time::ptime>::universal_time());
 391         last = now - time_per_tick;
 392     }
 393
 394     time_tag const & current_time(void) const
 395     {
 396         return now;
 397     }
 398
 399     sc_scheduled_bundles scheduled_bundles;
 400     time_tag now, last;
 401     time_tag time_per_tick;
 402 /* @} */
 403
 404
 405     void do_asynchronous_command(World * world, void* replyAddr, const char* cmdName, void *cmdData,
 406                                  AsyncStageFn stage2, AsyncStageFn stage3, AsyncStageFn stage4, AsyncFreeFn cleanup,
 407                                  int completionMsgSize, void* completionMsgData);
 408
 409     bool quit_received;
 410
 411 private:
 412     /* @{ */
 413 /*    udp::socket udp_socket_;*/
 414     udp::endpoint udp_remote_endpoint_;
 415
 416     tcp::acceptor tcp_acceptor_;
 417     const char * tcp_password_; /* we are not owning this! */
 418
 419     boost::array<char, 1<<15 > recv_buffer_;
 420
 421     std::vector<char> overflow_vector;
 422     /* @} */
 423 };
 424
 425 } /* namespace detail */
 426
 427 using detail::sc_osc_handler;
 428
 429 } /* namespace nova */
 430
 431
 432 #endif /* SERVER_SC_OSC_HANDLER_HPP */