diff --git a/download/common/constants.go b/download/common/constants.go new file mode 100644 index 00000000..ef11227b --- /dev/null +++ b/download/common/constants.go @@ -0,0 +1,23 @@ +package common + +type Status int + +const ( + DownloadStatusReady Status = iota + DownloadStatusStart + DownloadStatusPause + DownloadStatusError + DownloadStatusDone +) + +const ( + HttpCodeOK = 200 + HttpCodePartialContent = 206 + + HttpHeaderRange = "Range" + HttpHeaderContentLength = "Content-Length" + HttpHeaderContentRange = "Content-Range" + HttpHeaderContentDisposition = "Content-Disposition" + + HttpHeaderRangeFormat = "bytes=%d-%d" +) diff --git a/download/common/controller.go b/download/common/controller.go new file mode 100644 index 00000000..fe3e3d75 --- /dev/null +++ b/download/common/controller.go @@ -0,0 +1,129 @@ +package common + +import ( + "golang.org/x/net/proxy" + "net" + "os" + "time" +) + +// 下载请求 +type Request struct { + // 下载链接 + URL string + // 附加信息 + Extra interface{} +} + +// 资源信息 +type Resource struct { + Req *Request + // 资源总大小 + Size int64 + // 是否支持断点下载 + Range bool + // 资源所包含的文件列表 + Files []*FileInfo +} + +type Options struct { + // 保存文件名 + Name string + // 保存目录 + Path string + // 并发连接数 + Connections int +} + +type FileInfo struct { + Name string + Path string + Size int64 + Downloaded int64 +} + +type Process interface { + Start() error + Pause() error + Continue() error + Delete() error +} + +type Controller struct { + Files map[string]*os.File +} + +func (c *Controller) Touch(name string, size int64) error { + if size > 0 { + return os.Truncate(name, size) + } + _, err := os.Create(name) + return err +} + +func (c *Controller) Write(name string, offset int64, buf []byte) (int, error) { + file, ok := c.Files[name] + if !ok { + var err error + file, err = os.OpenFile(name, os.O_RDWR, 0666) + if err != nil { + return 0, err + } + c.Files[name] = file + } + return file.WriteAt(buf, offset) +} + +func (c *Controller) ContextDialer() (proxy.Dialer, error) { + // return proxy.SOCKS5("tpc", "127.0.0.1:9999", nil, nil) + var dialer proxy.Dialer + return &DialerWarp{dialer: dialer}, nil +} + +type DialerWarp struct { + dialer proxy.Dialer +} + +type ConnWarp struct { + conn net.Conn +} + +func (c *ConnWarp) Read(b []byte) (n int, err error) { + return c.conn.Read(b) +} + +func (c *ConnWarp) Write(b []byte) (n int, err error) { + return c.conn.Write(b) +} + +func (c *ConnWarp) Close() error { + return c.conn.Close() +} + +func (c *ConnWarp) LocalAddr() net.Addr { + return c.conn.LocalAddr() +} + +func (c *ConnWarp) RemoteAddr() net.Addr { + return c.conn.RemoteAddr() +} + +func (c *ConnWarp) SetDeadline(t time.Time) error { + return c.conn.SetDeadline(t) +} + +func (c *ConnWarp) SetReadDeadline(t time.Time) error { + return c.conn.SetReadDeadline(t) +} + +func (c *ConnWarp) SetWriteDeadline(t time.Time) error { + return c.conn.SetWriteDeadline(t) +} + +func (d *DialerWarp) Dial(network, addr string) (c net.Conn, err error) { + conn, err := d.dialer.Dial(network, addr) + if err != nil { + return nil, err + } + return &ConnWarp{conn: conn}, nil +} diff --git a/download/common/fetcher.go b/download/common/fetcher.go new file mode 100644 index 00000000..038228b0 --- /dev/null +++ b/download/common/fetcher.go @@ -0,0 +1,29 @@ +package common + +// 对应协议的下载支持 +type Fetcher interface { + BaseFetch + // 支持的协议列表 + Protocols() []string + // 解析请求 + Resolve(req *Request) (*Resource, error) + // 创建任务 + Create(res *Resource, opts *Options) (Process, error) +} + +type BaseFetch interface { + InitCtl(ctl *Controller) + GetCtl() *Controller +} + +type BaseFetcher struct { + ctl *Controller +} + +func (b *BaseFetcher) InitCtl(ctl *Controller) { + b.ctl = ctl +} + +func (b *BaseFetcher) GetCtl() *Controller { + return b.ctl +} diff --git a/download/downloader.go b/download/downloader.go new file mode 100644 index 00000000..d6b7a688 --- /dev/null +++ b/download/downloader.go @@ -0,0 +1,82 @@ +package download + +import ( + "errors" + "github.com/google/uuid" + "github.com/monkeyWie/gopeed/download/common" + "net/url" + "os" + "strings" +) + +type TaskInfo struct { + ID string + Res *common.Resource + Opts *common.Options + Status common.Status + Files map[string]*common.FileInfo + + process common.Process +} + +type Downloader struct { + fetches map[string]common.Fetcher + tasks map[string]*TaskInfo + *common.Controller +} + +func NewDownloader(fetchers ...common.Fetcher) *Downloader { + d := &Downloader{Controller: &common.Controller{Files: make(map[string]*os.File)}} + d.fetches = make(map[string]common.Fetcher) + for _, f := range fetchers { + for _, p := range f.Protocols() { + d.fetches[strings.ToUpper(p)] = f + } + f.InitCtl(d.Controller) + } + d.tasks = make(map[string]*TaskInfo) + return d +} + +func (d *Downloader) getFetcher(URL string) (common.Fetcher, error) { + url, err := url.Parse(URL) + if err != nil { + return nil, err + } + if fetch, ok := d.fetches[strings.ToUpper(url.Scheme)]; ok { + return fetch, nil + } + return nil, errors.New("unsupported protocol") +} + +func (d *Downloader) Resolve(req *common.Request) (*common.Resource, error) { + fetcher, err := d.getFetcher(req.URL) + if err != nil { + return nil, err + } + return fetcher.Resolve(req) +} + +func (d *Downloader) Create(res *common.Resource, opts *common.Options) error { + fetcher, err := d.getFetcher(res.Req.URL) + if err != nil { + return err + } + process, err := fetcher.Create(res, opts) + if err != nil { + return err + } + id := uuid.New().String() + d.tasks[id] = &TaskInfo{ + ID: id, + Res: res, + Opts: opts, + Status: common.DownloadStatusReady, + process: process, + } + return process.Start() +} + +func (d *Downloader) Pause(id string) { + d.tasks[id].process.Pause() +} diff --git a/download/downloader_test.go b/download/downloader_test.go new file mode 100644 index 00000000..51d1ef7f --- /dev/null +++ b/download/downloader_test.go @@ -0,0 +1,40 @@ +package download + +import ( + "fmt" + "github.com/monkeyWie/gopeed/download/common" + "github.com/monkeyWie/gopeed/download/http" + "testing" +) + +func TestDownloader_Resolve(t *testing.T) { + downloader := buildDownloader() + res, err := downloader.Resolve(&common.Request{ + URL: "http://192.168.200.163:8088/docker-compose", + }) + if err != nil { + t.Fatal(err) + } + fmt.Println(res) +} + +func TestDownloader_Create(t *testing.T) { + downloader := buildDownloader() + res, err := downloader.Resolve(&common.Request{ + URL: "http://192.168.200.163:8088/docker-compose", + }) + if err != nil { + t.Fatal(err) + } + err = downloader.Create(res, &common.Options{ + Path: "E:\\test\\gopeed\\http", + Connections: 4, + }) + if err != nil { + t.Fatal(err) + } +} + +func buildDownloader() *Downloader { + return NewDownloader(&http.Fetcher{BaseFetcher: &common.BaseFetcher{}}) +} diff --git a/download/http/fetcher.go b/download/http/fetcher.go new file mode 100644 index 00000000..d84745df --- /dev/null +++ b/download/http/fetcher.go @@ -0,0 +1,148 @@ +package http + +import ( + "bytes" + "errors" + "fmt" + "github.com/monkeyWie/gopeed/download/common" + "github.com/monkeyWie/gopeed/download/http/model" + "io/ioutil" + "mime" + "net/http" + "net/http/cookiejar" + "net/url" + "strconv" + "strings" + "time" +) + +type Fetcher struct { + *common.BaseFetcher +} + +func (h *Fetcher) Protocols() []string { + return []string{"HTTP", "HTTPS"} +} + +func (h *Fetcher) Resolve(req *common.Request) (*common.Resource, error) { + httpReq, err := BuildRequest(req) + if err != nil { + return nil, err + } + client := BuildClient() + // 只访问一个字节,测试资源是否支持Range请求 + httpReq.Header.Set(common.HttpHeaderRange, fmt.Sprintf(common.HttpHeaderRangeFormat, 0, 0)) + httpResp, err := client.Do(httpReq) + if err != nil { + return nil, err + } + defer httpResp.Body.Close() + res := &common.Resource{ + Req: req, + Range: false, + Files: []*common.FileInfo{}, + } + if common.HttpCodePartialContent == httpResp.StatusCode { + // 返回206响应码表示支持断点下载 + res.Range = true + // 解析资源大小: bytes 0-1000/1001 => 1001 + contentTotal := subLastSlash(httpResp.Header.Get(common.HttpHeaderContentRange)) + if contentTotal != "" { + parse, err := strconv.ParseInt(contentTotal, 10, 64) + if err != nil { + return nil, err + } + res.Size = parse + } + } else if common.HttpCodeOK == httpResp.StatusCode { + // 返回200响应码,不支持断点下载,通过Content-Length头获取文件大小,获取不到的话可能是chunked编码 + contentLength := httpResp.Header.Get(common.HttpHeaderContentLength) + parse, err := strconv.ParseInt(contentLength, 10, 64) + if err != nil { + return nil, err + } + res.Size = parse + } else { + return nil, errors.New("http request fail") + } + file := &common.FileInfo{ + Size: res.Size, + } + contentDisposition := httpResp.Header.Get(common.HttpHeaderContentDisposition) + if contentDisposition != "" { + _, params, _ := mime.ParseMediaType(contentDisposition) + filename := params["filename"] + if filename != "" { + file.Name = filename + } + } + // Get file name by URL + if file.Name == "" { + parse, err := url.Parse(req.URL) + if err == nil { + // e.g. /files/test.txt => test.txt + file.Name = subLastSlash(parse.Path) + } + } + // unknown file name + if file.Name == "" { + file.Name = "unknown" + } + res.Files = append(res.Files, file) + return res, nil +} + +func (h *Fetcher) Create(res *common.Resource, opts *common.Options) (common.Process, error) { + if opts.Connections != 1 && !res.Range { + opts.Connections = 1 + } + return NewProcess(h, res, opts), nil +} + +func subLastSlash(str string) string { + if str == "" { + return "" + } + index := strings.LastIndex(str, "/") + if index != -1 { + return str[index+1:] + } + return "" +} + +func BuildClient() *http.Client { + // Cookie handle + jar, _ := cookiejar.New(nil) + return &http.Client{ + Jar: jar, + Timeout: time.Second * 10, + } +} + +func BuildRequest(req *common.Request) (*http.Request, error) { + url, err := url.Parse(req.URL) + if err != nil { + return nil, err + } + httpReq := &http.Request{ + URL: url, + Header: map[string][]string{}, + } + + if req.Extra != nil { + if extra, ok := req.Extra.(model.Extra); ok { + if extra.Method != "" { + httpReq.Method = extra.Method + } else { + httpReq.Method = http.MethodGet + } + if len(extra.Header) > 0 { + for k, v := range extra.Header { + httpReq.Header[k] = []string{v} + } + } + httpReq.Body = ioutil.NopCloser(bytes.NewBufferString(extra.Body)) + } + } + return httpReq, nil +} diff --git a/download/http/fetcher_test.go b/download/http/fetcher_test.go new file mode 100644 index 00000000..a05bc86f --- /dev/null +++ b/download/http/fetcher_test.go @@ -0,0 +1,95 @@ +package http + +import ( + "encoding/json" + "fmt" + "github.com/monkeyWie/gopeed/download/common" + "golang.org/x/net/proxy" + "io" + "io/ioutil" + "net/http" + "os" + "testing" + "time" +) + +func TestFetcher_Resolve(t *testing.T) { + fetcher := &Fetcher{} + resolve, err := fetcher.Resolve(&common.Request{ + // URL: "https://github.com/monkeyWie/github-actions-demo/releases/download/v1.0.6/simple-v1.0.6-windows-x64.zip", + URL: "http://192.168.200.163:8088/go1.12.7.linux-amd64.tar.gz", + }) + if err != nil { + t.Fatal(err) + } + buf, err := json.Marshal(resolve) + if err != nil { + t.Fatal(err) + } + fmt.Printf("%s\n", string(buf)) +} + +func Test2(t *testing.T) { + socks5, err := proxy.SOCKS5("tcp", "127.0.0.1:1080", nil, nil) + if err != nil { + t.Fatal(err) + } + conn, err := socks5.Dial("tcp", "www.baidu.com:80") + if err != nil { + t.Fatal(err) + } + conn.Write([]byte("GET")) +} + +func Test3(t *testing.T) { + var u User + fmt.Println((&u).Say1()) + fmt.Println(u.Say2()) +} + +func Test4(t *testing.T) { + resp, err := http.Get("http://192.168.200.163:8088/go1.12.7.linux-amd64.tar.gz") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + go func() { + time.Sleep(time.Second * 3) + err2 := resp.Body.Close() + if err2 != nil { + fmt.Println("close err2:" + err2.Error()) + } + }() + _, err = io.Copy(ioutil.Discard, resp.Body) + if err != nil { + t.Fatal(err) + } +} + +func Test5(t *testing.T) { + name := "E:\\test\\gopeed\\http\\test.txt" + err := os.Truncate(name, 10) + if err != nil { + t.Fatal(err) + } + file, err := os.OpenFile(name, os.O_RDWR, 0666) + if err != nil { + t.Fatal(err) + } + _, err = file.WriteAt([]byte("test"), 5) + if err != nil { + t.Fatal(err) + } +} + +type User struct { + name string +} + +func (u *User) Say1() string { + return "hello1 " + u.name +} + +func (u User) Say2() string { + return "hello2 " + u.name +} diff --git a/download/http/model/chunk.go b/download/http/model/chunk.go new file mode 100644 index 00000000..6ae44152 --- /dev/null +++ b/download/http/model/chunk.go @@ -0,0 +1,20 @@ +package model + +import ( + common2 "github.com/monkeyWie/gopeed/download/common" +) + +type Chunk struct { + Status common2.Status + Begin int64 + End int64 + Downloaded int64 +} + +func NewChunk(begin int64, end int64) *Chunk { + return &Chunk{ + Status: common2.DownloadStatusReady, + Begin: begin, + End: end, + } +} diff --git a/download/http/model/extra.go b/download/http/model/extra.go new file mode 100644 index 00000000..48b5c4f1 --- /dev/null +++ b/download/http/model/extra.go @@ -0,0 +1,7 @@ +package model + +type Extra struct { + Method string + Header map[string]string + Body string +} diff --git a/download/http/process.go b/download/http/process.go new file mode 100644 index 00000000..6adcb273 --- /dev/null +++ b/download/http/process.go @@ -0,0 +1,156 @@ +package http + +import ( + "fmt" + "github.com/monkeyWie/gopeed/download/common" + "github.com/monkeyWie/gopeed/download/http/model" + "io" + "net/http" + "path/filepath" +) + +type Process struct { + fetcher *Fetcher + + res *common.Resource + opts *common.Options + status common.Status + clients []*http.Response + chunks []*model.Chunk +} + +func NewProcess(fetcher *Fetcher, res *common.Resource, opts *common.Options) *Process { + return &Process{ + fetcher: fetcher, + res: res, + opts: opts, + status: common.DownloadStatusReady, + } +} + +func (p *Process) Start() error { + ctl := p.fetcher.GetCtl() + // 创建文件 + var filename = p.opts.Name + if filename == "" { + filename = p.res.Files[0].Name + } + name := filepath.Join(p.opts.Path, filename) + err := ctl.Touch(name, p.res.Size) + if err != nil { + return err + } + if p.res.Range && p.res.Size > 0 { + // 每个连接平均需要下载的分块大小 + var ( + chunkSize = p.res.Size / int64(p.opts.Connections) + errCh = make(chan error) + ) + defer close(errCh) + p.chunks = make([]*model.Chunk, p.opts.Connections) + p.clients = make([]*http.Response, p.opts.Connections) + for i := 0; i < p.opts.Connections; i++ { + begin := chunkSize * int64(i) + end := begin + chunkSize + if i == p.opts.Connections-1 { + // 最后一个连接下载的分块大小需要特殊计算,保证把文件下载完 + chunkSize = end + p.res.Size%int64(p.opts.Connections) + } + p.chunks[i] = model.NewChunk(begin, end) + go func(i int) { + errCh <- p.doFetch(i, name, begin, end) + }(i) + } + var failErr error + for i := 0; i < p.opts.Connections; i++ { + err := <-errCh + if failErr == nil { + // 有一个连接下载失败,立即停止下载 + failErr = err + if failErr != nil { + p.Pause() + } + } + } + if failErr != nil { + return failErr + } + } + return nil +} + +func (p *Process) Pause() error { + if len(p.clients) > 0 { + for _, client := range p.clients { + client.Body.Close() + } + } + return nil +} + +func (p *Process) Continue() error { + return nil +} + +func (p *Process) Delete() error { + panic("implement me") +} + +func (p *Process) doFetch(index int, filename string, begin, end int64) (err error) { + httpReq, err := BuildRequest(p.res.Req) + if err != nil { + return err + } + httpReq.Header.Set(common.HttpHeaderRange, fmt.Sprintf(common.HttpHeaderRangeFormat, begin, end)) + client := BuildClient() + + // 重试5次 + for i := 0; i < 5; i++ { + var ( + resp *http.Response + retry bool + ) + resp, err = client.Do(httpReq) + if err != nil { + // 连接失败,直接重试 + continue + } + p.chunks[index].Downloaded = 0 + p.clients[index] = resp + retry, err = func() (bool, error) { + defer resp.Body.Close() + var ( + buf = make([]byte, 8192) + offset = begin + ) + for { + n, err := resp.Body.Read(buf) + if n > 0 { + len, err := p.fetcher.GetCtl().Write(filename, offset, buf[:n]) + if err != nil { + return false, err + } + offset += int64(len) + p.chunks[index].Downloaded = p.chunks[i].Downloaded + int64(len) + } + if err != nil { + if err != io.EOF { + return true, err + } + break + } + } + return false, nil + }() + if err == nil || !retry { + // 下载成功,跳出重试 + break + } + } + if err != nil { + p.chunks[index].Status = common.DownloadStatusError + } else { + p.chunks[index].Status = common.DownloadStatusDone + } + return +} diff --git a/go.mod b/go.mod index 7351aa56..09585352 100644 --- a/go.mod +++ b/go.mod @@ -5,6 +5,8 @@ go 1.13 require ( github.com/RoaringBitmap/roaring v0.4.21 github.com/cenkalti/mse v0.0.0-20140930130441-6ef65f170972 + github.com/google/uuid v1.1.1 github.com/marksamman/bencode v0.0.0-20150821143521-dc84f26e086e github.com/sirupsen/logrus v1.4.2 + golang.org/x/net v0.0.0-20200707034311-ab3426394381 ) diff --git a/go.sum b/go.sum index ac334005..5dac46a6 100644 --- a/go.sum +++ b/go.sum @@ -12,6 +12,8 @@ github.com/glycerine/goconvey v0.0.0-20190410193231-58a59202ab31 h1:gclg6gY70GLy github.com/glycerine/goconvey v0.0.0-20190410193231-58a59202ab31/go.mod h1:Ogl1Tioa0aV7gstGFO7KhffUsb9M4ydbEbbxpcEDc24= github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4= github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= +github.com/google/uuid v1.1.1 h1:Gkbcsh/GbpXz7lPftLA3P6TYMwjCLYm83jiFQZF/3gY= +github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/gopherjs/gopherjs v0.0.0-20190910122728-9d188e94fb99 h1:twflg0XRTjwKpxb/jFExr4HGq6on2dEOmnL6FV+fgPw= github.com/gopherjs/gopherjs v0.0.0-20190910122728-9d188e94fb99/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY= github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo= @@ -37,8 +39,17 @@ github.com/tinylib/msgp v1.1.0 h1:9fQd+ICuRIu/ue4vxJZu6/LzxN0HwMds2nq/0cFvxHU= github.com/tinylib/msgp v1.1.0/go.mod h1:+d+yLhGm8mzTaHzB+wgMYrodPfmZrzkirds8fDWklFE= github.com/willf/bitset v1.1.10 h1:NotGKqX0KwQ72NUzqrjZq5ipPNDQex9lo3WpaS8L2sc= github.com/willf/bitset v1.1.10/go.mod h1:RjeCKbqT1RxIR/KWY6phxZiaY1IyutSBfGjNPySAYV4= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= +golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.0.0-20200707034311-ab3426394381 h1:VXak5I6aEWmAXeQjA+QSZzlgNrpq9mjcfDemuexIKsU= +golang.org/x/net v0.0.0-20200707034311-ab3426394381/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20190422165155-953cdadca894 h1:Cz4ceDQGXuKRnVBDTS23GTn/pU5OE2C0WrNTOYK1Uuc= golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw= diff --git a/protocol/bt/dht/ping.go b/protocol/bt/dht/ping.go new file mode 100644 index 00000000..3a054be6 --- /dev/null +++ b/protocol/bt/dht/ping.go @@ -0,0 +1,34 @@ +package dht + +import ( + "github.com/marksamman/bencode" + "net" +) + +func ping() string { + data := map[string]interface{}{ + "t": "aa", + "y": "q", + "q": "ping", + "a": map[string]interface{}{"id": "abcdefghij0123456789"}, + } + req := bencode.Encode(data) + // session.add_dht_router("router.utorrent.com", 6881) + // session.add_dht_router("router.bittorrent.com", 6881) + // session.add_dht_router("dht.transmissionbt.com", 6881) + // session.add_dht_router("router.bitcomet.com", 6881) + // session.add_dht_router("dht.aelitis.com", 6881) + dial, err := net.Dial("udp", "router.utorrent.com:6881") + if err != nil { + panic(err) + } + dial.Write(req) + + reps := make([]byte, 65535) + + n, err := dial.Read(reps) + if err != nil { + panic(err) + } + return string(reps[:n]) +} diff --git a/protocol/bt/dht/ping_test.go b/protocol/bt/dht/ping_test.go new file mode 100644 index 00000000..ea0e1e79 --- /dev/null +++ b/protocol/bt/dht/ping_test.go @@ -0,0 +1,7 @@ +package dht + +import "testing" + +func Test_ping(t *testing.T) { + println(ping()) +} diff --git a/protocol/bt/metainfo/metainfo_test.go b/protocol/bt/metainfo/metainfo_test.go index aa1ab2ba..e940abfd 100644 --- a/protocol/bt/metainfo/metainfo_test.go +++ b/protocol/bt/metainfo/metainfo_test.go @@ -3,31 +3,20 @@ package metainfo import ( "encoding/hex" "fmt" - "github.com/RoaringBitmap/roaring" "testing" ) func TestParseFromFile(t *testing.T) { // metaInfo, err := ParseFromFile("../testdata/Game.of.Thrones.S08E05.720p.WEB.H264-MEMENTO.torrent") - metaInfo, err := ParseFromFile("../testdata/office.torrent") + metaInfo, err := ParseFromFile("E:\\bt\\win10.torrent") if err != nil { panic(err) } fmt.Printf("%+v\n", metaInfo.AnnounceList) - fmt.Printf("%+v\n", metaInfo.Info.Files) + for _, file := range metaInfo.Info.Files { + fmt.Printf("%+v\n", file) + } fmt.Printf("%+v\n", metaInfo.Info.PieceLength) fmt.Printf("%+v\n", len(metaInfo.Info.Pieces)) fmt.Printf("%+v\n", hex.EncodeToString(metaInfo.infoHash[:])) } - -func TestParseFromFile2(t *testing.T) { - a := 0b10010000 - b := 0b01111111 - fmt.Printf("%08b\n", (a&b)^b) - - have := roaring.BitmapOf(0, 3) - peer := roaring.BitmapOf(1, 2, 3, 4, 5, 6, 7) - - need := roaring.Xor(roaring.And(have, peer), peer) - fmt.Println(need.ToArray()) -} diff --git a/protocol/bt/peer/message/msg_bitfield.go b/protocol/bt/peer/message/msg_bitfield.go index 54ab50c1..73758181 100644 --- a/protocol/bt/peer/message/msg_bitfield.go +++ b/protocol/bt/peer/message/msg_bitfield.go @@ -60,10 +60,9 @@ func (b *Bitfield) IsComplete(i int) bool { } // 给定一组分片下载状态,计算出当前peer能提供下载的分片下标,如果已经拥有则不用提供 -// had = 0b10010000,has = 0b01111111 -// had & has = 0b10010000 & 0b01111111 = 0b00010000 -// had & has ^ has = 0b00010000 ^ 0b01111111 = 0b01101111 +// has = 0b01111111,had = 0b10010000 +// has & ~had = 0b01111111 & 0b01101111 = 0b01101111 func (b *Bitfield) Provide(had *roaring.Bitmap) []uint32 { - xor := roaring.Xor(roaring.And(had, b.bitmap), b.bitmap) + xor := roaring.AndNot(b.bitmap, had) return xor.ToArray() } diff --git a/protocol/bt/torrent/peer_conn.go b/protocol/bt/torrent/peer_conn.go index 478d2b71..c437380a 100644 --- a/protocol/bt/torrent/peer_conn.go +++ b/protocol/bt/torrent/peer_conn.go @@ -324,7 +324,7 @@ func (pc *peerConn) handlePiece(buf []byte) { fileWritable := f.End - blockFileBegin // 计算block剩余写入字节数 blockWritable := blockLength - blockSeek - fb := fileBlock{ + fb := &fileBlock{ filepath: filepath.Join(f.Path...), fileSeek: blockFileBegin - f.Begin, blockBegin: blockSeek, @@ -420,6 +420,7 @@ func (pc *peerConn) handlePiece(buf []byte) { } // 获取piece对应要写入的文件 +// TODO 使用sort.Search二分查找法检索 func getWriteFile(pieceBegin int64, fds []*metainfo.FileDetail) int { for i, f := range fds { if f.Begin <= pieceBegin && f.End > pieceBegin { diff --git a/protocol/bt/torrent/peer_conn_test.go b/protocol/bt/torrent/peer_conn_test.go index 795313df..f951e628 100644 --- a/protocol/bt/torrent/peer_conn_test.go +++ b/protocol/bt/torrent/peer_conn_test.go @@ -64,7 +64,7 @@ func getUsablePeerMore() chan *peerConn { peers := <-tracker.Tracker() ch := make(chan *peerConn) for i := range peers { - go peerTest(torrent, &peers[i], ch) + go peerTest(torrent, peers[i], ch) } return ch } diff --git a/protocol/bt/torrent/peer_pool.go b/protocol/bt/torrent/peer_pool.go index 2b2d2b97..c5f65794 100644 --- a/protocol/bt/torrent/peer_pool.go +++ b/protocol/bt/torrent/peer_pool.go @@ -11,7 +11,7 @@ import ( type peerPool struct { torrent *Torrent - states map[peer.Peer]*peerState + states map[*peer.Peer]*peerState lock *sync.Mutex } @@ -26,7 +26,7 @@ func newPeerPool(torrent *Torrent) *peerPool { return &peerPool{ torrent: torrent, lock: &sync.Mutex{}, - states: map[peer.Peer]*peerState{}, + states: map[*peer.Peer]*peerState{}, } } @@ -52,7 +52,7 @@ func (pp *peerPool) fetch() { }() } -func (pp *peerPool) put(peers []peer.Peer) { +func (pp *peerPool) put(peers []*peer.Peer) { pp.lock.Lock() defer pp.lock.Unlock() for _, peer := range peers { @@ -72,7 +72,7 @@ func (pp *peerPool) get() *peer.Peer { for peer, state := range pp.states { if !state.using { state.using = true - return &peer + return peer } } return nil @@ -82,18 +82,18 @@ func (pp *peerPool) get() *peer.Peer { func (pp *peerPool) release(peer *peer.Peer) { pp.lock.Lock() defer pp.lock.Unlock() - pp.states[*peer].using = false - pp.states[*peer].errors = 0 + pp.states[peer].using = false + pp.states[peer].errors = 0 } // 标记peer为不可用,超过3次则剔除该peer func (pp *peerPool) unavailable(peer *peer.Peer) { pp.lock.Lock() defer pp.lock.Unlock() - pp.states[*peer].using = false - pp.states[*peer].errors++ - if pp.states[*peer].errors > 3 { - delete(pp.states, *peer) + pp.states[peer].using = false + pp.states[peer].errors++ + if pp.states[peer].errors > 3 { + delete(pp.states, peer) } } diff --git a/protocol/bt/tracker/tracker.go b/protocol/bt/tracker/tracker.go index fef4b144..df567a8f 100644 --- a/protocol/bt/tracker/tracker.go +++ b/protocol/bt/tracker/tracker.go @@ -7,7 +7,7 @@ import ( "io" "io/ioutil" "math" - mrand "math/rand" + "math/rand" "net" "net/http" "net/url" @@ -47,7 +47,7 @@ func newUdpConnectRequest() *udpConnectRequest { return &udpConnectRequest{ protocolId: udpConnectRequestMagic, action: udpActionConnect, - transactionId: mrand.Uint32(), + transactionId: rand.Uint32(), } } @@ -118,7 +118,7 @@ func newUdpAnnounceRequest(connectionId uint64) *udpAnnounceRequest { return &udpAnnounceRequest{ connectionId: connectionId, action: udpActionAnnounce, - transactionId: mrand.Uint32(), + transactionId: rand.Uint32(), } } @@ -187,7 +187,7 @@ type Tracker struct { MetaInfo *metainfo.MetaInfo } -func (tracker *Tracker) connect(conn *net.UDPConn, timeout int64) (response *udpConnectResponse, err error) { +func (tracker *Tracker) connect(conn net.Conn, timeout int64) (response *udpConnectResponse, err error) { request := newUdpConnectRequest() buf := request.encode() conn.SetReadDeadline(time.Now().Add(time.Second * time.Duration(timeout))) @@ -204,7 +204,7 @@ func (tracker *Tracker) connect(conn *net.UDPConn, timeout int64) (response *udp return } -func (tracker *Tracker) announce(conn *net.UDPConn, timeout int64, connectionId uint64) (response *udpAnnounceResponse, err error) { +func (tracker *Tracker) announce(conn net.Conn, timeout int64, connectionId uint64) (response *udpAnnounceResponse, err error) { request := newUdpAnnounceRequest(connectionId) request.infoHash = tracker.MetaInfo.GetInfoHash() request.peerID = tracker.PeerID @@ -346,7 +346,7 @@ func (tracker *Tracker) udpTracker(url *url.URL) (peers []*peer.Peer, err error) var connectResponse *udpConnectResponse // 0:connect 1:announce state := udpActionConnect - // 访问超时最多重试8次,当有请求成功则将重试次数重置为0 + // 访问超时最多重试8次,当有请求成功则将重试次数重置为0 for n := 0; n < udpConnectRetries; n++ { // 15 * 2 ^ n seconds timeout := int64(udpConnectTimeout * math.Pow(2, float64(n))) @@ -385,26 +385,8 @@ func (tracker *Tracker) udpTracker(url *url.URL) (peers []*peer.Peer, err error) return announceResponse.peers, nil } -func dial(trackerUrl *url.URL) (*net.UDPConn, error) { - host, portStr, err := net.SplitHostPort(trackerUrl.Host) - if err != nil { - return nil, err - } - port, err := strconv.ParseInt(portStr, 10, 32) - if err != nil { - return nil, err - } - ip, err := lookupIP(host, trackerUrl.Scheme) - if err != nil { - return nil, err - } - - srcAddr := &net.UDPAddr{IP: net.IPv4zero, Port: 0} - dstAddr := &net.UDPAddr{ - IP: ip, - Port: int(port), - } - return net.DialUDP(trackerUrl.Scheme, srcAddr, dstAddr) +func dial(trackerUrl *url.URL) (net.Conn, error) { + return net.Dial(trackerUrl.Scheme, trackerUrl.Host) } func lookupIP(host string, scheme string) (ip net.IP, err error) { diff --git a/protocol/http/down.go b/protocol/http/client.go similarity index 96% rename from protocol/http/down.go rename to protocol/http/client.go index 8dba50aa..f1c1aa4e 100644 --- a/protocol/http/down.go +++ b/protocol/http/client.go @@ -13,11 +13,6 @@ import ( "sync" ) -type Download interface { - Resolve(request *http.Request) (*http.Response, error) - Down(request *http.Request) error -} - // Resolve return the file response to be downloaded func Resolve(request *Request) (*Response, error) { httpRequest, err := BuildHTTPRequest(request) @@ -53,9 +48,9 @@ func Resolve(request *Request) (*Response, error) { ret.Name = subLastSlash(parse.Path) } } - // Unknow file name + // Unknown file name if ret.Name == "" { - ret.Name = "unknow" + ret.Name = "unknown" } // Is support range ret.Range = response.StatusCode == 206 diff --git a/protocol/http/down_test.go b/protocol/http/client_test.go similarity index 100% rename from protocol/http/down_test.go rename to protocol/http/client_test.go