webseed-peer.go

   1 package torrent
   2
   3 import (
   4         "fmt"
   5         "strings"
   6         "sync"
   7
   8         "github.com/anacrolix/torrent/common"
   9         "github.com/anacrolix/torrent/metainfo"
  10         pp "github.com/anacrolix/torrent/peer_protocol"
  11         "github.com/anacrolix/torrent/segments"
  12         "github.com/anacrolix/torrent/webseed"
  13         "github.com/pkg/errors"
  14 )
  15
  16 type webseedPeer struct {
  17         client         webseed.Client
  18         activeRequests map[Request]webseed.Request
  19         requesterCond  sync.Cond
  20         peer           Peer
  21 }
  22
  23 var _ peerImpl = (*webseedPeer)(nil)
  24
  25 func (me *webseedPeer) connStatusString() string {
  26         return me.client.Url
  27 }
  28
  29 func (ws *webseedPeer) String() string {
  30         return fmt.Sprintf("webseed peer for %q", ws.client.Url)
  31 }
  32
  33 func (ws *webseedPeer) onGotInfo(info *metainfo.Info) {
  34         ws.client.FileIndex = segments.NewIndex(common.LengthIterFromUpvertedFiles(info.UpvertedFiles()))
  35         ws.client.Info = info
  36 }
  37
  38 func (ws *webseedPeer) _postCancel(r Request) {
  39         ws.cancel(r)
  40 }
  41
  42 func (ws *webseedPeer) writeInterested(interested bool) bool {
  43         return true
  44 }
  45
  46 func (ws *webseedPeer) cancel(r Request) bool {
  47         active, ok := ws.activeRequests[r]
  48         if !ok {
  49                 return false
  50         }
  51         active.Cancel()
  52         return true
  53 }
  54
  55 func (ws *webseedPeer) intoSpec(r Request) webseed.RequestSpec {
  56         return webseed.RequestSpec{ws.peer.t.requestOffset(r), int64(r.Length)}
  57 }
  58
  59 func (ws *webseedPeer) request(r Request) bool {
  60         ws.requesterCond.Signal()
  61         return true
  62 }
  63
  64 func (ws *webseedPeer) doRequest(r Request) {
  65         webseedRequest := ws.client.NewRequest(ws.intoSpec(r))
  66         ws.activeRequests[r] = webseedRequest
  67         ws.requesterCond.L.Unlock()
  68         ws.requestResultHandler(r, webseedRequest)
  69         ws.requesterCond.L.Lock()
  70         delete(ws.activeRequests, r)
  71 }
  72
  73 func (ws *webseedPeer) requester() {
  74         ws.requesterCond.L.Lock()
  75         defer ws.requesterCond.L.Unlock()
  76 start:
  77         for !ws.peer.closed.IsSet() {
  78                 for r := range ws.peer.requests {
  79                         if _, ok := ws.activeRequests[r]; ok {
  80                                 continue
  81                         }
  82                         ws.doRequest(r)
  83                         goto start
  84                 }
  85                 ws.requesterCond.Wait()
  86         }
  87 }
  88
  89 func (ws *webseedPeer) connectionFlags() string {
  90         return "WS"
  91 }
  92
  93 // TODO: This is called when banning peers. Perhaps we want to be able to ban webseeds too. We could
  94 // return bool if this is even possible, and if it isn't, skip to the next drop candidate.
  95 func (ws *webseedPeer) drop() {}
  96
  97 func (ws *webseedPeer) updateRequests() {
  98         ws.peer.doRequestState()
  99 }
 100
 101 func (ws *webseedPeer) onClose() {
 102         ws.requesterCond.Broadcast()
 103 }
 104
 105 func (ws *webseedPeer) requestResultHandler(r Request, webseedRequest webseed.Request) {
 106         result := <-webseedRequest.Result
 107         ws.peer.t.cl.lock()
 108         defer ws.peer.t.cl.unlock()
 109         if result.Err != nil {
 110                 ws.peer.logger.Printf("Request %v rejected: %v", r, result.Err)
 111                 // Always close for now. We need to filter out temporary errors, but this is a nightmare in
 112                 // Go. Currently a bad webseed URL can starve out the good ones due to the chunk selection
 113                 // algorithm.
 114                 const closeOnAllErrors = false
 115                 if closeOnAllErrors || strings.Contains(errors.Cause(result.Err).Error(), "unsupported protocol scheme") {
 116                         ws.peer.close()
 117                 } else {
 118                         ws.peer.remoteRejectedRequest(r)
 119                 }
 120         } else {
 121                 err := ws.peer.receiveChunk(&pp.Message{
 122                         Type:  pp.Piece,
 123                         Index: r.Index,
 124                         Begin: r.Begin,
 125                         Piece: result.Bytes,
 126                 })
 127                 if err != nil {
 128                         panic(err)
 129                 }
 130         }
 131 }