为 SQLite 的 BLOB 字段实现一个符合标准的流。

znone

2019-03-25 816f29a9c7ffd6696a03aa3da78bdf732ec9936c

 include/qtl_sqlite.hpp

@@ -70,7 +70,12 @@
      close();
      verify_error(sqlite3_prepare_v2(db, query_text, (int)text_length, &m_stmt, &tail));
      if(tail!=NULL)
         m_tail_text.assign(tail, query_text+text_length);
      {
         if(text_length==-1)
            m_tail_text.assign(tail);
         else
            m_tail_text.assign(tail, query_text+text_length);
      }
      else
         m_tail_text.clear();
   }
@@ -134,6 +139,10 @@
   {
      verify_error(sqlite3_bind_zeroblob(m_stmt, index+1, (int)n));
   }
   //void bind_zero_blob(int index, sqlite3_uint64 n)
   //{
   //   verify_error(sqlite3_bind_zeroblob64(m_stmt, index+1, (int)n));
   //}
   void bind_param(int index, qtl::null)
   {
      verify_error(sqlite3_bind_null(m_stmt, index+1));
@@ -484,6 +493,563 @@
   sqlite3* m_db;
};

// stream for blob field

class blobbuf : public std::streambuf
{
public:
   blobbuf() 
   {
      init();
   }
   blobbuf(const blobbuf&) = delete;
   blobbuf(blobbuf&& src) : std::streambuf(std::move(src))
   {
      init();
      swap(src);
   }
   virtual ~blobbuf()
   {
      if(m_blob)
      {
         close();
      }
   }

   blobbuf& operator=(const blobbuf&) = delete;
   blobbuf& operator=(blobbuf&& src)
   {
      if(this!=&src)
      {
         reset_back();
         close();
         swap(src);
      }
      return *this;
   }

   void swap( blobbuf& other )
   {
      std::swap(m_blob, other.m_blob);
      std::swap(m_inpos, other.m_inpos);
      std::swap(m_outpos, other.m_outpos);

      std::streambuf::swap(other);
      std::swap(m_back_char, other.m_back_char);
      if(eback() == &other.m_back_char)
         set_back();
      else
         reset_back();

      if(other.eback()==&m_back_char)
         other.set_back();
      else
         other.reset_back();
   }

   static void init_blob(database& db, const char* table, const char* column, int64_t row, int length)
   {
      statement stmt;
      std::ostringstream oss;
      oss<< "UPDATE " << table << " SET " << column << "=? WHERE rowid=?";
      stmt.open(db.handle(), oss.str().data());
      stmt.bind_zero_blob(0, length);
      stmt.bind_param(1, row);
      stmt.fetch();
   }
   static void init_blob(database& db, const std::string& table, const std::string& column, int64_t row, int length)
   {
      return init_blob(db, table.c_str(), column.c_str(), row, length);
   }

   bool is_open() const { return m_blob!=nullptr; }

   blobbuf* open(database& db, const char* table, const char* column, sqlite3_int64 row, 
      std::ios_base::openmode mode, const char* dbname="main")
   {
      int flags=0;
      if(mode&std::ios_base::out) flags=1;
      if(sqlite3_blob_open(db.handle(), dbname, table, column, row, flags, &m_blob)==SQLITE_OK)
      {
         m_size=sqlite3_blob_bytes(m_blob)/sizeof(char);
         // prepare buffer
         size_t bufsize=std::min<size_t>(default_buffer_size, m_size);
         if(mode&std::ios_base::in)
         {
            m_inbuf.resize(bufsize);
            m_inpos=0;
            setg(m_inbuf.data(), m_inbuf.data(), m_inbuf.data());
         }
         if(mode&std::ios_base::out)
         {
            m_outbuf.resize(bufsize);
            m_outpos=0;
            setp(m_outbuf.data(), m_outbuf.data()+bufsize);
         }
      }
      return this;
   }
   blobbuf* open(database& db, const std::string& table, const std::string& column, sqlite3_int64 row, 
      std::ios_base::openmode mode, const char* dbname="main")
   {
      return open(db, table.c_str(), column.c_str(), row, mode, dbname);
   }

   blobbuf* close()
   {
      if(m_blob==nullptr) 
         return nullptr;

      overflow();
      sqlite3_blob_close(m_blob);
      init();
      return this;
   }

   std::streamoff blob_size() const { return std::streamoff(m_size); }

   void flush() 
   {
      if(m_blob) 
         overflow();
   }

protected:
   enum { default_buffer_size = 4096 };

