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

znone

2019-03-25 816f29a9c7ffd6696a03aa3da78bdf732ec9936c

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

 README.md

@@ -250,6 +250,24 @@
- qtl::sqlite::query_result
表示一个SQLite的查询结果集，用于以迭代器方式遍历查询结果。

### SQLite的Blob字段

通过QTL，可以通过标准流的方式访问SQLite的BLOB字段。
下面的代码，先用数字0-9向BLOB字段填充，然后再次读取字段内容并显示到屏幕。

```C++
int64_t id=db->insert("INSERT INTO test_blob (Filename, Content, MD5) values(?, ?, ?)",
    forward_as_tuple("sample", qtl::const_blob_data(nullptr, 1024), nullptr));

qtl::sqlite::blobstream bs(*db, "test_blob", "Content", id);
generate_n(ostreambuf_iterator<char>(bs), bs.blob_size()/sizeof(char), [i=0]() mutable { 
    return char('0'+(i++)%10);
});
copy(istream_iterator<char>(bs), istream_iterator<char>(), ostream_iterator<char>(cout, nullptr));
cout<<endl;

```

## 有关ODBC的说明

通过ODBC访问数据库时，包含头文件qtl_odbc.hpp。

 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)
        {
            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>