11 "github.com/anacrolix/log"
12 "github.com/anacrolix/torrent/common"
13 "github.com/anacrolix/torrent/metainfo"
14 pp "github.com/anacrolix/torrent/peer_protocol"
15 "github.com/anacrolix/torrent/segments"
16 "github.com/anacrolix/torrent/webseed"
19 type webseedPeer struct {
21 activeRequests map[Request]webseed.Request
22 requesterCond sync.Cond
26 var _ peerImpl = (*webseedPeer)(nil)
28 func (me *webseedPeer) connStatusString() string {
32 func (ws *webseedPeer) String() string {
33 return fmt.Sprintf("webseed peer for %q", ws.client.Url)
36 func (ws *webseedPeer) onGotInfo(info *metainfo.Info) {
37 ws.client.FileIndex = segments.NewIndex(common.LengthIterFromUpvertedFiles(info.UpvertedFiles()))
41 func (ws *webseedPeer) writeInterested(interested bool) bool {
45 func (ws *webseedPeer) _cancel(r RequestIndex) bool {
46 active, ok := ws.activeRequests[ws.peer.t.requestIndexToRequest(r)]
49 if !ws.peer.deleteRequest(r) {
50 panic("cancelled webseed request should exist")
52 if ws.peer.actualRequestState.Requests.IsEmpty() {
53 ws.peer.updateRequests("webseedPeer._cancel")
59 func (ws *webseedPeer) intoSpec(r Request) webseed.RequestSpec {
60 return webseed.RequestSpec{ws.peer.t.requestOffset(r), int64(r.Length)}
63 func (ws *webseedPeer) _request(r Request) bool {
64 ws.requesterCond.Signal()
68 func (ws *webseedPeer) doRequest(r Request) {
69 webseedRequest := ws.client.NewRequest(ws.intoSpec(r))
70 ws.activeRequests[r] = webseedRequest
72 ws.requesterCond.L.Unlock()
73 defer ws.requesterCond.L.Lock()
74 ws.requestResultHandler(r, webseedRequest)
76 delete(ws.activeRequests, r)
79 func (ws *webseedPeer) requester() {
80 ws.requesterCond.L.Lock()
81 defer ws.requesterCond.L.Unlock()
83 for !ws.peer.closed.IsSet() {
85 ws.peer.actualRequestState.Requests.Iterate(func(x uint32) bool {
86 r := ws.peer.t.requestIndexToRequest(x)
87 if _, ok := ws.activeRequests[r]; ok {
97 ws.requesterCond.Wait()
101 func (ws *webseedPeer) connectionFlags() string {
105 // TODO: This is called when banning peers. Perhaps we want to be able to ban webseeds too. We could
106 // return bool if this is even possible, and if it isn't, skip to the next drop candidate.
107 func (ws *webseedPeer) drop() {}
109 func (ws *webseedPeer) handleUpdateRequests() {
110 ws.peer.maybeUpdateActualRequestState()
113 func (ws *webseedPeer) onClose() {
114 ws.peer.logger.WithLevel(log.Debug).Print("closing")
115 for _, r := range ws.activeRequests {
118 ws.requesterCond.Broadcast()
121 func (ws *webseedPeer) requestResultHandler(r Request, webseedRequest webseed.Request) {
122 result := <-webseedRequest.Result
123 // We do this here rather than inside receiveChunk, since we want to count errors too. I'm not
124 // sure if we can divine which errors indicate cancellation on our end without hitting the
126 ws.peer.doChunkReadStats(int64(len(result.Bytes)))
128 defer ws.peer.t.cl.unlock()
129 if result.Err != nil {
130 if !errors.Is(result.Err, context.Canceled) {
131 ws.peer.logger.Printf("Request %v rejected: %v", r, result.Err)
133 // We need to filter out temporary errors, but this is a nightmare in Go. Currently a bad
134 // webseed URL can starve out the good ones due to the chunk selection algorithm.
135 const closeOnAllErrors = false
136 if closeOnAllErrors ||
137 strings.Contains(result.Err.Error(), "unsupported protocol scheme") ||
139 var err webseed.ErrBadResponse
140 if !errors.As(result.Err, &err) {
143 return err.Response.StatusCode == http.StatusNotFound
147 ws.peer.remoteRejectedRequest(ws.peer.t.requestIndexFromRequest(r))
150 err := ws.peer.receiveChunk(&pp.Message{