   virtual pos_type seekoff( off_type off, std::ios_base::seekdir dir,
      std::ios_base::openmode which = std::ios_base::in | std::ios_base::out ) override
   {
      if(!is_open())
         return pos_type(off_type(-1));

      pos_type pos=0;
      if(which&std::ios_base::out)
      {
         pos=seekoff(m_outpos, off, dir);
      }
      else if(which&std::ios_base::in)
      {
         pos=seekoff(m_inpos, off, dir);
      }
      return seekpos(pos, which);
   }

   virtual pos_type seekpos( pos_type pos,
      std::ios_base::openmode which = std::ios_base::in | std::ios_base::out) override
   {
      if(!is_open())
         return pos_type(off_type(-1));
      if(pos>=m_size)
         return pos_type(off_type(-1));

      if(which&std::ios_base::out)
      {
         if(pos<m_outpos || pos>=m_outpos+off_type(egptr()-pbase()))
         {
            overflow();
            m_outpos=pos;
            setp(m_outbuf.data(), m_outbuf.data()+m_outbuf.size());
         }
         else
         {
            pbump(off_type(pos-pabs()));
         }
      }
      else if(which&std::ios_base::in)
      {
         if(pos<m_inpos || pos>=m_inpos+off_type(epptr()-eback()))
         {
            m_inpos=pos;
            setg(m_inbuf.data(), m_inbuf.data(), m_inbuf.data());
         }
         else
         {
            gbump(off_type(pos-gabs()));
         }
      }
      return pos;
   }

   virtual std::streamsize showmanyc() override
   {
      return m_size-pabs();
   }

   //reads characters from the associated input sequence to the get area 
   virtual int_type underflow() override
   {
      if(!is_open()) 
         return traits_type::eof();

      if(pptr()>pbase())
         overflow();

      off_type count=egptr()-eback();
      pos_type next_pos;
      if(count==0 && eback()==m_inbuf.data())
      {
         setg(m_inbuf.data(), m_inbuf.data(), m_inbuf.data()+m_inbuf.size());
         count=m_inbuf.size();
      }
      else
      {
         next_pos=m_inpos+pos_type(count);
      }
      if(next_pos>=m_size)
         return traits_type::eof();

      count=std::min(count, m_size-next_pos);
      if(sqlite3_blob_read(m_blob, eback(), count, next_pos)!=SQLITE_OK)
         return traits_type::eof();
      m_inpos=next_pos;
      setg(eback(), eback(), eback()+count);
      return traits_type::to_int_type(*gptr());
   }

   /*//reads characters from the associated input sequence to the get area and advances the next pointer 
   virtual int_type uflow() override
   {

   }*/

   //writes characters to the associated output sequence from the put area 
   virtual int_type overflow( int_type ch = traits_type::eof() ) override
   {
      if(!is_open()) 
         return traits_type::eof();

      if(pptr()!=pbase())
      {
         size_t count = pptr()-pbase();
         if(sqlite3_blob_write(m_blob, pbase(), count, m_outpos)!=SQLITE_OK)
            return traits_type::eof();

         auto intersection = interval_intersection(m_inpos, egptr()-eback(), m_outpos, epptr()-pbase());
         if(intersection.first!=intersection.second)
         {
            commit(intersection.first, intersection.second);
         }

         m_outpos+=count;
         setp(pbase(), epptr());
      }
      if(!traits_type::eq_int_type(ch, traits_type::eof()))
      {
         char_type c = traits_type::to_char_type(ch);
         if(m_outpos>=m_size)
            return traits_type::eof();
         if(sqlite3_blob_write(m_blob, &c, 1, m_outpos)!=SQLITE_OK)
            traits_type::eof();
         auto intersection = interval_intersection(m_inpos, egptr()-eback(), m_outpos, 1);
         if(intersection.first!=intersection.second)
         {
            eback()[intersection.first-m_inpos]=c;
         }
         m_outpos+=1;
			
      }
      return ch;
   }

   virtual int_type pbackfail( int_type c = traits_type::eof() ) override
   {
      if (gptr() == 0
         || gptr() <= eback()
         || (!traits_type::eq_int_type(traits_type::eof(), c)
         && !traits_type::eq(traits_type::to_char_type(c), gptr()[-1])))
      {
         return (traits_type::eof());   // can't put back, fail
      }
      else
      {   // back up one position and store put-back character
         gbump(-1);
         if (!traits_type::eq_int_type(traits_type::eof(), c))
            *gptr() = traits_type::to_char_type(c);
         return (traits_type::not_eof(c));
      }	
   }

private:
   sqlite3_blob* m_blob;
   std::vector<char> m_inbuf;
   std::vector<char> m_outbuf;
   pos_type m_size;
   pos_type m_inpos;   //position in the input sequence
   pos_type m_outpos;   //position in the output sequence

