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
  11         "bitbucket.org/anacrolix/go.torrent/util"
  12
  13         "bitbucket.org/anacrolix/go.torrent"
  14         "github.com/anacrolix/libtorgo/metainfo"
  15         "github.com/go-fsnotify/fsnotify"
  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        torrent.InfoHash
  30 }
  31
  32 type entity struct {
  33         torrent.InfoHash
  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[torrent.InfoHash]entity
  43 }
  44
  45 func (me *Instance) Close() {
  46         me.w.Close()
  47 }
  48
  49 func (me *Instance) handleEvents() {
  50         defer close(me.Events)
  51         for e := range me.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                         me.refresh()
  57                 }
  58         }
  59 }
  60
  61 func (me *Instance) handleErrors() {
  62         for err := range me.w.Errors {
  63                 log.Printf("error in torrent directory watcher: %s", err)
  64         }
  65 }
  66
  67 func torrentFileInfoHash(fileName string) (ih torrent.InfoHash, ok bool) {
  68         mi, _ := metainfo.LoadFromFile(fileName)
  69         if mi == nil {
  70                 return
  71         }
  72         util.CopyExact(ih[:], mi.Info.Hash)
  73         ok = true
  74         return
  75 }
  76
  77 func scanDir(dirName string) (ee map[torrent.InfoHash]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[torrent.InfoHash]entity, len(names))
  90         addEntity := func(e entity) {
  91                 e0, ok := ee[e.InfoHash]
  92                 if ok {
  93                         if e0.MagnetURI != "" && len(e.MagnetURI) < len(e0.MagnetURI) {
  94                                 return
  95                         }
  96                 }
  97                 ee[e.InfoHash] = 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                         util.CopyExact(e.InfoHash, 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 := torrent.ParseMagnetURI(uri)
 120                                 if err != nil {
 121                                         log.Print(err)
 122                                         continue
 123                                 }
 124                                 addEntity(entity{
 125                                         InfoHash:  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         for scanner.Scan() {
 142                 uris = append(uris, scanner.Text())
 143         }
 144         err = scanner.Err()
 145         return
 146 }
 147
 148 func (me *Instance) torrentRemoved(ih torrent.InfoHash) {
 149         me.Events <- Event{
 150                 InfoHash: ih,
 151                 Change:   Removed,
 152         }
 153 }
 154
 155 func (me *Instance) torrentAdded(e entity) {
 156         me.Events <- Event{
 157                 InfoHash:        e.InfoHash,
 158                 Change:          Added,
 159                 MagnetURI:       e.MagnetURI,
 160                 TorrentFilePath: e.TorrentFilePath,
 161         }
 162 }
 163
 164 func (me *Instance) refresh() {
 165         _new := scanDir(me.dirName)
 166         old := me.dirState
 167         for ih, _ := range old {
 168                 _, ok := _new[ih]
 169                 if !ok {
 170                         me.torrentRemoved(ih)
 171                 }
 172         }
 173         for ih, newE := range _new {
 174                 oldE, ok := old[ih]
 175                 if ok {
 176                         if newE == oldE {
 177                                 continue
 178                         }
 179                         me.torrentRemoved(ih)
 180                 }
 181                 me.torrentAdded(newE)
 182         }
 183         me.dirState = _new
 184 }
 185
 186 func New(dirName string) (i *Instance, err error) {
 187         w, err := fsnotify.NewWatcher()
 188         if err != nil {
 189                 return
 190         }
 191         err = w.Add(dirName)
 192         if err != nil {
 193                 w.Close()
 194                 return
 195         }
 196         i = &Instance{
 197                 w:        w,
 198                 dirName:  dirName,
 199                 Events:   make(chan Event),
 200                 dirState: make(map[torrent.InfoHash]entity, 0),
 201         }
 202         go func() {
 203                 i.refresh()
 204                 go i.handleEvents()
 205                 go i.handleErrors()
 206         }()
 207         return
 208 }