storage/piece_resource.go

   1 package storage
   2
   3 import (
   4         "bytes"
   5         "io"
   6         "path"
   7         "sort"
   8         "strconv"
   9
  10         "github.com/anacrolix/missinggo/v2/resource"
  11
  12         "github.com/anacrolix/torrent/metainfo"
  13 )
  14
  15 type piecePerResource struct {
  16         p resource.Provider
  17 }
  18
  19 func NewResourcePieces(p resource.Provider) ClientImpl {
  20         return &piecePerResource{
  21                 p: p,
  22         }
  23 }
  24
  25 func (s *piecePerResource) OpenTorrent(info *metainfo.Info, infoHash metainfo.Hash) (TorrentImpl, error) {
  26         return s, nil
  27 }
  28
  29 func (s *piecePerResource) Close() error {
  30         return nil
  31 }
  32
  33 func (s *piecePerResource) Piece(p metainfo.Piece) PieceImpl {
  34         return piecePerResourcePiece{
  35                 mp: p,
  36                 rp: s.p,
  37         }
  38 }
  39
  40 type piecePerResourcePiece struct {
  41         mp metainfo.Piece
  42         rp resource.Provider
  43 }
  44
  45 func (s piecePerResourcePiece) Completion() Completion {
  46         fi, err := s.completed().Stat()
  47         return Completion{
  48                 Complete: err == nil && fi.Size() == s.mp.Length(),
  49                 Ok:       true,
  50         }
  51 }
  52
  53 func (s piecePerResourcePiece) MarkComplete() error {
  54         incompleteChunks := s.getChunks()
  55         err := s.completed().Put(io.NewSectionReader(incompleteChunks, 0, s.mp.Length()))
  56         if err == nil {
  57                 for _, c := range incompleteChunks {
  58                         c.instance.Delete()
  59                 }
  60         }
  61         return err
  62 }
  63
  64 func (s piecePerResourcePiece) MarkNotComplete() error {
  65         return s.completed().Delete()
  66 }
  67
  68 func (s piecePerResourcePiece) ReadAt(b []byte, off int64) (int, error) {
  69         if s.Completion().Complete {
  70                 return s.completed().ReadAt(b, off)
  71         }
  72         return s.getChunks().ReadAt(b, off)
  73 }
  74
  75 func (s piecePerResourcePiece) WriteAt(b []byte, off int64) (n int, err error) {
  76         i, err := s.rp.NewInstance(path.Join(s.incompleteDirPath(), strconv.FormatInt(off, 10)))
  77         if err != nil {
  78                 panic(err)
  79         }
  80         r := bytes.NewReader(b)
  81         err = i.Put(r)
  82         n = len(b) - r.Len()
  83         return
  84 }
  85
  86 type chunk struct {
  87         offset   int64
  88         instance resource.Instance
  89 }
  90
  91 type chunks []chunk
  92
  93 func (me chunks) ReadAt(b []byte, off int64) (int, error) {
  94         for {
  95                 if len(me) == 0 {
  96                         return 0, io.EOF
  97                 }
  98                 if me[0].offset <= off {
  99                         break
 100                 }
 101                 me = me[1:]
 102         }
 103         n, err := me[0].instance.ReadAt(b, off-me[0].offset)
 104         if n == len(b) {
 105                 return n, nil
 106         }
 107         if err == nil || err == io.EOF {
 108                 n_, err := me[1:].ReadAt(b[n:], off+int64(n))
 109                 return n + n_, err
 110         }
 111         return n, err
 112 }
 113
 114 func (s piecePerResourcePiece) getChunks() (chunks chunks) {
 115         names, err := s.incompleteDir().Readdirnames()
 116         if err != nil {
 117                 return
 118         }
 119         for _, n := range names {
 120                 offset, err := strconv.ParseInt(n, 10, 64)
 121                 if err != nil {
 122                         continue
 123                 }
 124                 i, err := s.rp.NewInstance(path.Join(s.incompleteDirPath(), n))
 125                 if err != nil {
 126                         panic(err)
 127                 }
 128                 chunks = append(chunks, chunk{offset, i})
 129         }
 130         sort.Slice(chunks, func(i, j int) bool {
 131                 return chunks[i].offset < chunks[j].offset
 132         })
 133         return
 134 }
 135
 136 func (s piecePerResourcePiece) completed() resource.Instance {
 137         i, err := s.rp.NewInstance(path.Join("completed", s.mp.Hash().HexString()))
 138         if err != nil {
 139                 panic(err)
 140         }
 141         return i
 142 }
 143
 144 func (s piecePerResourcePiece) incompleteDirPath() string {
 145         return path.Join("incompleted", s.mp.Hash().HexString())
 146 }
 147
 148 func (s piecePerResourcePiece) incompleteDir() resource.DirInstance {
 149         i, err := s.rp.NewInstance(s.incompleteDirPath())
 150         if err != nil {
 151                 panic(err)
 152         }
 153         return i.(resource.DirInstance)
 154 }