lib/Spin/UDPSocket.cpp

   1 #include "UDPSocket.h"
   2 #include <Agelena/Logger.h>
   3 #include <boost/bind.hpp>
   4 #include "Exceptions/Socket.h"
   5 extern "C" {
   6 #if defined(_WIN32) && ! defined(__CYGWIN__)
   7 #       include <WinSock2.h>
   8 #       define ssize_t int
   9 #       define getLastError__ WSAGetLastError
  10 #       define socklen_t int
  11 #else
  12 #       include <sys/types.h>
  13 #       include <sys/socket.h>
  14 #       include <netinet/in.h>
  15 #       include <unistd.h>
  16 #       define getLastError__() errno
  17 #       define closesocket ::close
  18 #endif
  19 }
  20 #include <cerrno>
  21 #include "Handlers/UDPDataHandler.h"
  22 #include "Private/ConnectionHandler.h"
  23
  24 #define SOCKET_CALL(statement, check_on_error, message, function)                                       \
  25         if ((statement) check_on_error)                                                                                                 \
  26                 throw Exceptions::SocketError(message, function, getLastError__());                     \
  27         else                                                                                                                                                    \
  28         { /* all is well */ }
  29
  30 namespace Spin
  31 {
  32         UDPSocket::UDPSocket(const Details::Address & address, unsigned short port)
  33                 : status_(good__),
  34                   fd_(-1),
  35                   data_handler_(0)
  36         {
  37                 SOCKET_CALL(fd_ = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP), < 0, "socket allocation failed", "UDPSocket constructor");
  38                 if (port != 0)
  39                 {
  40                         int on(1);
  41                         SOCKET_CALL(setsockopt(fd_, SOL_SOCKET, SO_REUSEADDR, (const char *)&on, sizeof(on)), < 0, "setsockopt failed", "UDPSocket constructor");
  42                         sockaddr_in in_address;
  43                         memset(&in_address, 0, sizeof(in_address));
  44                         in_address.sin_addr.s_addr = address.u_.u32_;
  45                         in_address.sin_family = AF_INET;
  46                         in_address.sin_port = htons(port);
  47                         SOCKET_CALL(bind(fd_, (const sockaddr *)(&in_address), sizeof(sockaddr_in)), < 0, "bind failed", "UDPSocket constructor");
  48                 }
  49                 else
  50                 { /* no need to bind */ }
  51         }
  52
  53         UDPSocket::~UDPSocket()
  54         {
  55                 clearDataHandler();
  56                 closesocket(fd_);
  57         }
  58
  59         std::size_t UDPSocket::send(const Details::Address & to, unsigned short port, const std::vector< char > & data)
  60         {
  61                 checkStatus("UDPSocket::send");
  62                 sockaddr_in remote_address;
  63                 memset(&remote_address, 0, sizeof(remote_address));
  64                 remote_address.sin_addr.s_addr = to.u_.u32_;
  65                 remote_address.sin_family = AF_INET;
  66                 remote_address.sin_port = htons(port);
  67
  68                 const char * curr(&data[0]);
  69                 const char * end(curr + data.size());
  70                 ssize_t sent(::sendto(fd_, curr, std::distance(curr, end), 0, (const sockaddr *)(&remote_address), sizeof(remote_address)));
  71                 if (sent < 0)
  72                 {
  73                         status_ |= error__;
  74                         throw Exceptions::SocketError("Send failed", "UDPSocket::send", getLastError__());
  75                 }
  76                 else
  77                 { /* all is well */ }
  78
  79                 return sent;
  80         }
  81
  82         boost::tuple< Details::Address, unsigned short, std::size_t > UDPSocket::recv(std::vector< char > & buffer, unsigned long timeout/* = ~0*/)
  83         {
  84                 checkStatus("UDPSocket::recv");
  85                 boost::recursive_mutex::scoped_lock sentinel(peek_buffer_lock_);
  86                 if (boost::tuples::get<3>(peek_buffer_).empty())
  87                 {
  88                         sentinel.unlock();
  89                         bool we_own_the_buffer_size(false);
  90                         if (buffer.size() == 0)
  91                         {
  92                                 buffer.resize(4096, 0);
  93                                 we_own_the_buffer_size = true;
  94                         }
  95                         else
  96                         { /* keep the buffer as is */ }
  97                         sockaddr_in remote_address;
  98                         memset(&remote_address, 0, sizeof(remote_address));
  99                         socklen_t remote_address_size(sizeof(remote_address));
 100                         bool timed_out(false);
 101                         if (timeout != ~0)
 102                         {
 103                                 int highest_fd = fd_ + 1;
 104                                 fd_set read_set;
 105                                 fd_set write_set;
 106                                 fd_set exc_set;
 107                                 FD_ZERO(&read_set);
 108                                 FD_ZERO(&write_set);
 109                                 FD_ZERO(&exc_set);
 110                                 FD_SET(fd_, &read_set);
 111                                 timeval time_out;
 112                                 time_out.tv_sec = 0;
 113                                 time_out.tv_usec = timeout;
 114
 115                                 switch (select(highest_fd, &read_set, &write_set, &exc_set, &time_out))
 116                                 {
 117                                 case 0 :
 118                                         timed_out = true;
 119                                         break;
 120                                 case 1 :
 121                                         timed_out = false;
 122                                         break;
 123                                 default :
 124                                         status_ |= error__;
 125                                         throw Exceptions::SocketError("Select call for recv failed", "UDPSocket::recv", getLastError__());
 126                                 }
 127                         }
 128                         else
 129                         { /* no time-out period */ }
 130                         ssize_t received(timed_out ? 0 : recvfrom(fd_, &buffer[0], buffer.size(), 0, (sockaddr *)(&remote_address), &remote_address_size));
 131                         if (received < 0)
 132                         {
 133                                 status_ |= error__;
 134                                 throw Exceptions::SocketError("Recv failed", "UDPSocket::recv", getLastError__());
 135                         }
 136                         else
 137                         { /* all is well */ }
 138                         assert(remote_address_size == sizeof(remote_address));
 139                         if (we_own_the_buffer_size)
 140                                 buffer.resize(received);
 141                         else
 142                         { /* we don't own the buffer size, so let the called take care of it */ }
 143                         return boost::make_tuple(Details::Address(remote_address.sin_addr.s_addr), ntohs(remote_address.sin_port), received);
 144                 }
 145                 else
 146                 {
 147                         buffer = boost::tuples::get<3>(peek_buffer_);
 148                         boost::tuples::get<3>(peek_buffer_).clear();
 149                         return boost::make_tuple(boost::tuples::get<0>(peek_buffer_), boost::tuples::get<1>(peek_buffer_), boost::tuples::get<2>(peek_buffer_));
 150                 }
 151         }
 152
 153         boost::tuple< Details::Address, unsigned short, std::size_t > UDPSocket::peek(std::vector< char > & buffer, unsigned long timeout/* = ~0*/)
 154         {
 155                 boost::recursive_mutex::scoped_lock sentinel(peek_buffer_lock_);
 156                 if (boost::tuples::get<3>(peek_buffer_).empty())
 157                 {
 158                         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);
 159                 }
 160                 else
 161                 { /* already have a peeked buffer */ }
 162
 163                 buffer = boost::tuples::get<3>(peek_buffer_);
 164                 return boost::make_tuple(boost::tuples::get<0>(peek_buffer_), boost::tuples::get<1>(peek_buffer_), boost::tuples::get<2>(peek_buffer_));
 165         }
 166
 167         void UDPSocket::clearPeekBuffer()
 168         {
 169                 boost::recursive_mutex::scoped_lock sentinel(peek_buffer_lock_);
 170                 boost::tuples::get<3>(peek_buffer_).clear();
 171         }
 172
 173         bool UDPSocket::poll() const
 174         {
 175                 checkStatus("UDPSocket::poll");
 176                 bool retval(false);
 177
 178                 int highest_fd = fd_ + 1;
 179                 fd_set read_set;
 180                 fd_set write_set;
 181                 fd_set exc_set;
 182                 FD_ZERO(&read_set);
 183                 FD_ZERO(&write_set);
 184                 FD_ZERO(&exc_set);
 185                 FD_SET(fd_, &read_set);
 186                 timeval timeout;
 187                 timeout.tv_sec = 0;
 188                 timeout.tv_usec = 1;
 189
 190                 switch (select(highest_fd, &read_set, &write_set, &exc_set, &timeout))
 191                 {
 192                 case 0 :
 193                         retval = false;
 194                         break;
 195                 case 1 :
 196                         retval = true;
 197                         break;
 198                 default :
 199                         status_ |= error__;
 200                         throw Exceptions::SocketError("Select call for poll failed", "UDPSocket::poll", getLastError__());
 201                 }
 202
 203                 return retval;
 204         }
 205
 206         void UDPSocket::close()
 207         {
 208                 if (fd_ != -1)
 209                 {
 210                         closesocket(fd_);
 211                         fd_ = -1;
 212                 }
 213                 else
 214                 { /* already closed */ }
 215                 status_ |= done__;
 216         }
 217
 218         void UDPSocket::setDataHandler(Handlers::UDPDataHandler & handler, OnErrorCallback on_error_callback/* = OnErrorCallback()*/)
 219         {
 220                 boost::recursive_mutex::scoped_lock sentinel(fd_lock_);
 221                 data_handler_ = &handler;
 222                 Private::ConnectionHandler::getInstance().attach(fd_, boost::bind(&UDPSocket::onDataReady_, this), on_error_callback);
 223         }
 224
 225         void UDPSocket::clearDataHandler()
 226         {
 227                 boost::recursive_mutex::scoped_lock sentinel(fd_lock_);
 228                 if (data_handler_)
 229                 {
 230                         AGELENA_DEBUG_1("Detaching FD %1% from the connection handler", fd_);
 231                         Private::ConnectionHandler::getInstance().detach(fd_);
 232                         data_handler_ = 0;
 233                 }
 234                 else
 235                 { /* nothing to clear */ }
 236         }
 237
 238         void UDPSocket::checkStatus(const char * for_whom) const
 239         {
 240                 if (status_ != good__)
 241                 {
 242                         status_ |= error__;
 243                         throw Exceptions::SocketError("Socket not usable", for_whom, -1);
 244                 }
 245                 else
 246                 { /* all is well */ }
 247         }
 248
 249         void UDPSocket::onDataReady_()
 250         {
 251                 if (data_handler_ && status_ == good__)
 252                 {
 253                         /* If the internal weak pointer in enable_shared_from_this is
 254                          * NULL, we're not stored in a shared_ptr, which means we can't
 255                          * pass one to our users. This is possible, for example, if the
 256                          * client app uses RAII to handle the life-cycle of the UDP
 257                          * socket, rather than using a shared_ptr. */
 258                         if (_internal_weak_this.lock())
 259                                 (*data_handler_)(shared_from_this());
 260                         else
 261                                 (*data_handler_)(this);
 262                 }
 263                 else
 264                 { /* no-op */ }
 265         }
 266 }
 267