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