src/realmd/BufferedSocket.cpp

   1 /*
   2  * Copyright (C) 2005-2010 MaNGOS <http://getmangos.com/>
   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; if not, write to the Free Software
  16  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  17  */
  18
  19 /** \file
  20   \ingroup realmd
  21   */
  22
  23 #include "BufferedSocket.h"
  24
  25 #include <ace/OS_NS_string.h>
  26 #include <ace/INET_Addr.h>
  27 #include <ace/SString.h>
  28
  29 #ifndef MSG_NOSIGNAL
  30 #define MSG_NOSIGNAL 0
  31 #endif
  32
  33 BufferedSocket::BufferedSocket(void):
  34     input_buffer_(4096),
  35     remote_address_("<unknown>")
  36 {
  37 }
  38
  39 /*virtual*/ BufferedSocket::~BufferedSocket(void)
  40 {
  41 }
  42
  43 /*virtual*/ int BufferedSocket::open(void * arg)
  44 {
  45     if(Base::open(arg) == -1)
  46         return -1;
  47
  48     ACE_INET_Addr addr;
  49
  50     if(peer().get_remote_addr(addr) == -1)
  51         return -1;
  52
  53     char address[1024];
  54
  55     addr.get_host_addr(address, 1024);
  56
  57     this->remote_address_ = address;
  58
  59     this->OnAccept();
  60
  61     return 0;
  62 }
  63
  64 const std::string& BufferedSocket::get_remote_address(void) const
  65 {
  66     return this->remote_address_;
  67 }
  68
  69 size_t BufferedSocket::recv_len(void) const
  70 {
  71     return this->input_buffer_.length();
  72 }
  73
  74 bool BufferedSocket::recv_soft(char *buf, size_t len)
  75 {
  76     if(this->input_buffer_.length() < len)
  77         return false;
  78
  79     ACE_OS::memcpy(buf, this->input_buffer_.rd_ptr(), len);
  80
  81     return true;
  82 }
  83
  84 bool BufferedSocket::recv(char *buf, size_t len)
  85 {
  86     bool ret = this->recv_soft(buf, len);
  87
  88     if(ret)
  89         this->recv_skip(len);
  90
  91     return ret;
  92 }
  93
  94 void BufferedSocket::recv_skip(size_t len)
  95 {
  96     this->input_buffer_.rd_ptr(len);
  97 }
  98
  99 ssize_t BufferedSocket::noblk_send(ACE_Message_Block &message_block)
 100 {
 101     const size_t len = message_block.length();
 102
 103     if(len == 0)
 104         return -1;
 105
 106     // Try to send the message directly.
 107     ssize_t n = this->peer().send(message_block.rd_ptr(), len, MSG_NOSIGNAL);
 108
 109     if(n < 0)
 110     {
 111         if(errno == EWOULDBLOCK)
 112             // Blocking signal
 113             return 0;
 114         else
 115             // Error
 116             return -1;
 117     }
 118     else if(n == 0)
 119     {
 120         // Can this happen ?
 121         return -1;
 122     }
 123
 124     // return bytes transmitted
 125     return n;
 126 }
 127
 128 bool BufferedSocket::send(const char *buf, size_t len)
 129 {
 130     if(buf == NULL || len == 0)
 131         return true;
 132
 133     ACE_Data_Block db(
 134             len,
 135             ACE_Message_Block::MB_DATA,
 136             (const char*)buf,
 137             0,
 138             0,
 139             ACE_Message_Block::DONT_DELETE,
 140             0);
 141
 142     ACE_Message_Block message_block(
 143             &db,
 144             ACE_Message_Block::DONT_DELETE,
 145             0);
 146
 147     message_block.wr_ptr(len);
 148
 149     if(this->msg_queue()->is_empty())
 150     {
 151         // Try to send it directly.
 152         ssize_t n = this->noblk_send(message_block);
 153
 154         if(n < 0)
 155             return false;
 156         else if(n == len)
 157             return true;
 158
 159         // adjust how much bytes we sent
 160         message_block.rd_ptr((size_t)n);
 161
 162         // fall down
 163     }
 164
 165     // enqueue the message, note: clone is needed cause we cant enqueue stuff on the stack
 166     ACE_Message_Block *mb = message_block.clone();
 167
 168     if(this->msg_queue()->enqueue_tail(mb, (ACE_Time_Value *) &ACE_Time_Value::zero) == -1)
 169     {
 170         mb->release();
 171         return false;
 172     }
 173
 174     // tell reactor to call handle_output() when we can send more data
 175     if(this->reactor()->schedule_wakeup(this, ACE_Event_Handler::WRITE_MASK) == -1)
 176         return false;
 177
 178     return true;
 179 }
 180
 181 /*virtual*/ int BufferedSocket::handle_output(ACE_HANDLE /*= ACE_INVALID_HANDLE*/)
 182 {
 183     ACE_Message_Block *mb = 0;
 184
 185     if(this->msg_queue()->is_empty())
 186     {
 187         // if no more data to send, then cancel notification
 188         this->reactor()->cancel_wakeup(this, ACE_Event_Handler::WRITE_MASK);
 189         return 0;
 190     }
 191
 192     if(this->msg_queue()->dequeue_head(mb, (ACE_Time_Value *) &ACE_Time_Value::zero) == -1)
 193         return -1;
 194
 195     ssize_t n = this->noblk_send(*mb);
 196
 197     if(n < 0)
 198     {
 199         mb->release();
 200         return -1;
 201     }
 202     else if(n == mb->length())
 203     {
 204         mb->release();
 205         return 1;
 206     }
 207     else
 208     {
 209         mb->rd_ptr(n);
 210
 211         if(this->msg_queue()->enqueue_head(mb, (ACE_Time_Value *) &ACE_Time_Value::zero) == -1)
 212         {
 213             mb->release();
 214             return -1;
 215         }
 216
 217         return 0;
 218     }
 219
 220     ACE_NOTREACHED(return -1);
 221 }
 222
 223 /*virtual*/ int BufferedSocket::handle_input(ACE_HANDLE /*= ACE_INVALID_HANDLE*/)
 224 {
 225     const ssize_t space = this->input_buffer_.space();
 226
 227     ssize_t n = this->peer().recv(this->input_buffer_.wr_ptr(), space);
 228
 229     if(n < 0)
 230     {
 231         // blocking signal or error
 232         return errno == EWOULDBLOCK ? 0 : -1;
 233     }
 234     else if(n == 0)
 235     {
 236         // EOF
 237         return -1;
 238     }
 239
 240     this->input_buffer_.wr_ptr((size_t)n);
 241
 242     this->OnRead();
 243
 244     // move data in the buffer to the beginning of the buffer
 245     this->input_buffer_.crunch();
 246
 247     // return 1 in case there might be more data to read from OS
 248     return n == space ? 1 : 0;
 249 }
 250
 251 /*virtual*/ int BufferedSocket::handle_close(ACE_HANDLE h, ACE_Reactor_Mask m)
 252 {
 253     this->OnClose();
 254
 255     Base::handle_close();
 256
 257     return 0;
 258 }
 259
 260 void BufferedSocket::close_connection(void)
 261 {
 262     this->peer().close_reader();
 263     this->peer().close_writer();
 264 }
 265