   void init()
   {
      m_blob=nullptr;
      m_size=0;
      m_inpos=m_outpos=0;
      m_set_eback=m_set_egptr=nullptr;
      m_back_char=0;
      setg(nullptr, nullptr, nullptr);
      setp(nullptr, nullptr);
   }

   off_type seekoff(off_type position, off_type off, std::ios_base::seekdir dir)
   {
      off_type result=0;
      switch(dir)
      {
      case std::ios_base::beg:
         result=off;
      case std::ios_base::cur:
         result=position+off;
      case std::ios_base::end:
         result=m_size-off;
      }
      if(result>m_size)
         result=m_size;
      return result;
   }

   void reset_back()
   {   // restore buffer after putback
      if (this->eback() == &m_back_char)
         this->setg(m_set_eback, m_set_eback, m_set_egptr);
   }

   void set_back()
   {   // set up putback area
      if (this->eback() != &m_back_char)
      {   // save current get buffer
         m_set_eback = this->eback();
         m_set_egptr = this->egptr();
      }
      this->setg(&m_back_char, &m_back_char, &m_back_char + 1);
   }

   char_type *m_set_eback { nullptr };   // saves eback() during one-element putback
   char_type *m_set_egptr { nullptr };   // saves egptr()
   char_type m_back_char { 0 };

   void move_data(blobbuf& other)
   {
      m_blob=other.m_blob;
      other.m_blob=nullptr;
      m_size=other.m_size;
      other.m_size=0;
      m_inpos=other.m_inpos;
      other.m_inpos=0;
      m_outpos=other.m_outpos;
      other.m_outpos=0;
   }

   static std::pair<pos_type, pos_type> interval_intersection(pos_type first1, pos_type last1, pos_type first2, pos_type last2)
   {
      if(first1>first2)
      {
         std::swap(first1, first2);
         std::swap(last1, last2);
      }

      if(first2<last1)
         return std::make_pair(first2, std::min(last1, last2));
      else
         return std::make_pair(0, 0);
   }

   static std::pair<pos_type, pos_type> interval_intersection(pos_type first1, off_type count1, pos_type first2, off_type count2)
   {
      return interval_intersection(first1, first1+count1, first2, first2+count2);
   }

   void commit(off_type first, off_type last)
   {
      char_type* src= pbase()+(first-m_outpos);
      char_type* dest = eback()+(first-m_inpos);
      memmove(dest, src, last-first);
   }

   pos_type gabs() const // absolute offset of input pointer in blob field
   {
      return m_inpos+off_type(gptr()-eback());
   }

   pos_type pabs() const // absolute offset of output pointer in blob field
   {
      return m_outpos+off_type(pptr()-pbase());
   }
};

class iblobstream : public std::istream
{
public:
   iblobstream() : std::istream(&m_buffer) { }
   iblobstream(database& db, const char* table, const char* column, sqlite3_int64 row, 
      std::ios_base::openmode mode=std::ios_base::in, const char* dbname="main")
      : std::istream(&m_buffer)
   {
      open(db, table, column, row, mode, dbname);
   }
   iblobstream(database& db, const std::string& table, const std::string& column, sqlite3_int64 row, 
      std::ios_base::openmode mode=std::ios_base::in, const char* dbname="main")
      : std::istream(&m_buffer)
   {
      open(db, table, column, row, mode, dbname);
   }
   iblobstream(const iblobstream&) = delete;
   iblobstream(iblobstream&& src) : std::istream(&m_buffer), m_buffer(std::move(src.m_buffer)) { }

   iblobstream& operator=(const iblobstream&) = delete;
   iblobstream& operator=(iblobstream&& src) 
   {
      m_buffer.operator =(std::move(src.m_buffer));
   }

   bool is_open() const { return m_buffer.is_open(); }

   void open(database& db, const char* table, const char* column, sqlite3_int64 row, 
      std::ios_base::openmode mode=std::ios_base::in, const char* dbname="main")
   {
      if(m_buffer.open(db, table, column, row, mode|std::ios_base::in, dbname)==nullptr)
         this->setstate(std::ios_base::failbit);
      else
         this->clear();
   }
   void open(database& db, const std::string& table, const std::string& column, sqlite3_int64 row, 
      std::ios_base::openmode mode=std::ios_base::in, const char* dbname="main")
   {
      open(db, table.c_str(), column.c_str(), row, mode, dbname);
   }

