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