storage/sqlite/direct.go

   1 //go:build cgo
   2 // +build cgo
   3
   4 package sqliteStorage
   5
   6 import (
   7         "encoding/hex"
   8         "io"
   9         "sync"
  10         "time"
  11
  12         g "github.com/anacrolix/generics"
  13         "github.com/anacrolix/squirrel"
  14
  15         "github.com/anacrolix/torrent/metainfo"
  16         "github.com/anacrolix/torrent/storage"
  17 )
  18
  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         cache, err := squirrel.NewCache(opts)
  23         if err != nil {
  24                 return
  25         }
  26         return &client{
  27                 cache: cache,
  28         }, nil
  29 }
  30
  31 // Creates a storage.ClientImpl from a provided squirrel.Cache. The caller is responsible for
  32 // closing the squirrel.Cache.
  33 func NewWrappingClient(cache *squirrel.Cache) storage.ClientImpl {
  34         return &client{
  35                 cache: cache,
  36         }
  37 }
  38
  39 type client struct {
  40         cache           *squirrel.Cache
  41         capacityMu      sync.Mutex
  42         capacityFetched time.Time
  43         capacityCap     int64
  44         capacityCapped  bool
  45 }
  46
  47 func (c *client) Close() error {
  48         return c.cache.Close()
  49 }
  50
  51 func (c *client) capacity() (cap int64, capped bool) {
  52         c.capacityMu.Lock()
  53         defer c.capacityMu.Unlock()
  54         if !c.capacityFetched.IsZero() && time.Since(c.capacityFetched) < time.Second {
  55                 cap, capped = c.capacityCap, c.capacityCapped
  56                 return
  57         }
  58         c.capacityCap, c.capacityCapped = c.cache.GetCapacity()
  59         // Should this go before or after the capacityCap and capacityCapped assignments?
  60         c.capacityFetched = time.Now()
  61         cap, capped = c.capacityCap, c.capacityCapped
  62         return
  63 }
  64
  65 func (c *client) OpenTorrent(*metainfo.Info, metainfo.Hash) (storage.TorrentImpl, error) {
  66         t := torrent{c.cache}
  67         capFunc := c.capacity
  68         return storage.TorrentImpl{PieceWithHash: t.Piece, Close: t.Close, Capacity: &capFunc}, nil
  69 }
  70
  71 type torrent struct {
  72         c *squirrel.Cache
  73 }
  74
  75 func (t torrent) Piece(p metainfo.Piece, pieceHash g.Option[[]byte]) storage.PieceImpl {
  76         ret := piece{
  77                 sb: t.c.OpenWithLength(hex.EncodeToString(pieceHash.Unwrap()), p.Length()),
  78         }
  79         ret.ReaderAt = &ret.sb
  80         ret.WriterAt = &ret.sb
  81         return ret
  82 }
  83
  84 func (t torrent) Close() error {
  85         return nil
  86 }
  87
  88 type piece struct {
  89         sb squirrel.Blob
  90         io.ReaderAt
  91         io.WriterAt
  92 }
  93
  94 func (p piece) MarkComplete() error {
  95         return p.sb.SetTag("verified", true)
  96 }
  97
  98 func (p piece) MarkNotComplete() error {
  99         return p.sb.SetTag("verified", false)
 100 }
 101
 102 func (p piece) Completion() (ret storage.Completion) {
 103         err := p.sb.GetTag("verified", func(stmt squirrel.SqliteStmt) {
 104                 ret.Complete = stmt.ColumnInt(0) != 0
 105         })
 106         ret.Ok = err == nil
 107         ret.Err = err
 108         return
 109 }