lib/Spin/UDPSocket.cpp

   1 #if defined(_WIN32) && ! defined(__CYGWIN__)
   2 #       include <WinSock2.h>
   3 #endif
   4 #include "UDPSocket.h"
   5 #include <Agelena/Logger.h>
   6 #include <boost/bind.hpp>
   7 #include <boost/version.hpp>
   8 #include "Exceptions/Socket.h"
   9 #include "Exceptions/Connection/UnusableUDPSocket.h"
  10 extern "C" {
  11 #if defined(_WIN32) && ! defined(__CYGWIN__)
  12 #       define ssize_t int
  13 #       define getLastError__ WSAGetLastError
  14 #       define socklen_t int
  15 #else
  16 #       include <sys/types.h>
  17 #       include <sys/socket.h>
  18 #       include <netinet/in.h>
  19 #       include <unistd.h>
  20 #       define getLastError__() errno
  21 #       define closesocket ::close
  22 #endif
  23 }
  24 #include <cerrno>
  25 #include "Handlers/UDPDataHandler.h"
  26 #include "Private/ConnectionHandler.h"
  27
  28 #define SOCKET_CALL(statement, check_on_error, message, function)                                       \
  29         if ((statement) check_on_error)                                                                                                 \
  30                 throw Exceptions::SocketError(message, function, getLastError__());                     \
  31         else                                                                                                                                                    \
  32         { /* all is well */ }
  33
  34 #ifdef _MSC_VER
  35 #pragma warning(disable: 4389) // == signed/unsigned mismatch - part of FD_SET
  36 #endif
  37
  38 #define AGELENA_COMPONENT_ID    0x5350493eUL
  39 #define AGELENA_SUBCOMPONENT_ID 0x55445053UL
  40
  41 namespace Spin
  42 {
  43         UDPSocket::UDPSocket(const Details::Address & address, unsigned short port)
  44                 : fd_(-1),
  45                   data_handler_(0)
  46         {
  47                 SOCKET_CALL(fd_ = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP), < 0, "socket allocation failed", "UDPSocket constructor");
  48                 if (port != 0)
  49                 {
  50                         int on(1);
  51                         SOCKET_CALL(setsockopt(fd_, SOL_SOCKET, SO_REUSEADDR, (const char *)&on, sizeof(on)), < 0, "setsockopt failed", "UDPSocket constructor");
  52                         sockaddr_in in_address;
  53                         memset(&in_address, 0, sizeof(in_address));
  54                         in_address.sin_addr.s_addr = address.u_.u32_;
  55                         in_address.sin_family = AF_INET;
  56                         in_address.sin_port = htons(port);
  57                         SOCKET_CALL(bind(fd_, (const sockaddr *)(&in_address), sizeof(sockaddr_in)), < 0, "bind failed", "UDPSocket constructor");
  58                 }
  59                 else
  60                 { /* no need to bind */ }
  61         }
  62
  63         UDPSocket::~UDPSocket()
  64         {
  65                 clearDataHandler();
  66                 closesocket(fd_);
  67         }
  68
  69         std::size_t UDPSocket::send(const Details::Address & to, unsigned short port, const std::vector< char > & data)
  70         {
  71                 sockaddr_in remote_address;
  72                 memset(&remote_address, 0, sizeof(remote_address));
  73                 remote_address.sin_addr.s_addr = to.u_.u32_;
  74                 remote_address.sin_family = AF_INET;
  75                 remote_address.sin_port = htons(port);
  76
  77                 const char * curr(&data[0]);
  78                 const char * end(curr + data.size());
  79                 if (fd_ == -1)
  80                         throw Exceptions::Connection::UnusableUDPSocket();
  81                 else
  82                 { /* all is well */ }
  83                 ssize_t sent(::sendto(fd_, curr, std::distance(curr, end), 0, (const sockaddr *)(&remote_address), sizeof(remote_address)));
  84                 if (sent < 0)
  85                 {
  86                         throw Exceptions::SocketError("Send failed", "UDPSocket::send", getLastError__());
  87                 }
  88                 else
  89                 { /* all is well */ }
  90
  91                 return sent;
  92         }
  93
  94         boost::tuple< Details::Address, unsigned short, std::size_t > UDPSocket::recv(std::vector< char > & buffer, unsigned long timeout/* = ~0*/)
  95         {
  96                 boost::recursive_mutex::scoped_lock sentinel(peek_buffer_lock_);
  97                 if (fd_ == -1)
  98                         throw Exceptions::Connection::UnusableUDPSocket();
  99                 else
 100                 { /* all is well */ }
 101
 102                 if (boost::tuples::get<3>(peek_buffer_).empty())
 103                 {
 104                         sentinel.unlock();
 105                         bool we_own_the_buffer_size(false);
 106                         if (buffer.size() == 0)
 107                         {
 108                                 buffer.resize(4096, 0);
 109                                 we_own_the_buffer_size = true;
 110                         }
 111                         else
 112                         { /* keep the buffer as is */ }
 113                         sockaddr_in remote_address;
 114                         memset(&remote_address, 0, sizeof(remote_address));
 115                         socklen_t remote_address_size(sizeof(remote_address));
 116                         bool timed_out(false);
 117                         if (timeout != ~0)
 118                         {
 119                                 int highest_fd = fd_ + 1;
 120                                 fd_set read_set;
 121                                 fd_set write_set;
 122                                 fd_set exc_set;
 123                                 FD_ZERO(&read_set);
 124                                 FD_ZERO(&write_set);
 125                                 FD_ZERO(&exc_set);
 126                                 FD_SET(fd_, &read_set);
 127                                 timeval time_out;
 128                                 time_out.tv_sec = 0;
 129                                 time_out.tv_usec = timeout;
 130
 131                                 switch (select(highest_fd, &read_set, &write_set, &exc_set, &time_out))
 132                                 {
 133                                 case 0 :
 134                                         timed_out = true;
 135                                         break;
 136                                 case 1 :
 137                                         timed_out = false;
 138                                         break;
 139                                 default :
 140                                         throw Exceptions::SocketError("Select call for recv failed", "UDPSocket::recv", getLastError__());
 141                                 }
 142                         }
 143                         else
 144                         { /* no time-out period */ }
 145                         ssize_t received(timed_out ? 0 : recvfrom(fd_, &buffer[0], buffer.size(), 0, (sockaddr *)(&remote_address), &remote_address_size));
 146                         if (received < 0)
 147                         {
 148                                 throw Exceptions::SocketError("Recv failed", "UDPSocket::recv", getLastError__());
 149                         }
 150                         else
 151                         { /* all is well */ }
 152                         assert(remote_address_size == sizeof(remote_address));
 153                         if (we_own_the_buffer_size)
 154                                 buffer.resize(received);
 155                         else
 156                         { /* we don't own the buffer size, so let the called take care of it */ }
 157                         return boost::make_tuple(Details::Address(remote_address.sin_addr.s_addr), ntohs(remote_address.sin_port), received);
 158                 }
 159                 else
 160                 {
 161                         buffer = boost::tuples::get<3>(peek_buffer_);
 162                         boost::tuples::get<3>(peek_buffer_).clear();
 163                         return boost::make_tuple(boost::tuples::get<0>(peek_buffer_), boost::tuples::get<1>(peek_buffer_), boost::tuples::get<2>(peek_buffer_));
 164                 }
 165         }
 166
 167         boost::tuple< Details::Address, unsigned short, std::size_t > UDPSocket::peek(std::vector< char > & buffer, unsigned long timeout/* = ~0*/)
 168         {
 169                 boost::recursive_mutex::scoped_lock sentinel(peek_buffer_lock_);
 170                 if (fd_ == -1)
 171                         throw Exceptions::Connection::UnusableUDPSocket();
 172                 else
 173                 { /* all is well */ }
 174                 if (boost::tuples::get<3>(peek_buffer_).empty())
 175                 {
 176                         boost::tie(boost::tuples::get<0>(peek_buffer_), boost::tuples::get<1>(peek_buffer_), boost::tuples::get<2>(peek_buffer_)) = recv(boost::tuples::get<3>(peek_buffer_), timeout);
 177                 }
 178                 else
 179                 { /* already have a peeked buffer */ }
 180
 181                 buffer = boost::tuples::get<3>(peek_buffer_);
 182                 return boost::make_tuple(boost::tuples::get<0>(peek_buffer_), boost::tuples::get<1>(peek_buffer_), boost::tuples::get<2>(peek_buffer_));
 183         }
 184
 185         void UDPSocket::clearPeekBuffer()
 186         {
 187                 boost::recursive_mutex::scoped_lock sentinel(peek_buffer_lock_);
 188                 boost::tuples::get<3>(peek_buffer_).clear();
 189         }
 190
 191         bool UDPSocket::poll() const
 192         {
 193                 boost::recursive_mutex::scoped_lock sentinel(fd_lock_);
 194                 if (fd_ == -1)
 195                         throw Exceptions::Connection::UnusableUDPSocket();
 196                 else
 197                 { /* all is well */ }
 198
 199                 bool retval(false);
 200
 201                 int highest_fd = fd_ + 1;
 202                 fd_set read_set;
 203                 fd_set write_set;
 204                 fd_set exc_set;
 205                 FD_ZERO(&read_set);
 206                 FD_ZERO(&write_set);
 207                 FD_ZERO(&exc_set);
 208                 FD_SET(fd_, &read_set);
 209                 timeval timeout;
 210                 timeout.tv_sec = 0;
 211                 timeout.tv_usec = 1;
 212
 213                 switch (select(highest_fd, &read_set, &write_set, &exc_set, &timeout))
 214                 {
 215                 case 0 :
 216                         retval = false;
 217                         break;
 218                 case 1 :
 219                         retval = true;
 220                         break;
 221                 default :
 222                         throw Exceptions::SocketError("Select call for poll failed", "UDPSocket::poll", getLastError__());
 223                 }
 224
 225                 return retval;
 226         }
 227
 228         void UDPSocket::close()
 229         {
 230                 if (fd_ != -1)
 231                 {
 232                         closesocket(fd_);
 233                         fd_ = -1;
 234                 }
 235                 else
 236                 { /* already closed */ }
 237         }
 238
 239         void UDPSocket::setDataHandler(Handlers::UDPDataHandler & handler, OnErrorCallback on_error_callback/* = OnErrorCallback()*/)
 240         {
 241                 boost::recursive_mutex::scoped_lock sentinel(fd_lock_);
 242                 if (fd_ == -1)
 243                         throw Exceptions::Connection::UnusableUDPSocket();
 244                 else
 245                 { /* all is well */ }
 246                 data_handler_ = &handler;
 247                 Private::ConnectionHandler::getInstance().attach(fd_, boost::bind(&UDPSocket::onDataReady_, this), on_error_callback);
 248         }
 249
 250         void UDPSocket::clearDataHandler()
 251         {
 252                 boost::recursive_mutex::scoped_lock sentinel(fd_lock_);
 253                 if (data_handler_ && fd_ != -1)
 254                 {
 255                         AGELENA_DEBUG_1("Detaching FD %1% from the connection handler", fd_);
 256                         Private::ConnectionHandler::getInstance().detach(fd_);
 257                         data_handler_ = 0;
 258                 }
 259                 else
 260                 { /* nothing to clear */ }
 261         }
 262
 263         void UDPSocket::onDataReady_()
 264         {
 265                 if (data_handler_)
 266                 {
 267                         /* If the internal weak pointer in enable_shared_from_this is
 268                          * NULL, we're not stored in a shared_ptr, which means we can't
 269                          * pass one to our users. This is possible, for example, if the
 270                          * client app uses RAII to handle the life-cycle of the UDP
 271                          * socket, rather than using a shared_ptr. */
 272                         /* NOTE: this is only known to work in version 1.33.1 of Boost,
 273                          *       so using UDPSocket with RAII is only supported in that
 274                          *       particular version of Boost. The reason for this being
 275                          *       here is to accomodate one particular user of Arachnida.
 276                          *       Use of UDPSocet in this fashion is deprecated. */
 277 #if BOOST_VERSION == 103301
 278                         if (_internal_weak_this.lock())
 279                                 (*data_handler_)(shared_from_this());
 280                         else
 281                                 (*data_handler_)(this);
 282 #else
 283                         /* NOTE: if your application crashed here or passes you an empty
 284                          *       shared pointer, it means you are not using version 1.33.1
 285                          *       of boost and you are using the UDPSocket as a free or
 286                          *       stack-based object, rather than through a shared_ptr.
 287                          *       If that is the case, you have three avenues before you:
 288                          *                      1.      check whether the version of Boost you are using
 289                          *                              has an enable_shared_from_this with an accessible
 290                          *                              _internal_weak member and, if so, add it to the
 291                          *                              exceptions above (and contribute the change back
 292                          *                              to support@vlinder.ca, please);
 293                          *                      2.      fix your code to use UDPSocket from a shared_ptr
 294                          *                      3.      contact support@vlinder.ca and tell us about your
 295                          *                              problem - we'll be happy to help you fix it. As you
 296                          *                              can see above, we're happy to adjust our code for
 297                          *                              customers. */
 298                         (*data_handler_)(shared_from_this());
 299 #endif
 300                 }
 301                 else
 302                 { /* no-op */ }
 303         }
 304 }
 305