source/io/buffered.cpp

   1 #include <cstring>
   2 #include <msp/core/except.h>
   3 #include "buffered.h"
   4 #include "handle.h"
   5
   6 using namespace std;
   7
   8 namespace Msp {
   9 namespace IO {
  10
  11 Buffered::Buffered(Base &b, unsigned s):
  12         below(b),
  13         buf_size(s),
  14         buf(new char[buf_size]),
  15         begin(buf),
  16         end(buf)
  17 {
  18         mode = below.get_mode();
  19         below.signal_flush_required.connect(sigc::mem_fun(this, &Buffered::flush));
  20 }
  21
  22 Buffered::~Buffered()
  23 {
  24         try
  25         {
  26                 flush();
  27         }
  28         catch(...)
  29         { }
  30
  31         delete[] buf;
  32 }
  33
  34 void Buffered::set_block(bool)
  35 {
  36         throw unsupported("Buffered::set_block");
  37 }
  38
  39 void Buffered::set_inherit(bool)
  40 {
  41         throw unsupported("Buffered::set_block");
  42 }
  43
  44 void Buffered::flush()
  45 {
  46         if(cur_op==M_WRITE)
  47         {
  48                 unsigned used = end-begin;
  49                 if(used)
  50                 {
  51                         unsigned len = below.write(begin, used);
  52
  53                         begin = end = buf;
  54
  55                         if(len<used)
  56                                 throw runtime_error("Couldn't flush all data");
  57                 }
  58         }
  59         else if(cur_op==M_READ)
  60                 begin = end = buf;
  61 }
  62
  63 size_t Buffered::do_write(const char *data, size_t size)
  64 {
  65         set_op(M_WRITE);
  66
  67         if(end+size<buf+buf_size)
  68         {
  69                 // All data fits in buffer with whatever is already there
  70                 memcpy(end, data, size);
  71                 end += size;
  72
  73                 return size;
  74         }
  75         else
  76         {
  77                 // Clear the buffer to make more room
  78                 flush();
  79
  80                 if(size<buf_size)
  81                 {
  82                         // Put new data in the buffer to wait for more
  83                         memcpy(end, data, size);
  84                         end += size;
  85
  86                         return size;
  87                 }
  88                 else
  89                         // New data still doesn't fit in the buffer, so write it directly
  90                         return below.write(data, size);
  91         }
  92 }
  93
  94 size_t Buffered::do_read(char *data, size_t size)
  95 {
  96         set_op(M_READ);
  97
  98         if(begin+size<=end)
  99         {
 100                 // The request can be served from the buffer
 101                 memcpy(data, begin, size);
 102                 begin += size;
 103
 104                 eof_flag = (below.eof() && begin==end);
 105
 106                 return size;
 107         }
 108         else
 109         {
 110                 // Give out whatever is in the buffer already
 111                 memcpy(data, begin, end-begin);
 112                 size_t ret = end-begin;
 113                 begin = end = buf;
 114
 115                 data += ret;
 116                 size -= ret;
 117
 118                 if(size<buf_size)
 119                 {
 120                         // Fill the buffer and serve the rest of the request from it
 121                         size_t len = below.read(end, buf+buf_size-end);
 122                         end += len;
 123
 124                         len = min<size_t>(end-begin, size);
 125                         memcpy(data, begin, len);
 126                         begin += len;
 127                         ret += len;
 128                 }
 129                 else
 130                         // Read the rest directly from the underlying object
 131                         ret += below.read(data, size);
 132
 133                 eof_flag = (below.eof() && begin==end);
 134
 135                 return ret;
 136         }
 137 }
 138
 139 size_t Buffered::put(char c)
 140 {
 141         set_op(M_WRITE);
 142
 143         if(end<buf+buf_size)
 144         {
 145                 *end++ = c;
 146                 return 1;
 147         }
 148         else
 149                 return do_write(&c, 1);
 150 }
 151
 152 bool Buffered::getline(string &line)
 153 {
 154         set_op(M_READ);
 155
 156         for(char *i=begin; i!=end; ++i)
 157                 if(*i=='\n')
 158                 {
 159                         line.assign(begin, i-begin);
 160                         begin = i+1;
 161                         return true;
 162                 }
 163
 164         return Base::getline(line);
 165 }
 166
 167 int Buffered::get()
 168 {
 169         set_op(M_READ);
 170
 171         if(begin<end)
 172                 return static_cast<unsigned char>(*begin++);
 173
 174         char c;
 175         if(do_read(&c, 1)==0)
 176                 return -1;
 177         return static_cast<unsigned char>(c);
 178 }
 179
 180 const Handle &Buffered::get_handle(Mode)
 181 {
 182         throw unsupported("Buffered::get_handle");
 183 }
 184
 185 void Buffered::set_op(Mode op)
 186 {
 187         if(op!=cur_op)
 188                 flush();
 189         cur_op = op;
 190 }
 191
 192 size_t Buffered::get_current_size() const
 193 {
 194         return end-begin;
 195 }
 196
 197 } // namespace IO
 198 } // namespace Msp