8 "crawshaw.io/sqlite/sqlitex"
9 "github.com/anacrolix/torrent/metainfo"
10 "github.com/anacrolix/torrent/storage"
13 type NewDirectStorageOpts struct {
19 // A convenience function that creates a connection pool, resource provider, and a pieces storage
20 // ClientImpl and returns them all with a Close attached.
21 func NewDirectStorage(opts NewDirectStorageOpts) (_ storage.ClientImplCloser, err error) {
22 conn, err := newConn(opts.NewConnOpts)
26 err = initConn(conn, opts.InitConnOpts)
31 err = initDatabase(conn, opts.InitDbOpts)
37 blobs: make(map[string]*sqlite.Blob),
44 blobs map[string]*sqlite.Blob
47 func (c *client) OpenTorrent(info *metainfo.Info, infoHash metainfo.Hash) (storage.TorrentImpl, error) {
48 return torrent{c}, nil
51 func (c *client) Close() error {
52 for _, b := range c.blobs {
62 func rowidForBlob(c conn, name string, length int64) (rowid int64, err error) {
63 err = sqlitex.Exec(c, "select rowid from blob where name=?", func(stmt *sqlite.Stmt) error {
64 rowid = stmt.ColumnInt64(0)
73 err = sqlitex.Exec(c, "insert into blob(name, data) values(?, zeroblob(?))", nil, name, length)
77 rowid = c.LastInsertRowID()
81 func (t torrent) Piece(p metainfo.Piece) storage.PieceImpl {
84 name := p.Hash().HexString()
85 return piece{t.c.conn, &t.c.l, name, t.c.blobs, p.Length()}
88 func (t torrent) Close() error {
96 blobs map[string]*sqlite.Blob
100 func (p2 piece) doAtIoWithBlob(
101 atIo func(*sqlite.Blob) func([]byte, int64) (int, error),
104 ) (n int, err error) {
107 //defer p2.blobWouldExpire()
108 n, err = atIo(p2.getBlob())(p, off)
110 if !errors.As(err, &se) || se.Code != sqlite.SQLITE_ABORT {
114 return atIo(p2.getBlob())(p, off)
117 func (p2 piece) ReadAt(p []byte, off int64) (n int, err error) {
118 return p2.doAtIoWithBlob(func(blob *sqlite.Blob) func([]byte, int64) (int, error) {
123 func (p2 piece) WriteAt(p []byte, off int64) (n int, err error) {
124 return p2.doAtIoWithBlob(func(blob *sqlite.Blob) func([]byte, int64) (int, error) {
129 func (p2 piece) MarkComplete() error {
132 err := sqlitex.Exec(p2.conn, "update blob set verified=true where name=?", nil, p2.name)
136 changes := p2.conn.Changes()
143 func (p2 piece) blobWouldExpire() {
144 blob, ok := p2.blobs[p2.name]
149 delete(p2.blobs, p2.name)
152 func (p2 piece) MarkNotComplete() error {
153 return sqlitex.Exec(p2.conn, "update blob set verified=false where name=?", nil, p2.name)
156 func (p2 piece) Completion() (ret storage.Completion) {
159 err := sqlitex.Exec(p2.conn, "select verified from blob where name=?", func(stmt *sqlite.Stmt) error {
160 ret.Complete = stmt.ColumnInt(0) != 0
170 func (p2 piece) closeBlobIfExists() {
171 if b, ok := p2.blobs[p2.name]; ok {
173 delete(p2.blobs, p2.name)
177 func (p2 piece) getBlob() *sqlite.Blob {
178 blob, ok := p2.blobs[p2.name]
180 rowid, err := rowidForBlob(p2.conn, p2.name, p2.length)
184 blob, err = p2.conn.OpenBlob("main", "blob", "data", rowid, true)
188 p2.blobs[p2.name] = blob