ACE/ACEXML/common/Mem_Map_Stream.cpp

   1 #include "ace/FILE_Addr.h"
   2 #include "ace/OS_NS_unistd.h"
   3 #include "ace/Truncate.h"
   4
   5 #include "ACEXML/common/Mem_Map_Stream.h"
   6
   7 ACEXML_Mem_Map_Stream::ACEXML_Mem_Map_Stream ()
   8   : svc_handler_ (0),
   9     recv_pos_ (0),
  10     get_pos_ (0),
  11     end_of_mapping_plus1_ (0)
  12 {
  13 }
  14
  15 ACE_SOCK_Stream &
  16 ACEXML_Mem_Map_Stream::stream ()
  17 {
  18   return svc_handler_->peer ();
  19 }
  20
  21 ssize_t
  22 ACEXML_Mem_Map_Stream::send_n (const void *buf, size_t size,
  23                                ACE_Time_Value *tv)
  24 {
  25   return svc_handler_->peer ().send_n (buf, size, 0, tv);
  26 }
  27
  28 int
  29 ACEXML_Mem_Map_Stream::eof () const
  30 {
  31   return this->get_pos_ >= this->end_of_mapping_plus1_;
  32 }
  33
  34 int
  35 ACEXML_Mem_Map_Stream::get_char ()
  36 {
  37   if (this->eof () && this->grow_file_and_remap () == -1)
  38     return EOF;
  39   return *this->get_pos_++;
  40 }
  41
  42 void
  43 ACEXML_Mem_Map_Stream::rewind ()
  44 {
  45   this->recv_pos_ = reinterpret_cast<char *> (this->mem_map_.addr ());
  46   this->get_pos_ = this->recv_pos_;
  47   this->end_of_mapping_plus1_ = this->recv_pos_ + this->mem_map_.size ();
  48 }
  49
  50 int
  51 ACEXML_Mem_Map_Stream::peek_char (size_t offset)
  52 {
  53   // We may need to iterate if the size of <n> is large.
  54   while (this->get_pos_ + offset >= this->end_of_mapping_plus1_)
  55     if (this->grow_file_and_remap () == -1)
  56       return EOF;
  57
  58   return this->get_pos_[offset];
  59 }
  60
  61 const char *
  62 ACEXML_Mem_Map_Stream::recv () const
  63 {
  64   return this->recv_pos_;
  65 }
  66
  67 const char *
  68 ACEXML_Mem_Map_Stream::recv (size_t &len)
  69 {
  70   if (this->eof () && this->grow_file_and_remap () == -1)
  71     {
  72       len = 0;
  73       return 0;
  74     }
  75   const char *s = this->recv_pos_;
  76   this->seek (static_cast<ACE_OFF_T> (len), SEEK_CUR);
  77   len = this->get_pos_ - s;
  78   return s;
  79 }
  80
  81 size_t
  82 ACEXML_Mem_Map_Stream::recv_len () const
  83 {
  84   return this->get_pos_ - this->recv_pos_;
  85 }
  86
  87 const char *
  88 ACEXML_Mem_Map_Stream::peek_str (size_t offset,
  89                                  size_t size)
  90 {
  91   // We will iterate if the size of <offset> is large.
  92   while (this->get_pos_ + (offset + size) > this->end_of_mapping_plus1_)
  93     if (this->grow_file_and_remap () == -1)
  94       return 0;
  95
  96   return &this->get_pos_[offset];
  97 }
  98
  99 ACE_OFF_T
 100 ACEXML_Mem_Map_Stream::seek (ACE_OFF_T offset, int whence)
 101 {
 102   switch (whence)
 103     {
 104     case SEEK_SET:
 105       this->get_pos_ =
 106         reinterpret_cast<char *> (this->mem_map_.addr ())
 107         + offset;
 108       break;
 109
 110     case SEEK_CUR:
 111       this->get_pos_ += offset;
 112       break;
 113
 114     case SEEK_END:
 115       this->get_pos_ =
 116         this->end_of_mapping_plus1_ + offset;
 117       // @@ Not sure how to implement this (yet).
 118       ACE_NOTSUP_RETURN (-1);
 119     }
 120
 121   // Make sure that the backing store will cover this.
 122   while (this->get_pos_ > this->end_of_mapping_plus1_)
 123     if (this->grow_file_and_remap () == -1)
 124       this->get_pos_ = this->end_of_mapping_plus1_;
 125
 126   this->recv_pos_ = this->get_pos_;
 127
 128   return
 129     ACE_Utils::truncate_cast<ACE_OFF_T> (
 130       this->recv_pos_ - reinterpret_cast<char *> (this->mem_map_.addr ()));
 131 }
 132
 133 Svc_Handler *
 134 ACEXML_Mem_Map_Stream::svc_handler ()
 135 {
 136   return this->svc_handler_;
 137 }
 138
 139 size_t
 140 ACEXML_Mem_Map_Stream::available () const
 141 {
 142   return this->end_of_mapping_plus1_ - this->get_pos_;
 143 }
 144
 145 int
 146 ACEXML_Mem_Map_Stream::open (Connector *connector,
 147                              const ACE_INET_Addr &addr)
 148 {
 149   svc_handler_ = 0;
 150
 151   // Connect to the server at <addr>. If the handler has to be
 152   // connected to the server again, the Caching strategy takes care
 153   // and uses the same connection.
 154   if (connector->connect (svc_handler_,
 155                           addr) == -1)
 156     {
 157       ACE_ERROR_RETURN ((LM_ERROR,
 158                          "%p %s %d\n",
 159                          "Connect failed",
 160                          addr.get_host_name (),
 161                          addr.get_port_number ()),
 162                         -1);
 163     }
 164   // Create a temporary filename.
 165   ACE_FILE_Addr file (ACE_sap_any_cast (ACE_FILE_Addr &));
 166
 167   // Create the temporary file via the <ACE_Mem_Map> class API.
 168   if (this->mem_map_.open (file.get_path_name (),
 169                            O_RDWR | O_CREAT | O_APPEND,
 170                            ACE_DEFAULT_FILE_PERMS) == -1)
 171     ACE_ERROR_RETURN ((LM_ERROR,
 172                        "%p\n",
 173                        "open"),
 174                       -1);
 175   // Make sure to unlink this right away so that if this process
 176   // crashes these files will be removed automatically.
 177   else if (ACE_OS::unlink (file.get_path_name ()) == -1)
 178     ACE_ERROR_RETURN ((LM_ERROR,
 179                        "%p\n",
 180                        "unlink"),
 181                       -1);
 182   else
 183     // Initialize all the position pointers to 0.
 184     this->rewind ();
 185
 186   return 0;
 187 }
 188
 189 int
 190 ACEXML_Mem_Map_Stream::grow_file_and_remap ()
 191 {
 192   char buf[8192];
 193
 194   // Copy the next chunk of bytes from the socket into the temporary
 195   // file.
 196   ACE_Time_Value tv (ACE_DEFAULT_TIMEOUT);
 197   ssize_t bytes = 0;
 198   while (1)
 199     {
 200       ssize_t const n = this->svc_handler_->peer ().recv (buf, sizeof buf, 0, &tv);
 201       if (n < 0)
 202         {
 203           if (errno != EWOULDBLOCK)
 204             {
 205               ACE_ERROR ((LM_ERROR, "%p\n", "recv"));
 206             }
 207
 208           return -1;
 209         }
 210       bytes += n;
 211       if (n == 0 && !bytes)
 212         return -1;
 213       else if (n == 0)
 214         break;
 215       else if (ACE::write_n (this->mem_map_.handle (), buf, n) != n)
 216         ACE_ERROR_RETURN ((LM_ERROR,
 217                            "%p\n",
 218                            "write_n"),
 219                           -1);
 220     }
 221
 222 //   ssize_t n = this->svc_handler_->peer ().recv (buf, sizeof buf, 0, &tv);
 223 //   if (n == -1)
 224 //     {
 225 //       ACE_ERROR ((LM_ERROR, "%p\n", "recv"));
 226 //       return -1;
 227 //     }
 228 //   else if (n == 0)
 229 //     return -1;
 230 //   else if (ACE::write_n (this->mem_map_.handle (), buf, n) != n)
 231 //         ACE_ERROR_RETURN ((LM_ERROR,
 232 //                            "%p\n",
 233 //                            "write_n"),
 234 //                           -1);
 235
 236   // Grow the memory-mapping to encompass the entire temporary file.
 237   if (this->mem_map_.map (static_cast<size_t> (-1),
 238                           PROT_RDWR,
 239                           ACE_MAP_PRIVATE,
 240                           (void*)0) == -1)
 241     ACE_ERROR_RETURN ((LM_ERROR,
 242                        "%p\n",
 243                        "map"),
 244                       -1);
 245   // MAP_FAILED is used as a "first time in" flag.
 246   if (this->recv_pos_ == MAP_FAILED)
 247     {
 248       this->recv_pos_ = reinterpret_cast<char *> (this->mem_map_.addr ());
 249       this->get_pos_ = this->recv_pos_;
 250     }
 251
 252   this->end_of_mapping_plus1_ =
 253     reinterpret_cast<char *> (this->mem_map_.addr ())
 254     + this->mem_map_.size ();
 255
 256   return 0;
 257 }
 258
 259 ACEXML_Mem_Map_Stream::~ACEXML_Mem_Map_Stream ()
 260 {
 261   // Remove the mapping and the file.
 262   this->mem_map_.remove ();
 263   delete this->svc_handler_;
 264 }
 265