webseed/client.go

   1 package webseed
   2
   3 import (
   4         "bytes"
   5         "context"
   6         "fmt"
   7         "io"
   8         "net/http"
   9
  10         "github.com/anacrolix/torrent/metainfo"
  11         "github.com/anacrolix/torrent/segments"
  12 )
  13
  14 type RequestSpec = segments.Extent
  15
  16 type httpRequestResult struct {
  17         resp *http.Response
  18         err  error
  19 }
  20
  21 type requestPart struct {
  22         req    *http.Request
  23         e      segments.Extent
  24         result chan httpRequestResult
  25 }
  26
  27 type request struct {
  28         cancel func()
  29 }
  30
  31 type Client struct {
  32         HttpClient *http.Client
  33         Url        string
  34         FileIndex  segments.Index
  35         Info       *metainfo.Info
  36
  37         requests map[RequestSpec]request
  38         Events   chan ClientEvent
  39 }
  40
  41 type ClientEvent struct {
  42         RequestSpec RequestSpec
  43         Bytes       []byte
  44         Err         error
  45 }
  46
  47 func (ws *Client) Cancel(r RequestSpec) {
  48         ws.requests[r].cancel()
  49 }
  50
  51 func (ws *Client) Request(r RequestSpec) {
  52         ctx, cancel := context.WithCancel(context.Background())
  53         var requestParts []requestPart
  54         if !ws.FileIndex.Locate(r, func(i int, e segments.Extent) bool {
  55                 req, err := NewRequest(ws.Url, i, ws.Info, e.Start, e.Length)
  56                 if err != nil {
  57                         panic(err)
  58                 }
  59                 req = req.WithContext(ctx)
  60                 part := requestPart{
  61                         req:    req,
  62                         result: make(chan httpRequestResult, 1),
  63                         e:      e,
  64                 }
  65                 go func() {
  66                         resp, err := ws.HttpClient.Do(req)
  67                         part.result <- httpRequestResult{
  68                                 resp: resp,
  69                                 err:  err,
  70                         }
  71                 }()
  72                 requestParts = append(requestParts, part)
  73                 return true
  74         }) {
  75                 panic("request out of file bounds")
  76         }
  77         if ws.requests == nil {
  78                 ws.requests = make(map[RequestSpec]request)
  79         }
  80         ws.requests[r] = request{cancel}
  81         go func() {
  82                 b, err := readRequestPartResponses(requestParts)
  83                 ws.Events <- ClientEvent{
  84                         RequestSpec: r,
  85                         Bytes:       b,
  86                         Err:         err,
  87                 }
  88         }()
  89 }
  90
  91 func recvPartResult(buf io.Writer, part requestPart) error {
  92         result := <-part.result
  93         if result.err != nil {
  94                 return result.err
  95         }
  96         defer result.resp.Body.Close()
  97         if part.e.Start != 0 && result.resp.StatusCode != http.StatusPartialContent {
  98                 return fmt.Errorf("expected partial content response got %v", result.resp.StatusCode)
  99         }
 100         copied, err := io.Copy(buf, result.resp.Body)
 101         if err != nil {
 102                 return err
 103         }
 104         if copied != part.e.Length {
 105                 return fmt.Errorf("got %v bytes, expected %v", copied, part.e.Length)
 106         }
 107         return nil
 108 }
 109
 110 func readRequestPartResponses(parts []requestPart) ([]byte, error) {
 111         var buf bytes.Buffer
 112         for _, part := range parts {
 113                 err := recvPartResult(&buf, part)
 114                 if err != nil {
 115                         return buf.Bytes(), err
 116                 }
 117         }
 118         return buf.Bytes(), nil
 119 }