util/dirwatch/dirwatch.go

   1 // Package dirwatch provides filesystem-notification based tracking of torrent
   2 // info files and magnet URIs in a directory.
   3 package dirwatch
   4
   5 import (
   6         "bufio"
   7         "os"
   8         "path/filepath"
   9         "strings"
  10
  11         "github.com/anacrolix/log"
  12         "github.com/anacrolix/missinggo/v2"
  13         "github.com/fsnotify/fsnotify"
  14
  15         "github.com/anacrolix/torrent/metainfo"
  16 )
  17
  18 type Change uint
  19
  20 const (
  21         Added Change = iota
  22         Removed
  23 )
  24
  25 type Event struct {
  26         MagnetURI string
  27         Change
  28         TorrentFilePath string
  29         InfoHash        metainfo.Hash
  30 }
  31
  32 type entity struct {
  33         metainfo.Hash
  34         MagnetURI       string
  35         TorrentFilePath string
  36 }
  37
  38 type Instance struct {
  39         w        *fsnotify.Watcher
  40         dirName  string
  41         Events   chan Event
  42         dirState map[metainfo.Hash]entity
  43         Logger   log.Logger
  44 }
  45
  46 func (i *Instance) Close() {
  47         i.w.Close()
  48 }
  49
  50 func (i *Instance) handleEvents() {
  51         defer close(i.Events)
  52         for e := range i.w.Events {
  53                 i.Logger.WithDefaultLevel(log.Debug).Printf("event: %v", e)
  54                 if e.Op == fsnotify.Write {
  55                         // TODO: Special treatment as an existing torrent may have changed.
  56                 } else {
  57                         i.refresh()
  58                 }
  59         }
  60 }
  61
  62 func (i *Instance) handleErrors() {
  63         for err := range i.w.Errors {
  64                 log.Printf("error in torrent directory watcher: %s", err)
  65         }
  66 }
  67
  68 func torrentFileInfoHash(fileName string) (ih metainfo.Hash, ok bool) {
  69         mi, _ := metainfo.LoadFromFile(fileName)
  70         if mi == nil {
  71                 return
  72         }
  73         ih = mi.HashInfoBytes()
  74         ok = true
  75         return
  76 }
  77
  78 func scanDir(dirName string) (ee map[metainfo.Hash]entity) {
  79         d, err := os.Open(dirName)
  80         if err != nil {
  81                 log.Print(err)
  82                 return
  83         }
  84         defer d.Close()
  85         names, err := d.Readdirnames(-1)
  86         if err != nil {
  87                 log.Print(err)
  88                 return
  89         }
  90         ee = make(map[metainfo.Hash]entity, len(names))
  91         addEntity := func(e entity) {
  92                 e0, ok := ee[e.Hash]
  93                 if ok {
  94                         if e0.MagnetURI == "" || len(e.MagnetURI) < len(e0.MagnetURI) {
  95                                 return
  96                         }
  97                 }
  98                 ee[e.Hash] = e
  99         }
 100         for _, n := range names {
 101                 fullName := filepath.Join(dirName, n)
 102                 switch filepath.Ext(n) {
 103                 case ".torrent":
 104                         ih, ok := torrentFileInfoHash(fullName)
 105                         if !ok {
 106                                 break
 107                         }
 108                         e := entity{
 109                                 TorrentFilePath: fullName,
 110                         }
 111                         missinggo.CopyExact(&e.Hash, ih)
 112                         addEntity(e)
 113                 case ".magnet":
 114                         uris, err := magnetFileURIs(fullName)
 115                         if err != nil {
 116                                 log.Print(err)
 117                                 break
 118                         }
 119                         for _, uri := range uris {
 120                                 m, err := metainfo.ParseMagnetUri(uri)
 121                                 if err != nil {
 122                                         log.Printf("error parsing %q in file %q: %s", uri, fullName, err)
 123                                         continue
 124                                 }
 125                                 addEntity(entity{
 126                                         Hash:      m.InfoHash,
 127                                         MagnetURI: uri,
 128                                 })
 129                         }
 130                 }
 131         }
 132         return
 133 }
 134
 135 func magnetFileURIs(name string) (uris []string, err error) {
 136         f, err := os.Open(name)
 137         if err != nil {
 138                 return
 139         }
 140         defer f.Close()
 141         scanner := bufio.NewScanner(f)
 142         scanner.Split(bufio.ScanWords)
 143         for scanner.Scan() {
 144                 // Allow magnet URIs to be "commented" out.
 145                 if strings.HasPrefix(scanner.Text(), "#") {
 146                         continue
 147                 }
 148                 uris = append(uris, scanner.Text())
 149         }
 150         err = scanner.Err()
 151         return
 152 }
 153
 154 func (i *Instance) torrentRemoved(ih metainfo.Hash) {
 155         i.Events <- Event{
 156                 InfoHash: ih,
 157                 Change:   Removed,
 158         }
 159 }
 160
 161 func (i *Instance) torrentAdded(e entity) {
 162         i.Events <- Event{
 163                 InfoHash:        e.Hash,
 164                 Change:          Added,
 165                 MagnetURI:       e.MagnetURI,
 166                 TorrentFilePath: e.TorrentFilePath,
 167         }
 168 }
 169
 170 func (i *Instance) refresh() {
 171         _new := scanDir(i.dirName)
 172         old := i.dirState
 173         for ih := range old {
 174                 _, ok := _new[ih]
 175                 if !ok {
 176                         i.torrentRemoved(ih)
 177                 }
 178         }
 179         for ih, newE := range _new {
 180                 oldE, ok := old[ih]
 181                 if ok {
 182                         if newE == oldE {
 183                                 continue
 184                         }
 185                         i.torrentRemoved(ih)
 186                 }
 187                 i.torrentAdded(newE)
 188         }
 189         i.dirState = _new
 190 }
 191
 192 func New(dirName string) (i *Instance, err error) {
 193         w, err := fsnotify.NewWatcher()
 194         if err != nil {
 195                 return
 196         }
 197         err = w.Add(dirName)
 198         if err != nil {
 199                 w.Close()
 200                 return
 201         }
 202         i = &Instance{
 203                 w:        w,
 204                 dirName:  dirName,
 205                 Events:   make(chan Event),
 206                 dirState: make(map[metainfo.Hash]entity),
 207                 Logger:   log.Default,
 208         }
 209         go func() {
 210                 i.refresh()
 211                 go i.handleEvents()
 212                 go i.handleErrors()
 213         }()
 214         return
 215 }