包名重构

This commit is contained in:
liwei committed 2020-02-25 14:22:17 +08:00
1 parent fb45066228
commit 8298051c81
37 files changed
+220 -134

No files matched your search

+57
View File
@@ -0,0 +1,57 @@
package bt
import (
"crypto/rand"
"github.com/monkeyWie/gopeed/protocol/bt/metainfo"
"github.com/monkeyWie/gopeed/protocol/bt/torrent"
"net/url"
)
type Client struct {
PeerID [20]byte
// torrents []torrent.Torrent
}
func NewClient() *Client {
client := &Client{}
// 生成Peer ID,规则为前三位固定字母(-GP)+SemVer(xyz)+End(-),后面随机生成
// 参考:https://wiki.theory.org/index.php/BitTorrentSpecification#peer_id
peerID := [20]byte{'-', 'G', 'P', '0', '0', '1', '-'}
_, err := rand.Read(peerID[7:])
if err != nil {
panic(err)
}
client.PeerID = peerID
return client
}
func (client *Client) AddTorrent(rawurl string) error {
parse, err := url.Parse(rawurl)
if err != nil {
return err
}
var metaInfo *metainfo.MetaInfo
switch parse.Scheme {
case "magnet":
// TODO 磁力链接解析
default:
metaInfo, err = metainfo.ParseFromFile(rawurl)
break
}
if err != nil {
return err
}
torrent := torrent.Torrent{
client: client,
MetaInfo: metaInfo,
}
client.torrents = append(client.torrents, torrent)
torrent.Download()
return nil
}
+21
View File
@@ -0,0 +1,21 @@
package bt
import (
"testing"
)
func TestClient_AddTorrent(t *testing.T) {
/*client := NewClient()
metaInfo, err := metainfo.ParseFromFile("testdata/Game.of.Thrones.S08E05.720p.WEB.H264-MEMENTO.torrent")
if err != nil {
panic(err)
}
torrent := &
torrent.MetaInfo.Announce = "udp://tracker.opentrackr.org:1337/announce"
torrent.MetaInfo.AnnounceList = [][]string{}
err = torrent.Tracker()
if err != nil {
panic(err)
return
}*/
}
+143
View File
@@ -0,0 +1,143 @@
package metainfo
import (
"crypto/sha1"
"encoding/json"
"github.com/marksamman/bencode"
"os"
)
type MetaInfo struct {
// tracker URL
// http://bittorrent.org/beps/bep_0003.html#metainfo-files
Announce string `json:"announce"`
// tracker URL
// http://bittorrent.org/beps/bep_0012.html
AnnounceList [][]string `json:"announce-list"`
Comment string `json:"comment"`
CreatedBy string `json:"created by"`
CreationDate int64 `json:"creation date"`
Encoding string `json:"encoding"`
// http://bittorrent.org/beps/bep_0019.html
UrlList []string `json:"url-list"`
Info *Info `json:"info"`
infoHash [20]byte `json:"-"`
totalSize uint64 `json:"-"`
fileDetails []FileDetail `json:"-"`
}
type Info struct {
Name string `json:"name"`
PieceLength int `json:"piece length"`
Pieces [][20]byte
// 单个文件时,length为该文件的长度
Length uint64 `json:"length"`
// 多个文件
Files []File `json:"files"`
}
type File struct {
Length uint64 `json:"length"`
Path []string `json:"path"`
}
type FileDetail struct {
Length uint64
Path []string
Begin int64
End int64
}
type PieceMapping struct {
file int
begin int64
end int64
}
func ParseFromFile(path string) (*MetaInfo, error) {
file, err := os.Open(path)
if err != nil {
return nil, err
}
// http://bittorrent.org/beps/bep_0003.html#bencoding
btDecode, err := bencode.Decode(file)
if err != nil {
return nil, err
}
bytes, err := json.Marshal(btDecode)
if err != nil {
return nil, err
}
var metaInfo MetaInfo
json.Unmarshal(bytes, &metaInfo)
if metaInfo.Info != nil {
// encode info hash
info := btDecode["info"].(map[string]interface{})
infoBtEncode := bencode.Encode(info)
metaInfo.infoHash = sha1.Sum(infoBtEncode[:])
// Split pieces hash
pieces := info["pieces"].(string)
if pieces != "" {
metaInfo.Info.Pieces = make([][20]byte, len(pieces)/20)
// pieces parse
for i := range metaInfo.Info.Pieces {
begin := i * 20
metaInfo.Info.Pieces[i] = [20]byte{}
copy(metaInfo.Info.Pieces[i][:], pieces[begin:begin+20])
}
}
}
calcFileSize(&metaInfo)
return &metaInfo, nil
}
// 获取某个分片的大小
func (metaInfo *MetaInfo) GetPieceLength(index int) int {
// 是否为最后一个分片
if index == len(metaInfo.Info.Pieces)-1 {
size := metaInfo.totalSize % uint64(metaInfo.Info.PieceLength)
if size > 0 {
return int(size)
}
}
return metaInfo.Info.PieceLength
}
// 获取所有文件的大小
func (metaInfo *MetaInfo) GetTotalSize() uint64 {
return metaInfo.totalSize
}
// 获取info hash
func (metaInfo *MetaInfo) GetInfoHash() [20]byte {
return metaInfo.infoHash
}
// 获取所有文件的开始偏移字节
func (metaInfo *MetaInfo) GetFileDetails() []FileDetail {
return metaInfo.fileDetails
}
// 计算所有文件的大小
func calcFileSize(metaInfo *MetaInfo) {
if metaInfo.Info != nil {
fileCount := len(metaInfo.Info.Files)
if fileCount > 0 {
metaInfo.fileDetails = make([]FileDetail, fileCount)
var length uint64
for i, f := range metaInfo.Info.Files {
metaInfo.fileDetails[i] = FileDetail{
Length: f.Length,
Path: f.Path,
Begin: int64(length),
End: int64(length + f.Length),
}
length += f.Length
}
metaInfo.totalSize = length
} else {
metaInfo.totalSize = metaInfo.Info.Length
}
}
}
+29
View File
@@ -0,0 +1,29 @@
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")
if err != nil {
panic(err)
}
fmt.Printf("%+v\n", metaInfo.AnnounceList)
fmt.Printf("%+v\n", metaInfo.Info.Files)
fmt.Printf("%+v\n", metaInfo.Info.PieceLength)
fmt.Printf("%+v\n", hex.EncodeToString(metaInfo.infoHash[:]))
}
func TestParseFromFile2(t *testing.T) {
bitmap := roaring.NewBitmap()
bitmap.AddMany([]uint32{1, 2, 3, 4, 5, 6, 7, 10, 20, 9, 9, 1})
fmt.Println(bitmap.Contains(1))
fmt.Println(bitmap.Contains(2))
fmt.Println(bitmap.Contains(8))
fmt.Println(bitmap.GetCardinality())
}
+51
View File
@@ -0,0 +1,51 @@
package peer
import (
"bytes"
"encoding/binary"
)
const (
ProtocolIdentifier = "BitTorrent protocol"
ProtocolIdentifierLength = 0x13
)
// Handshake of Peer wire protocol
// Per https://wiki.theory.org/index.php/BitTorrentSpecification#Handshake
type Handshake struct {
Pstrlen byte
Pstr [19]byte
Reserved [8]byte
InfoHash [20]byte
PeerID [20]byte
}
func NewHandshake(reserved [8]byte, infoHash [20]byte, peerID [20]byte) *Handshake {
var arr [ProtocolIdentifierLength]byte
copy(arr[:], ProtocolIdentifier)
return &Handshake{
Pstrlen: ProtocolIdentifierLength,
Pstr: arr,
Reserved: reserved,
InfoHash: infoHash,
PeerID: peerID,
}
}
func (handshake *Handshake) Encode() []byte {
writer := new(bytes.Buffer)
err := binary.Write(writer, binary.BigEndian, handshake)
if err != nil {
panic(err)
}
return writer.Bytes()
}
func (handshake *Handshake) Decode(buf []byte) error {
reader := bytes.NewReader(buf)
err := binary.Read(reader, binary.BigEndian, handshake)
if err != nil {
return err
}
return nil
}
+96
View File
@@ -0,0 +1,96 @@
package message
import (
"encoding/binary"
)
// message ID
type ID byte
const (
IdChoke ID = iota
IdUnchoke
IdInterested
IdNotinterested
IdHave
IdBitfield
IdRequest
IdPiece
IdCancel
)
/*type Message struct {
Length uint32
ID MessageID
Payload []byte
}*/
type Message struct {
Length uint32
ID ID
}
func (msg *Message) Encode() []byte {
buf := make([]byte, 5)
binary.BigEndian.PutUint32(buf, msg.Length)
buf[4] = byte(msg.ID)
return buf
}
func (msg *Message) Decode(buf []byte) {
msg.Length = binary.BigEndian.Uint32(buf)
msg.ID = ID(buf[4])
}
type Serialize interface {
Encode() []byte
Decode(buf []byte) Serialize
}
/*func NewMessage(id MessageID, payload []byte) *Message {
message := &Message{ID: id, Payload: payload}
if id == Keepalive {
message.Length = 0
} else {
message.Length = uint32(len(payload) + 1)
}
return message
}
func (msg *Message) Encode() []byte {
if msg.ID == Keepalive {
return make([]byte, 4)
} else {
buf := make([]byte, 5+len(msg.Payload))
binary.BigEndian.PutUint32(buf, msg.Length)
buf[4] = byte(msg.ID)
copy(buf[5:], msg.Payload[:])
return buf
}
}
func (msg *Message) Decode(buf []byte) {
msg.Length = binary.BigEndian.Uint32(buf)
if msg.Length == 0 {
msg.ID = -1
} else {
msg.ID = MessageID(buf[4])
if msg.Length > 4 {
msg.Payload = buf[5:]
}
}
}*/
// Message protocol
// Length prefix| message ID | Payload
// 4-byte | 1-byte | remaining bytes
// 100 | 5 | 99-bytes
// https://wiki.theory.org/index.php/BitTorrentSpecification#Messages
func SplitMessage(data []byte, atEOF bool) (advance int, token []byte, err error) {
if !atEOF && len(data) > 4 {
length := int(binary.BigEndian.Uint32(data))
if len(data)-4 >= length {
return length + 4, data[:length+4], nil
}
}
return
}
+60
View File
@@ -0,0 +1,60 @@
package message
// bitfield: <len=0001+X><id=5><bitfield>
type Bitfield struct {
*Message
payload []byte
}
func NewBitfield(payload []byte) *Bitfield {
return &Bitfield{
Message: &Message{
Length: uint32(len(payload) + 1),
ID: IdBitfield,
},
payload: payload,
}
}
func (b *Bitfield) Encode() []byte {
return append(b.Message.Encode(), b.payload...)
}
func (b *Bitfield) Decode(buf []byte) Serialize {
b.Message = &Message{}
b.Message.Decode(buf)
b.payload = buf[5:]
return b
}
// 某个分片是否下载完成
func (b *Bitfield) IsComplete(i int) bool {
index := i / 8
if index >= len(b.payload) {
return false
}
return b.payload[index]&(1<<(7-i%8)) > 0
}
// 给定一组分片下载状态,计算出当前peer能提供下载的分片下标
func (b *Bitfield) Have(pieces []bool) []int {
arr := make([]int, 0)
length := len(pieces) / 8
if len(pieces)%8 != 0 {
length++
}
for i := 0; i < length; i++ {
for j := 0; j < 8; j++ {
index := i*8 + j
// 如果此分片在本地还未下载,检查peer是否能提供该分片下载
if index < len(pieces) &&
!pieces[index] &&
i < len(b.payload) &&
b.payload[i]&(1<<(7-j)) > 0 {
arr = append(arr, index)
}
}
}
return arr
}
@@ -0,0 +1,162 @@
package message
import (
"reflect"
"testing"
)
func TestBitfield_IsComplete(t *testing.T) {
type fields struct {
Message Message
payload []byte
}
type args struct {
i int
}
tests := []struct {
name string
fields fields
args args
want bool
}{
{
"case-0",
fields{
payload: []byte{0b10000000},
},
args{
0,
},
true,
},
{
"case-1",
fields{
payload: []byte{0b01000000},
},
args{
1,
},
true,
},
{
"case-2",
fields{
payload: []byte{0b00000000, 0b00100000},
},
args{
2,
},
false,
},
{
"case-10",
fields{
payload: []byte{0b00000000, 0b00100000},
},
args{
10,
},
true,
},
{
"case-11",
fields{
payload: []byte{0b00000000, 0b00100000},
},
args{
11,
},
false,
},
{
"case-20",
fields{
payload: []byte{0b00000000, 0b00100000},
},
args{
20,
},
false,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
b := &Bitfield{
Message: &tt.fields.Message,
payload: tt.fields.payload,
}
if got := b.IsComplete(tt.args.i); got != tt.want {
t.Errorf("Bitfield.IsComplete() = %v, want %v", got, tt.want)
}
})
}
}
func TestBitfield_Have(t *testing.T) {
type fields struct {
Message Message
payload []byte
}
type args struct {
pieces []bool
}
tests := []struct {
name string
fields fields
args args
want []int
}{
{
"case-1",
fields{
payload: []byte{0b11111111, 0b00000000},
},
args{
[]bool{false, false, false, false, false, false, false, false},
},
[]int{0, 1, 2, 3, 4, 5, 6, 7},
},
{
"case-2",
fields{
payload: []byte{0b01010101, 0b01000000},
},
args{
[]bool{false, false, false, false, false, false, false, false, false, false, false, false, false, false, false, false},
},
[]int{1, 3, 5, 7, 9},
},
{
"case-3",
fields{
payload: []byte{0b11010101, 0b01000000},
},
args{
[]bool{false, true, false},
},
[]int{0},
},
{
"case-4",
fields{
payload: []byte{0b11010101},
},
args{
[]bool{false, true, false, false, true, false, false, true, false},
},
[]int{0, 3, 5},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
b := &Bitfield{
Message: &tt.fields.Message,
payload: tt.fields.payload,
}
if got := b.Have(tt.args.pieces); !reflect.DeepEqual(got, tt.want) {
t.Errorf("Bitfield.Have() = %v, want %v", got, tt.want)
}
})
}
}
+24
View File
@@ -0,0 +1,24 @@
package message
// choke: <len=0001><id=0>
type Choke struct {
Message
}
func NewChoke() *Choke {
return &Choke{
Message{
Length: 1,
ID: IdChoke,
},
}
}
func (c *Choke) Encode() []byte {
return c.Message.Encode()
}
func (c *Choke) Decode(buf []byte) Serialize {
c.Message.Decode(buf)
return c
}
@@ -0,0 +1,24 @@
package message
// choke: <len=0001><id=0>
type Interested struct {
Message
}
func NewInterested() *Interested {
return &Interested{
Message{
Length: 1,
ID: IdInterested,
},
}
}
func (c *Interested) Encode() []byte {
return c.Message.Encode()
}
func (c *Interested) Decode(buf []byte) Serialize {
c.Message.Decode(buf)
return c
}
+12
View File
@@ -0,0 +1,12 @@
package message
// keep-alive: <len=0000>
type Keepalive struct{}
func (k *Keepalive) Encode() []byte {
return []byte{0, 0, 0, 0}
}
func (k *Keepalive) Decode(buf []byte) Serialize {
return &Keepalive{}
}
@@ -0,0 +1,24 @@
package message
// choke: <len=0001><id=0>
type NotInterested struct {
Message
}
func NewNotInterested() *NotInterested {
return &NotInterested{
Message{
Length: 1,
ID: IdNotinterested,
},
}
}
func (ni *NotInterested) Encode() []byte {
return ni.Message.Encode()
}
func (ni *NotInterested) Decode(buf []byte) Serialize {
ni.Message.Decode(buf)
return ni
}
+41
View File
@@ -0,0 +1,41 @@
package message
import "encoding/binary"
// piece: <len=0009+X><id=7><index><begin><block>
type Piece struct {
*Message
Index uint32
Begin uint32
Block []byte
}
func NewPiece(index uint32, begin uint32, block []byte) *Piece {
return &Piece{
Message: &Message{
Length: 9 + uint32(len(block)),
ID: IdPiece,
},
Index: index,
Begin: begin,
// Block: io.LimitReader(reader, int64(length)),
Block: block,
}
}
func (p *Piece) Encode() []byte {
buf := make([]byte, 8)
binary.BigEndian.PutUint32(buf[:4], p.Index)
binary.BigEndian.PutUint32(buf[4:8], p.Begin)
buf = append(buf, p.Block...)
return append(p.Message.Encode(), buf...)
}
func (p *Piece) Decode(buf []byte) Serialize {
p.Message = &Message{}
p.Message.Decode(buf)
p.Index = binary.BigEndian.Uint32(buf[5:9])
p.Begin = binary.BigEndian.Uint32(buf[9:13])
p.Block = buf[13:]
return p
}
+41
View File
@@ -0,0 +1,41 @@
package message
import (
"encoding/binary"
)
// request: <len=0013><id=6><index><begin><length>
type Request struct {
Message
Index uint32
Begin uint32
Length uint32
}
func NewRequest(index uint32, begin uint32, length uint32) *Request {
return &Request{
Message: Message{
Length: 13,
ID: IdRequest,
},
Index: index,
Begin: begin,
Length: length,
}
}
func (r *Request) Encode() []byte {
buf := make([]byte, 12)
binary.BigEndian.PutUint32(buf[0:4], r.Index)
binary.BigEndian.PutUint32(buf[4:8], r.Begin)
binary.BigEndian.PutUint32(buf[8:12], r.Length)
return append(r.Message.Encode(), buf...)
}
func (r *Request) Decode(buf []byte) Serialize {
r.Message.Decode(buf)
r.Length = binary.BigEndian.Uint32(buf[5:9])
r.Begin = binary.BigEndian.Uint32(buf[9:13])
r.Index = binary.BigEndian.Uint32(buf[13:17])
return r
}
+24
View File
@@ -0,0 +1,24 @@
package message
// unchoke: <len=0001><id=1>
type Unchoke struct {
Message
}
func NewUnchoke() *Unchoke {
return &Unchoke{
Message{
Length: 1,
ID: IdUnchoke,
},
}
}
func (u *Unchoke) Encode() []byte {
return u.Message.Encode()
}
func (u *Unchoke) Decode(buf []byte) Serialize {
u.Message.Decode(buf)
return u
}
+30
View File
@@ -0,0 +1,30 @@
package peer
import (
"encoding/binary"
"math/rand"
"net"
"strconv"
)
type Peer struct {
IP uint32
Port uint16
}
func (peer *Peer) Address() string {
bts := make([]byte, 4)
binary.BigEndian.PutUint32(bts, peer.IP)
return net.IP(bts).String() + ":" + strconv.Itoa(int(peer.Port))
}
// 生成Peer ID,规则为前三位固定字母(-GP)+SemVer(xyz)+End(-),后面随机生成
// 参考:https://wiki.theory.org/index.php/BitTorrentSpecification#peer_id
func GenPeerID() [20]byte {
peerID := [20]byte{'-', 'G', 'P', '0', '0', '1', '-'}
_, err := rand.Read(peerID[7:])
if err != nil {
panic(err)
}
return peerID
}
Binary file not shown.
+408
View File
@@ -0,0 +1,408 @@
package torrent
import (
"bufio"
"crypto/sha1"
"encoding/binary"
"errors"
"fmt"
"github.com/cenkalti/mse"
"github.com/monkeyWie/gopeed/protocol/bt/metainfo"
"github.com/monkeyWie/gopeed/protocol/bt/peer"
"github.com/monkeyWie/gopeed/protocol/bt/peer/message"
"path/filepath"
"sync/atomic"
"io"
"math"
"net"
"os"
"path"
"time"
)
// 现在主流客户端的block大小都是16KB
const BlockSize = 2 << 13
type peerConn struct {
torrent *Torrent
peer *peer.Peer
conn net.Conn
// this client is choking the peer
amChoking bool
// this client is interested in the peer
amInterested bool
// peer is choking this client
peerChoking bool
// peer is interested in this client
peerInterested bool
bitfield *message.Bitfield
readyComplete bool
downloadedCh chan bool
// block个数
blockCount int
// 已经下载的block个数
blockDownloadedCount int32
// block下载队列,官方推荐为5
blockQueue chan interface{}
}
func NewPeerConn(torrent *Torrent, peer *peer.Peer) *peerConn {
return &peerConn{
torrent: torrent,
peer: peer,
}
}
// 使用MSE加密来避免运营商对bt流量的封锁,基本上现在市面上BT客户端都默认开启了,不用MSE的话很多Peer拒绝连接
// see http://wiki.vuze.com/w/Message_Stream_Encryption
func (pc *peerConn) dialMse() error {
conn, err := net.DialTimeout("tcp", pc.peer.Address(), time.Second*time.Duration(30))
if err != nil {
return err
}
mseConn := mse.WrapConn(conn)
infoHash := pc.torrent.MetaInfo.GetInfoHash()
_, err = mseConn.HandshakeOutgoing(infoHash[:], mse.PlainText, nil)
if err != nil {
mseConn.Close()
return err
}
pc.conn = mseConn
return nil
}
// Handshake of Peer wire protocol
// see https://wiki.theory.org/index.php/BitTorrentSpecification#Handshake
func (pc *peerConn) handshake() (*peer.Handshake, error) {
handshakeRes, err := func() (*peer.Handshake, error) {
handshakeReq := peer.NewHandshake([8]byte{}, pc.torrent.MetaInfo.GetInfoHash(), pc.torrent.PeerID)
_, err := pc.conn.Write(handshakeReq.Encode())
if err != nil {
return nil, err
}
var read [68]byte
_, err = io.ReadFull(pc.conn, read[:])
if err != nil {
return nil, err
}
handshakeRes := &peer.Handshake{}
err = handshakeRes.Decode(read[:])
if err != nil {
return nil, err
}
// InfoHash不匹配
if handshakeRes.InfoHash != handshakeReq.InfoHash {
return nil, fmt.Errorf("info_hash not currently serving")
}
return handshakeRes, nil
}()
if err != nil {
pc.conn.Close()
return nil, err
}
// init state
pc.amChoking = true
pc.amInterested = false
pc.peerChoking = true
pc.peerInterested = false
return handshakeRes, nil
}
// 准备下载
func (pc *peerConn) ready() error {
if err := pc.dialMse(); err != nil {
return err
}
if _, err := pc.handshake(); err != nil {
return err
}
pc.readyComplete = false
readyCh := make(chan bool)
defer close(readyCh)
go func() {
scanner := bufio.NewScanner(pc.conn)
scanner.Split(message.SplitMessage)
for scanner.Scan() {
buf := scanner.Bytes()
length := binary.BigEndian.Uint32(buf[:4])
if length == 0 {
// keepalive
} else {
messageID := message.ID(buf[4])
switch messageID {
case message.IdChoke:
break
case message.IdUnchoke:
pc.handleUnchoke(readyCh)
break
case message.IdInterested:
break
case message.IdNotinterested:
break
case message.IdHave:
break
case message.IdBitfield:
pc.handleBitfield(buf)
break
case message.IdRequest:
break
case message.IdPiece:
pc.handlePiece(buf)
break
case message.IdCancel:
break
}
}
}
}()
err := func() error {
select {
case status := <-readyCh:
if status {
return nil
} else {
return errors.New("ready fail")
}
case <-time.After(time.Second * 30):
// 30秒之后超时
return errors.New("ready time out")
}
}()
pc.readyComplete = true
if err != nil {
pc.conn.Close()
return err
}
pc.blockQueue = make(chan interface{}, 5)
return nil
}
// 下载指定piece
func (pc *peerConn) downloadPiece(index int) error {
pieceLength := pc.torrent.MetaInfo.GetPieceLength(index)
lastOffset := 0
pc.blockCount = int(math.Ceil(float64(pieceLength) / BlockSize))
pc.downloadedCh = make(chan bool)
defer close(pc.downloadedCh)
waitDuration := time.Second * 30
timer := time.NewTimer(waitDuration)
defer timer.Stop()
// 按块下载分片
for i := 0; i < pc.blockCount; i++ {
offset := pieceLength * i
var blockLength uint32
if i == pc.blockCount-1 {
// 最后一个block大小需要计算出来
blockLength = uint32(pieceLength - offset)
} else {
blockLength = BlockSize
}
// 标记block开始下载
offset32 := uint32(lastOffset)
// block下载排队,并做超时处理
timer.Reset(waitDuration)
select {
case pc.blockQueue <- nil:
case <-timer.C:
return errors.New("download timeout")
}
// 发起request,对方会响应piece
_, err := pc.conn.Write(message.NewRequest(uint32(index), offset32, blockLength).Encode())
if err != nil {
fmt.Println(err)
}
}
// 监听是否下载完成
<-pc.downloadedCh
return nil
}
func (pc *peerConn) handleUnchoke(readyCh chan<- bool) {
pc.peerChoking = false
// 已经处理过Unchoke信号
if pc.readyComplete {
return
}
// 如果客户端对peer感兴趣并且peer没有choked客户端,就可以开始下载了
if pc.amInterested {
readyCh <- true
} else {
readyCh <- false
}
}
func (pc *peerConn) handleBitfield(buf []byte) {
pc.bitfield = &message.Bitfield{}
pc.bitfield.Decode(buf)
have := pc.getHavePieces(pc.bitfield)
if len(have) > 0 {
// 表示对该peer感兴趣,并且不choked该peer
pc.conn.Write(message.NewInterested().Encode())
pc.amInterested = true
pc.conn.Write(message.NewUnchoke().Encode())
pc.amChoking = false
} else {
pc.conn.Close()
}
}
// 处理下载响应,每次接收到响应直接将block写入到对应文件中
func (pc *peerConn) handlePiece(buf []byte) {
piece := &message.Piece{}
piece.Decode(buf)
info := pc.torrent.MetaInfo.Info
fds := pc.torrent.MetaInfo.GetFileDetails()
blockLength := int64(len(piece.Block))
pieceBegin := int64(piece.Index) * int64(info.PieceLength)
blockBegin := pieceBegin + int64(piece.Begin)
// 计算block对应要写到的文件偏移
var fileBlocks []fileBlock
if len(info.Files) == 0 {
// 单文件
fileBlocks = append(fileBlocks, fileBlock{
filepath: info.Name,
fileSeek: blockBegin,
blockBegin: 0,
blockEnd: blockLength,
})
} else {
// 获取要写入的第一个文件和偏移
writeIndex := getWriteFile(blockBegin, fds)
// block对应种子的偏移
blockFileBegin := blockBegin
// block对应文件的偏移
var blockSeek int64 = 0
for _, f := range fds[writeIndex:] {
// 计算文件可剩余写入字节数
fileWritable := f.End - blockFileBegin
// 计算block剩余写入字节数
blockWritable := blockLength - blockSeek
fb := fileBlock{
filepath: filepath.Join(f.Path...),
fileSeek: blockFileBegin - f.Begin,
blockBegin: blockSeek,
}
if fileWritable >= blockWritable {
// 若够block写入直接跳出循环
fb.blockEnd = blockSeek + blockWritable
fileBlocks = append(fileBlocks, fb)
break
} else {
// 否则计算剩余可写字节数,写入到下一个文件中
fb.blockEnd = blockSeek + fileWritable
fileBlocks = append(fileBlocks, fb)
blockFileBegin = f.End
blockSeek += fileWritable
}
}
}
// 开始写入文件
for _, f := range fileBlocks {
func() {
name := filepath.Join(pc.torrent.Path, f.filepath)
file, err := os.OpenFile(name, os.O_RDWR|os.O_CREATE, 0644)
if err != nil {
panic(err)
}
defer file.Close()
_, err = file.WriteAt(piece.Block[f.blockBegin:f.blockEnd], f.fileSeek)
if err != nil {
panic(err)
}
}()
}
// 出队
<-pc.blockQueue
// piece全部下载完
if atomic.AddInt32(&pc.blockDownloadedCount, 1) == int32(pc.blockCount) {
// 计算piece对应的文件偏移
sha1 := sha1.New()
pieceLength := pc.torrent.MetaInfo.GetPieceLength(int(piece.Index))
writeIndex := getWriteFile(pieceBegin, fds)
fileBegin := pieceBegin - fds[writeIndex].Begin
for _, fd := range fds[writeIndex:] {
func() {
file, err := os.Open(path.Join(pc.torrent.Path, fd.Path[0]))
if err != nil {
panic(err)
}
defer file.Close()
_, err = file.Seek(fileBegin, 0)
if err != nil {
panic(err)
}
written, err := io.CopyN(sha1, file, int64(pieceLength))
if err != nil {
if err != io.EOF {
panic(err)
}
}
pieceLength -= int(written)
}()
if pieceLength > 0 {
// 继续读下个文件
fileBegin = 0
} else {
break
}
}
// 校验piece SHA-1 hash
downHash := [20]byte{}
copy(downHash[:], sha1.Sum(nil))
if downHash == pc.torrent.MetaInfo.Info.Pieces[piece.Index] {
fmt.Printf("piece %d 校验通过\n", piece.Index)
} else {
fmt.Printf("piece %d 校验失败\n", piece.Index)
}
// piece下载完成
pc.downloadedCh <- true
}
}
// 获取piece对应要写入的文件
func getWriteFile(pieceBegin int64, fds []metainfo.FileDetail) int {
for i, f := range fds {
if f.Begin <= pieceBegin && f.End > pieceBegin {
return i
}
}
return -1
}
type fileBlock struct {
filepath string
fileSeek int64
blockBegin int64
blockEnd int64
}
// 获取peer能提供需要下载的文件分片
func (pc *peerConn) getHavePieces(bitfield *message.Bitfield) []int {
states := make([]bool, pc.torrent.PiecesState.Size())
for i := range states {
states[i] = pc.torrent.PiecesState.getState(i) == stateFinished
}
return bitfield.Have(states)
}
// 获取要写入到的文件
/*func (ps *peerConn) getWriteFile(request *message.Request) string {
info := ps.torrent.MetaInfo.Info
// 单文件
if len(info.Files) == 0 {
return info.Name
} else {
request.Index * info.PieceLength + be
for i := 0; i < len(info.Files); i++ {
}
}
}
*/
+166
View File
@@ -0,0 +1,166 @@
package torrent
import (
"fmt"
"github.com/monkeyWie/gopeed/protocol/bt/peer"
"github.com/monkeyWie/gopeed/protocol/bt/tracker"
"os"
"sync"
"testing"
"time"
)
func Test_peerState_ready(t *testing.T) {
ps := getUsablePeer()
fmt.Println("get a peer")
// 下载前准备
err := ps.ready()
fmt.Println("ready end")
if err != nil {
fmt.Println(ps.peer.Address())
t.Fatal(err)
}
}
func Test_peerState_download(t *testing.T) {
psCh := getUsablePeerMore()
lock := sync.Mutex{}
state := map[int]bool{}
for {
select {
case ps := <-psCh:
go func() {
// 下载前准备
err := ps.ready()
if err != nil {
fmt.Println("ready error")
return
}
have := ps.getHavePieces(ps.bitfield)
fmt.Printf("download is ready,have %d\n", len(have))
if len(have) > 0 {
// 获取分片的长度
for _, index := range have {
lock.Lock()
if state[index] {
lock.Unlock()
continue
} else {
state[index] = true
lock.Unlock()
}
fmt.Printf("download index %d\n", index)
ps.downloadPiece(index)
}
return
}
}()
}
}
}
func Test_peerState_downloadPiece(t *testing.T) {
ps := getUsablePeer()
fmt.Println("get a peer")
// 下载前准备
err := ps.ready()
if err != nil {
t.Fatal(err)
}
have := ps.getHavePieces(ps.bitfield)
fmt.Printf("download is ready,have %d\n", len(have))
if len(have) > 0 {
// 获取分片的长度
for _, index := range have {
fmt.Printf("download index %d\n", index)
ps.downloadPiece(index)
}
return
}
}
// 获取一个能连接上的peer
func getUsablePeer() *peerConn {
ch := getUsablePeerMore()
ps := <-ch
return ps
}
// 获取多个能连接上的peer
func getUsablePeerMore() chan *peerConn {
torrent := buildTorrent()
tracker := &tracker.Tracker{
PeerID: torrent.PeerID,
MetaInfo: torrent.MetaInfo,
}
peers, err := tracker.Tracker()
if err != nil {
panic(err)
}
ch := make(chan *peerConn)
for i := range peers {
go peerTest(torrent, &peers[i], ch)
}
return ch
}
func peerTest(torrent *Torrent, peer *peer.Peer, ch chan *peerConn) {
ps := &peerConn{
torrent: torrent,
peer: peer,
}
// 连接至peer
err := ps.ready()
if err != nil {
return
}
ps.conn.Close()
ch <- ps
}
func TestSome(t *testing.T) {
ch := make(chan interface{})
go func() {
time.Sleep(time.Second)
select {
case ch <- nil:
fmt.Println("write")
case <-time.After(time.Second * 3):
fmt.Println("timeout")
}
}()
time.Sleep(time.Second * 5)
r := <-ch
fmt.Println(r)
fmt.Println("111")
}
func TestSome2(t *testing.T) {
wg := sync.WaitGroup{}
wg.Add(10)
file, err := os.OpenFile("e:/testbt/test.data", os.O_RDWR|os.O_CREATE, 0644)
for i := 0; i < 10; i++ {
go func(i int) {
for j := 0; j < 100; j++ {
if err != nil {
panic(err)
}
_, err = file.WriteAt([]byte{byte(i)}, int64(i*8))
}
wg.Done()
}(i)
}
wg.Wait()
}
+82
View File
@@ -0,0 +1,82 @@
package torrent
import (
"github.com/monkeyWie/gopeed/protocol/bt/peer"
"sync"
)
type peerPool struct {
peers []peerState
lock *sync.Mutex
}
type peerState struct {
peer *peer.Peer
using bool
ready bool
}
func newPeerPool() *peerPool {
return &peerPool{
lock: &sync.Mutex{},
}
}
func (pp *peerPool) put(peers []peer.Peer) {
pp.lock.Lock()
defer pp.lock.Unlock()
for i := range peers {
pp.peers = append(pp.peers, peerState{
peer: &peers[i],
using: false,
})
}
}
func (pp *peerPool) get() *peer.Peer {
pp.lock.Lock()
defer pp.lock.Unlock()
for i, p := range pp.peers {
if !p.using {
pp.peers[i].using = true
return pp.peers[i].peer
}
}
return nil
}
func (pp *peerPool) ready(peer *peer.Peer) {
pp.lock.Lock()
defer pp.lock.Unlock()
i := findPeer(pp, peer)
if i != -1 {
pp.peers[i].ready = true
}
}
func (pp *peerPool) release(peer *peer.Peer) {
pp.lock.Lock()
defer pp.lock.Unlock()
i := findPeer(pp, peer)
if i != -1 {
pp.peers[i].using = false
}
}
func (pp *peerPool) remove(peer *peer.Peer) {
pp.lock.Lock()
defer pp.lock.Unlock()
i := findPeer(pp, peer)
if i != -1 {
pp.peers = append(pp.peers[0:i], pp.peers[i+1:]...)
}
}
func findPeer(pp *peerPool, peer *peer.Peer) int {
for i := range pp.peers {
if pp.peers[i].peer == peer {
return i
}
}
return -1
}
+40
View File
@@ -0,0 +1,40 @@
package torrent
import (
"fmt"
"github.com/monkeyWie/gopeed/protocol/bt/peer"
"testing"
)
func Test_PeerPool(t *testing.T) {
pool := newPeerPool()
pool.put([]peer.Peer{
{
IP: 0,
Port: 0,
},
{
IP: 1,
Port: 0,
},
{
IP: 2,
Port: 0,
},
{
IP: 3,
Port: 0,
},
{
IP: 4,
Port: 0,
},
})
get := pool.get()
pool.release(get)
pool.remove(get)
for i := 0; i < 5; i++ {
fmt.Println(pool.get())
}
fmt.Println(pool.get())
}
+56
View File
@@ -0,0 +1,56 @@
package torrent
import "sync"
type state int
const (
stateReady = iota
stateDownloading
stateFinished
)
// 所有piece的下载状态
type piecesState struct {
rw *sync.RWMutex
states []state
}
func NewPiecesState(size int) *piecesState {
ps := &piecesState{
rw: &sync.RWMutex{},
states: make([]state, size),
}
for i := 0; i < size; i++ {
ps.states[i] = stateReady
}
return ps
}
func (ps *piecesState) getState(index int) state {
ps.rw.RLock()
defer ps.rw.RUnlock()
return ps.states[index]
}
func (ps *piecesState) setState(index int, state state) {
ps.rw.Lock()
defer ps.rw.Unlock()
ps.states[index] = state
}
func (ps *piecesState) getReadyAndDownload() int {
ps.rw.Lock()
defer ps.rw.Unlock()
for i, s := range ps.states {
if s == stateReady {
ps.states[i] = stateDownloading
return i
}
}
return -1
}
func (ps *piecesState) Size() int {
return len(ps.states)
}
+100
View File
@@ -0,0 +1,100 @@
package torrent
import (
"fmt"
"github.com/monkeyWie/gopeed/protocol/bt/metainfo"
"github.com/monkeyWie/gopeed/protocol/bt/tracker"
"sync"
"time"
)
type Torrent struct {
PeerID [20]byte
MetaInfo *metainfo.MetaInfo
PiecesState *piecesState
peerPool *peerPool
Path string
}
func NewTorrent(peerID [20]byte, metaInfo *metainfo.MetaInfo) *Torrent {
torrent := &Torrent{
PeerID: peerID,
MetaInfo: metaInfo,
PiecesState: NewPiecesState(len(metaInfo.Info.Pieces)),
}
return torrent
}
func (t *Torrent) Download(path string) {
t.Path = path
t.peerPool = newPeerPool()
t.fetchPeers()
taskCh := make(chan interface{}, 16)
defer close(taskCh)
for {
index := t.PiecesState.getReadyAndDownload()
if index == -1 {
return
}
taskCh <- nil
go func(index int) {
peer := t.peerPool.get()
if peer == nil {
time.Sleep(time.Second * 10)
t.PiecesState.setState(index, stateReady)
<-taskCh
return
}
pc := NewPeerConn(t, peer)
err := pc.ready()
if err != nil {
t.PiecesState.setState(index, stateReady)
t.peerPool.remove(peer)
} else {
fmt.Printf("peer ready:%s,download %d\n", peer.Address(), index)
err = pc.downloadPiece(index)
if err != nil {
fmt.Println(err)
// 下载失败
t.PiecesState.setState(index, stateReady)
} else {
// 下载完成
t.PiecesState.setState(index, stateFinished)
}
t.peerPool.release(peer)
}
<-taskCh
}(index)
}
}
// 获取可用的peer
func (t *Torrent) fetchPeers() {
tracker := &tracker.Tracker{
PeerID: t.PeerID,
MetaInfo: t.MetaInfo,
}
urls := tracker.MetaInfo.AnnounceList[0]
var wg sync.WaitGroup
ch := make(chan interface{})
wg.Add(len(urls))
for _, url := range urls {
go func(url string) {
defer wg.Done()
peers, err := tracker.DoTracker(url)
if err == nil {
fmt.Printf("Tracker end,url:%s,count:%d\n", url, len(peers))
ch <- nil
t.peerPool.put(peers)
}
}(url)
}
go func() {
wg.Wait()
close(ch)
}()
<-ch
}
+44
View File
@@ -0,0 +1,44 @@
package torrent
import (
"github.com/monkeyWie/gopeed/protocol/bt/metainfo"
"github.com/monkeyWie/gopeed/protocol/bt/peer"
"testing"
)
func TestTorrent_Download(t *testing.T) {
torrent := buildTorrent()
torrent.Download("e:/testbt/download")
}
func buildTorrent() *Torrent {
// metaInfo, err := metainfo.ParseFromFile("../testdata/Game.of.Thrones.S08E05.720p.WEB.H264-MEMENTO.torrent")
metaInfo, err := metainfo.ParseFromFile("../testdata/office.torrent")
if err != nil {
panic(err)
}
metaInfo.AnnounceList[0] = append(metaInfo.AnnounceList[0], []string{
"udp://tracker.coppersurfer.tk:6969/announce",
"udp://tracker.leechers-paradise.org:6969/announce",
"udp://tracker.opentrackr.org:1337/announce",
"udp://p4p.arenabg.com:1337/announce",
"udp://9.rarbg.to:2710/announce",
"udp://9.rarbg.me:2710/announce",
"udp://tracker.internetwarriors.net:1337/announce",
"udp://exodus.desync.com:6969/announce",
"udp://tracker.tiny-vps.com:6969/announce",
"udp://tracker.sbsub.com:2710/announce",
"udp://retracker.lanta-net.ru:2710/announce",
"udp://open.stealth.si:80/announce",
"udp://open.demonii.si:1337/announce",
"udp://tracker.torrent.eu.org:451/announce",
"udp://tracker.moeking.me:6969/announce",
"udp://tracker.cyberia.is:6969/announce",
"udp://denis.stalker.upeer.me:6969/announce",
"udp://ipv4.tracker.harry.lu:80/announce",
"udp://zephir.monocul.us:6969/announce",
"udp://xxxtor.com:2710/announce",
}...)
return NewTorrent(peer.GenPeerID(), metaInfo)
}
+29
View File
@@ -0,0 +1,29 @@
package tracker
import (
"fmt"
)
type ErrorCode int
const (
ErrTimeout ErrorCode = -1
ErrResponse ErrorCode = -2
)
type BtError interface {
error
}
type TrackerError struct {
Code ErrorCode
Err error
}
func (te *TrackerError) Error() string {
return fmt.Sprintf("%s:%d", te.Err.Error(), te.Code)
}
func NewTrackerError(code ErrorCode, err error) *TrackerError {
return &TrackerError{code, err}
}
+408
View File
@@ -0,0 +1,408 @@
package tracker
import (
"encoding/binary"
"errors"
"fmt"
"github.com/monkeyWie/gopeed/protocol/bt/metainfo"
"github.com/monkeyWie/gopeed/protocol/bt/peer"
"io"
"io/ioutil"
"math"
mrand "math/rand"
"net"
"net/http"
"net/url"
"strconv"
"time"
)
const (
udpConnectRequestMagic = 0x41727101980
udpConnectTimeout = 15
udpConnectRetries = 8
udpActionConnect = 0
udpActionAnnounce = 1
udpActionScrape = 2
udpActionError = 3
)
/*
Offset Size Name Value
0 64-bit integer protocol_id 0x41727101980 // magic constant
8 32-bit integer action 0 // connect
12 32-bit integer transaction_id
16
Per https://www.libtorrent.org/udp_tracker_protocol.html#connecting
*/
type udpConnectRequest struct {
protocolId uint64
action uint32
transactionId uint32
}
func newUdpConnectRequest() *udpConnectRequest {
return &udpConnectRequest{
protocolId: udpConnectRequestMagic,
action: udpActionConnect,
transactionId: mrand.Uint32(),
}
}
func (req *udpConnectRequest) encode() []byte {
buf := make([]byte, 16)
binary.BigEndian.PutUint64(buf[:8], req.protocolId)
binary.BigEndian.PutUint32(buf[8:12], req.action)
binary.BigEndian.PutUint32(buf[12:], req.transactionId)
return buf
}
/*
Offset Size Name Value
0 32-bit integer action 0 // connect
4 32-bit integer transaction_id
8 64-bit integer connection_id
16
*/
type udpConnectResponse struct {
action uint32
transactionId uint32
connectionId uint64
}
func newUdpConnectResponse(buf []byte) *udpConnectResponse {
return &udpConnectResponse{
action: binary.BigEndian.Uint32(buf[:4]),
transactionId: binary.BigEndian.Uint32(buf[4:8]),
connectionId: binary.BigEndian.Uint64(buf[8:]),
}
}
/*
Offset Size Name Value
0 64-bit integer connection_id
8 32-bit integer action 1 // announce
12 32-bit integer transaction_id
16 20-byte string info_hash
36 20-byte string peer_id
56 64-bit integer downloaded
64 64-bit integer left
72 64-bit integer uploaded
80 32-bit integer event 0 // 0: none; 1: completed; 2: started; 3: stopped
84 32-bit integer ip address 0 // default
88 32-bit integer key
92 32-bit integer num_want -1 // default
96 16-bit integer port
98
Per https://www.libtorrent.org/udp_tracker_protocol.html#announcing
*/
type udpAnnounceRequest struct {
connectionId uint64
action uint32
transactionId uint32
infoHash [20]byte
peerID [20]byte
downloaded uint64
left uint64
uploaded uint64
event uint32
ip uint32
key uint32
numWant int32
port uint16
}
func newUdpAnnounceRequest(connectionId uint64) *udpAnnounceRequest {
return &udpAnnounceRequest{
connectionId: connectionId,
action: udpActionAnnounce,
transactionId: mrand.Uint32(),
}
}
func (req *udpAnnounceRequest) encode() []byte {
buf := make([]byte, 98, 98)
binary.BigEndian.PutUint64(buf[:8], req.connectionId)
binary.BigEndian.PutUint32(buf[8:12], req.action)
binary.BigEndian.PutUint32(buf[12:16], req.transactionId)
copy(buf[16:36], req.infoHash[:])
copy(buf[36:56], req.peerID[:])
binary.BigEndian.PutUint64(buf[56:64], req.downloaded)
binary.BigEndian.PutUint64(buf[64:72], req.left)
binary.BigEndian.PutUint64(buf[72:80], req.uploaded)
binary.BigEndian.PutUint32(buf[80:84], req.event)
binary.BigEndian.PutUint32(buf[84:88], req.ip)
binary.BigEndian.PutUint32(buf[88:92], req.key)
binary.BigEndian.PutUint32(buf[92:96], uint32(req.numWant))
binary.BigEndian.PutUint16(buf[96:98], req.port)
return buf
}
/*
Offset Size Name Value
0 32-bit integer action 1 // announce
4 32-bit integer transaction_id
8 32-bit integer interval
12 32-bit integer leechers
16 32-bit integer seeders
20 + 6 * n 32-bit integer ip address
24 + 6 * n 16-bit integer TCP port
20 + 6 * N
Per https://www.libtorrent.org/udp_tracker_protocol.html#announcing
*/
type udpAnnounceResponse struct {
action uint32
transactionId uint32
interval uint32
leechers uint32
seeders uint32
peers []peer.Peer
}
func newUdpAnnounceResponse(buf []byte) *udpAnnounceResponse {
response := &udpAnnounceResponse{
action: binary.BigEndian.Uint32(buf[:4]),
transactionId: binary.BigEndian.Uint32(buf[4:8]),
interval: binary.BigEndian.Uint32(buf[8:12]),
leechers: binary.BigEndian.Uint32(buf[12:16]),
seeders: binary.BigEndian.Uint32(buf[16:20]),
}
count := (len(buf) - 20) / 6
response.peers = make([]peer.Peer, count, count)
for i := 0; i < count; i++ {
ipBegin := 20 + 6*i
portBegin := ipBegin + 4
response.peers[i] = peer.Peer{
IP: binary.BigEndian.Uint32(buf[ipBegin:portBegin]),
Port: binary.BigEndian.Uint16(buf[portBegin : portBegin+2]),
}
}
return response
}
type Tracker struct {
PeerID [20]byte
MetaInfo *metainfo.MetaInfo
}
func (tracker *Tracker) connect(conn *net.UDPConn, timeout int64) (response *udpConnectResponse, err error) {
request := newUdpConnectRequest()
buf := request.encode()
conn.SetReadDeadline(time.Now().Add(time.Second * time.Duration(timeout)))
conn.Write(buf)
len, err := conn.Read(buf)
if err != nil {
return
}
if len != 16 {
err = NewTrackerError(ErrResponse, errors.New("invalid response"))
return
}
response = newUdpConnectResponse(buf)
return
}
func (tracker *Tracker) announce(conn *net.UDPConn, timeout int64, connectionId uint64) (response *udpAnnounceResponse, err error) {
request := newUdpAnnounceRequest(connectionId)
request.infoHash = tracker.MetaInfo.GetInfoHash()
request.peerID = tracker.PeerID
request.downloaded = 0
request.uploaded = 0
request.left = tracker.MetaInfo.GetTotalSize() - request.downloaded
request.event = 0
request.ip = 0
request.key = 0
request.numWant = 50
request.port = 6882
encode := request.encode()
conn.SetReadDeadline(time.Now().Add(time.Second * time.Duration(timeout)))
conn.Write(encode)
buf := make([]byte, 512)
len, err := conn.Read(buf)
if err != nil {
return
}
response = newUdpAnnounceResponse(buf[:len])
return
}
// 通过tracker服务器获取peer信息
func (tracker *Tracker) Tracker() (peers []peer.Peer, err error) {
metaInfo := tracker.MetaInfo
checkAnnounceList := func(announceList [][]string) bool {
if len(announceList) > 0 {
for _, l := range announceList {
if len(l) > 0 {
return true
}
}
}
return false
}
// 只要获取到了peers就直接返回
if checkAnnounceList(metaInfo.AnnounceList) {
for _, announceArr := range metaInfo.AnnounceList {
if len(announceArr) > 0 {
for _, announce := range announceArr {
peers, err = tracker.DoTracker(announce)
if err == nil {
return
}
}
}
}
} else {
peers, err = tracker.DoTracker(metaInfo.Announce)
}
return
}
func (tracker *Tracker) DoTracker(announce string) (peers []peer.Peer, err error) {
if announce != "" {
url, _ := url.Parse(announce)
switch url.Scheme {
case "http", "https":
return tracker.httpTracker(url)
case "udp", "udp4", "udp6":
return tracker.udpTracker(url)
default:
return nil, errors.New("unsupported protocol")
}
}
return nil, errors.New("empty announce")
}
// http://bittorrent.org/beps/bep_0003.html#trackers
func (tracker *Tracker) httpTracker(url *url.URL) (peers []peer.Peer, err error) {
metaInfo := tracker.MetaInfo
peerID := tracker.PeerID
query := url.Query()
infoHash := metaInfo.GetInfoHash()
query.Add("info_hash", string(infoHash[:]))
query.Add("peer_id", string(peerID[:]))
query.Add("port", "6882")
query.Add("uploaded", "0")
query.Add("downloaded", "0")
query.Add("left", strconv.FormatInt(int64(metaInfo.GetTotalSize()), 10))
url.RawQuery = query.Encode()
request, err := http.NewRequest("GET", url.String(), nil)
if err != nil {
return
}
var httpClient = &http.Client{
Timeout: time.Second * time.Duration(15),
}
request.Header.Set("User-Agent", "gopeed/0.0.1")
response, err := httpClient.Do(request)
if err != nil {
return
}
defer response.Body.Close()
resp, err := ioutil.ReadAll(response.Body)
if err != nil {
if err != io.EOF {
return
}
}
// TODO Parse HTTP tracker response
fmt.Println(resp)
return nil, nil
}
// http://bittorrent.org/beps/bep_0015.html
func (tracker *Tracker) udpTracker(url *url.URL) (peers []peer.Peer, err error) {
conn, err := dial(url)
if err != nil {
return
}
defer conn.Close()
announceResponse, err := func() (announceResponse *udpAnnounceResponse, err error) {
var connectResponse *udpConnectResponse
// 0:connect 1:announce
state := 0
// 访问超时最多重试8次,当有请求成功则将重试次数重置为0
for n := 0; n < udpConnectRetries; n++ {
// 15 * 2 ^ n seconds
timeout := int64(udpConnectTimeout * math.Pow(2, float64(n)))
switch state {
case 0:
connectResponse, err = tracker.connect(conn, timeout)
if err != nil {
if oe, ok := err.(*net.OpError); ok && oe.Timeout() {
break
} else {
return
}
}
n = -1
state = 1
break
case 1:
announceResponse, err = tracker.announce(conn, timeout, connectResponse.connectionId)
if err != nil {
if oe, ok := err.(*net.OpError); ok && oe.Timeout() {
break
} else {
return
}
}
return
}
}
return
}()
if err != nil {
return
}
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 lookupIP(host string, scheme string) (ip net.IP, err error) {
ips, _ := net.LookupIP(host)
if len(ips) == 0 {
err = errors.New("no ips")
return
}
for _, ip = range ips {
switch scheme {
case "udp", "udp4":
if ip.To4() != nil {
return
}
case "udp6":
if ip.To4() == nil {
return
}
}
}
err = errors.New("no available ips ")
return
}
+26
View File
@@ -0,0 +1,26 @@
package tracker
import (
"fmt"
"github.com/monkeyWie/gopeed/protocol/bt/metainfo"
"testing"
)
func TestTracker_Tracker(t *testing.T) {
metaInfo, err := metainfo.ParseFromFile("../testdata/Game.of.Thrones.S08E05.720p.WEB.H264-MEMENTO.torrent")
if err != nil {
panic(err)
}
tracker := &Tracker{
PeerID: [20]byte{},
MetaInfo: metaInfo,
}
tracker.MetaInfo.Announce = "udp://tracker.opentrackr.org:1337/announce"
tracker.MetaInfo.AnnounceList = [][]string{}
peers, err := tracker.Tracker()
if err != nil {
panic(err)
return
}
fmt.Println(len(peers))
}
+182
View File
@@ -0,0 +1,182 @@
package http
import (
"fmt"
"io"
"mime"
"net/http"
"net/http/cookiejar"
"net/url"
"os"
"strconv"
"strings"
"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)
if err != nil {
return nil, err
}
// Use "Range" header to resolve
httpRequest.Header.Add("Range", "bytes=0-0")
httpClient := BuildHTTPClient()
response, err := httpClient.Do(httpRequest)
if err != nil {
return nil, err
}
defer response.Body.Close()
if response.StatusCode != 200 && response.StatusCode != 206 {
return nil, fmt.Errorf("response status error:%d", response.StatusCode)
}
ret := &Response{}
// Get file name by "Content-Disposition"
contentDisposition := response.Header.Get("Content-Disposition")
if contentDisposition != "" {
_, params, _ := mime.ParseMediaType(contentDisposition)
filename := params["filename"]
if filename != "" {
ret.Name = filename
}
}
// Get file name by URL
if ret.Name == "" {
parse, err := url.Parse(httpRequest.URL.String())
if err == nil {
// e.g. /files/test.txt => test.txt
ret.Name = subLastSlash(parse.Path)
}
}
// Unknow file name
if ret.Name == "" {
ret.Name = "unknow"
}
// Is support range
ret.Range = response.StatusCode == 206
// Get file size
if ret.Range {
contentRange := response.Header.Get("Content-Range")
if contentRange != "" {
// e.g. bytes 0-1000/1001 => 1001
total := subLastSlash(contentRange)
if total != "" && total != "*" {
parse, err := strconv.ParseInt(total, 10, 64)
if err != nil {
return nil, err
}
ret.Size = parse
}
}
} else {
contentLength := response.Header.Get("Content-Length")
if contentLength != "" {
ret.Size, _ = strconv.ParseInt(contentLength, 10, 64)
}
}
return ret, nil
}
// Down
func Down(request *Request) error {
response, err := Resolve(request)
if err != nil {
return err
}
// allocate file
file, err := os.Create(response.Name)
if err != nil {
return err
}
defer file.Close()
if err := file.Truncate(response.Size); err != nil {
return err
}
// support range
if response.Range {
cons := 16
chunkSize := response.Size / int64(cons)
waitGroup := &sync.WaitGroup{}
waitGroup.Add(cons)
for i := 0; i < cons; i++ {
start := int64(i) * chunkSize
end := start + chunkSize
if i == cons-1 {
end = response.Size
}
go downChunk(request, file, start, end-1, waitGroup)
}
waitGroup.Wait()
} else {
downChunk(request, file, 0, response.Size, nil)
}
return nil
}
func subLastSlash(str string) string {
index := strings.LastIndex(str, "/")
if index != -1 {
return str[index+1:]
}
return ""
}
func BuildHTTPRequest(request *Request) (*http.Request, error) {
// Build request
httpRequest, err := http.NewRequest(strings.ToUpper(request.Method), request.URL, nil)
if err != nil {
return nil, err
}
for k, v := range request.Header {
httpRequest.Header.Add(k, v)
}
return httpRequest, nil
}
func BuildHTTPClient() *http.Client {
// Cookie handle
jar, _ := cookiejar.New(nil)
return &http.Client{Jar: jar}
}
func downChunk(request *Request, file *os.File, start int64, end int64, waitGroup *sync.WaitGroup) {
if waitGroup != nil {
defer waitGroup.Done()
}
httpRequest, _ := BuildHTTPRequest(request)
httpRequest.Header.Add("Range", fmt.Sprintf("bytes=%d-%d", start, end))
fmt.Printf("down %d-%d\n", start, end)
httpClient := BuildHTTPClient()
httpResponse, err := httpClient.Do(httpRequest)
if err != nil {
fmt.Println(err)
return
}
defer httpResponse.Body.Close()
buf := make([]byte, 8192)
writeIndex := start
for {
n, err := httpResponse.Body.Read(buf)
if n > 0 {
writeSize, err := file.WriteAt(buf[0:n], writeIndex)
if err != nil {
fmt.Println(err)
return
}
writeIndex += int64(writeSize)
}
if err != nil {
if err != io.EOF {
fmt.Println(err)
return
}
break
}
}
}
+154
View File
@@ -0,0 +1,154 @@
package http
import (
"crypto/md5"
"encoding/hex"
"fmt"
"io"
"log"
"os"
"reflect"
"testing"
"time"
)
func TestDown(t *testing.T) {
type args struct {
request *Request
}
tests := []struct {
name string
args args
}{
{
"proxyee-down.ico",
args{
&Request{
"get",
"https://raw.githubusercontent.com/proxyee-down-org/proxyee-down/master/front/public/favicon.ico",
map[string]string{
"Host": "github.com",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/72.0.3626.121 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8",
"Referer": "http://github.com/proxyee-down-org/proxyee-down/releases",
"Accept-Encoding": "gzip, deflate, br",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
},
nil,
},
},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
os.Remove("favicon.ico")
err := Down(tt.args.request)
if err != nil {
t.Errorf("error down= %v", err)
return
}
downMd5 := fileMd5("favicon.ico")
os.Remove("favicon.ico")
if "8de7a6a2e786861013d61b77b2394012" != downMd5 {
t.Errorf("error md5= %v", downMd5)
return
}
})
}
}
func TestResolve(t *testing.T) {
// os.Setenv("HTTP_PROXY", "http://127.0.0.1:8888")
type args struct {
request *Request
}
tests := []struct {
name string
args args
want *Response
wantErr bool
}{
{
"proxyee-down.ico",
args{
&Request{
"get",
"https://raw.githubusercontent.com/proxyee-down-org/proxyee-down/master/front/public/favicon.ico",
map[string]string{
"Host": "github.com",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/72.0.3626.121 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8",
"Referer": "http://github.com/proxyee-down-org/proxyee-down/releases",
"Accept-Encoding": "gzip, deflate, br",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
},
nil,
},
},
&Response{"favicon.ico", 116093, true},
false,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got, err := Resolve(tt.args.request)
if (err != nil) != tt.wantErr {
t.Errorf("Resolve() error = %v, wantErr %v", err, tt.wantErr)
return
}
if !reflect.DeepEqual(got, tt.want) {
t.Errorf("Resolve() = %v, want %v", got, tt.want)
}
})
}
}
func TestTemp(t *testing.T) {
f, err := os.Create("e:/testbt/test.data")
if err != nil {
log.Fatal(err)
}
if err := f.Truncate(1024 * 1024 * 512); err != nil {
log.Fatal(err)
}
bts := []byte{1, 2, 3, 4, 5, 6, 7, 8, 9}
s := time.Now().UnixNano()
f.Write(bts)
fmt.Printf("write from 0 use: %d\n", time.Now().UnixNano()-s)
s = time.Now().UnixNano()
f.Seek(1024*1024*128, 0)
f.Write(bts)
fmt.Printf("write from 128M use %d\n", time.Now().UnixNano()-s)
s = time.Now().UnixNano()
f.Write(bts)
fmt.Printf("write from 128M+9b use %d\n", time.Now().UnixNano()-s)
s = time.Now().UnixNano()
f.Seek(0, 0)
f.Write(bts)
fmt.Printf("write from 0 agein use %d\n", time.Now().UnixNano()-s)
s = time.Now().UnixNano()
f.Seek(1024*1024*64, 0)
f.Write(bts)
fmt.Printf("write from agein 64M use %d\n", time.Now().UnixNano()-s)
}
func fileMd5(filePath string) string {
file, _ := os.Open(filePath)
// Tell the program to call the following function when the current function returns
defer file.Close()
// Open a new hash interface to write to
hash := md5.New()
// Copy the file in the hash interface and check for any error
if _, err := io.Copy(hash, file); err != nil {
return ""
}
return hex.EncodeToString(hash.Sum(nil))
}
func TestTemp2(t *testing.T) {
fmt.Println("123456789"[2:3])
}
+8
View File
@@ -0,0 +1,8 @@
package http
type Request struct {
Method string
URL string
Header map[string]string
Content []byte
}
+7
View File
@@ -0,0 +1,7 @@
package http
type Response struct {
Name string
Size int64
Range bool
}