   void close()
   {
      if(m_buffer.close()==nullptr)
         this->setstate(std::ios_base::failbit);
   }

   blobbuf* rdbuf() const
   {
      return const_cast<blobbuf*>(&m_buffer);
   }

   std::streamoff blob_size() const { return m_buffer.blob_size(); }

private:
   blobbuf m_buffer;
};

class oblobstream : public std::ostream
{
public:
   oblobstream() : std::ostream(&m_buffer) { }
   oblobstream(database& db, const char* table, const char* column, sqlite3_int64 row, 
      std::ios_base::openmode mode=std::ios_base::in, const char* dbname="main")
      : std::ostream(&m_buffer)
   {
      open(db, table, column, row, mode, dbname);
   }
   oblobstream(database& db, const std::string& table, const std::string& column, sqlite3_int64 row, 
      std::ios_base::openmode mode=std::ios_base::in, const char* dbname="main")
      : std::ostream(&m_buffer)
   {
      open(db, table, column, row, mode, dbname);
   }
   oblobstream(const oblobstream&) = delete;
   oblobstream(oblobstream&& src) : std::ostream(&m_buffer), m_buffer(std::move(src.m_buffer)) { }

   oblobstream& operator=(const oblobstream&) = delete;
   oblobstream& operator=(oblobstream&& src) 
   {
      m_buffer.operator =(std::move(src.m_buffer));
   }

   bool is_open() const { return m_buffer.is_open(); }

   void open(database& db, const char* table, const char* column, sqlite3_int64 row, 
      std::ios_base::openmode mode=std::ios_base::out, const char* dbname="main")
   {
      if(m_buffer.open(db, table, column, row, mode|std::ios_base::out, dbname)==nullptr)
         this->setstate(std::ios_base::failbit);
      else
         this->clear();
   }
   void open(database& db, const std::string& table, const std::string& column, sqlite3_int64 row, 
      std::ios_base::openmode mode=std::ios_base::out, const char* dbname="main")
   {
      open(db, table.c_str(), column.c_str(), row, mode, dbname);
   }

   void close()
   {
      if(m_buffer.close()==nullptr)
         this->setstate(std::ios_base::failbit);
   }

   blobbuf* rdbuf() const
   {
      return const_cast<blobbuf*>(&m_buffer);
   }

   std::streamoff blob_size() const { return m_buffer.blob_size(); }

private:
   blobbuf m_buffer;
};

class blobstream : public std::iostream
{
public:
   blobstream() : std::iostream(&m_buffer) { }
   blobstream(database& db, const char* table, const char* column, sqlite3_int64 row, 
      std::ios_base::openmode mode=std::ios_base::in, const char* dbname="main")
      : std::iostream(&m_buffer) 
   {
      open(db, table, column, row, mode, dbname);
   }
   blobstream(database& db, const std::string& table, const std::string& column, sqlite3_int64 row, 
      std::ios_base::openmode mode=std::ios_base::in, const char* dbname="main")
      : std::iostream(&m_buffer) 
   {
      open(db, table, column, row, mode, dbname);
   }
   blobstream(const blobstream&) = delete;
   blobstream(blobstream&& src) : std::iostream(&m_buffer), m_buffer(std::move(src.m_buffer)) { }

   blobstream& operator=(const blobstream&) = delete;
   blobstream& operator=(blobstream&& src) 
   {
      m_buffer.operator =(std::move(src.m_buffer));
   }

   bool is_open() const { return m_buffer.is_open(); }

   void open(database& db, const char* table, const char* column, sqlite3_int64 row, 
      std::ios_base::openmode mode=std::ios_base::in|std::ios_base::out, const char* dbname="main")
   {
      if(m_buffer.open(db, table, column, row, mode|std::ios_base::in|std::ios_base::out, dbname)==nullptr)
         this->setstate(std::ios_base::failbit);
      else
         this->clear();
   }
   void open(database& db, const std::string& table, const std::string& column, sqlite3_int64 row, 
      std::ios_base::openmode mode=std::ios_base::in|std::ios_base::out, const char* dbname="main")
   {
      open(db, table.c_str(), column.c_str(), row, mode, dbname);
   }

   void close()
   {
      if(m_buffer.close()==nullptr)
         this->setstate(std::ios_base::failbit);
   }

   blobbuf* rdbuf() const
   {
      return const_cast<blobbuf*>(&m_buffer);
   }

   std::streamoff blob_size() const { return m_buffer.blob_size(); }

private:
   blobbuf m_buffer;
};


typedef qtl::transaction<database> transaction;

template<typename Record>