nprasad2077/gopeed
1
0 Fork 0
mirror of https://github.com/GopeedLab/gopeed.git synced 2026-11-06 08:30:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

简单实现peer wire protocol协议

Browse Source
This commit is contained in:
liwei committed 2019-11-01 15:05:25 +08:00
1 parent ae1c605143
commit fb45066228
13 files changed
+859 -404
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

down/bt/metainfo/metainfo.go
+4 -3
View File
@@ -30,7 +30,7 @@ type MetaInfo struct {
type Info struct {
Name string `json:"name"`
PieceLength uint64 `json:"piece length"`
Pieces []string
Pieces [][20]byte
Length uint64 `json:"length"`
Files []File `json:"files"`
}
@@ -71,11 +71,12 @@ func ParseFromFile(path string) (*MetaInfo, error) {
// Split pieces hash
pieces := info["pieces"].(string)
if pieces != "" {
metaInfo.Info.Pieces = make([]string, len(pieces)/20)
metaInfo.Info.Pieces = make([][20]byte, len(pieces)/20)
// pieces parse
for i := range metaInfo.Info.Pieces {
begin := i * 20
metaInfo.Info.Pieces[i] = pieces[begin : begin+20]
metaInfo.Info.Pieces[i] = [20]byte{}
copy(metaInfo.Info.Pieces[i][:], pieces[begin:begin+20])
}
}
}
down/bt/metainfo/metainfo_test.go
+1 -1
View File
@@ -10,6 +10,6 @@ func TestParseFromFile(t *testing.T) {
if err != nil {
panic(err)
}
fmt.Printf("%+v\n", metaInfo)
fmt.Printf("%+v\n", metaInfo.AnnounceList)
fmt.Printf("%+v\n", metaInfo.Info.Files)
}
down/bt/peer/message/message.go
+9 -9
View File
@@ -8,15 +8,15 @@ import (
type ID byte
const (
IdChoke ID = 0
IdUnchoke ID = 1
IdInterested ID = 2
IdNotinterested ID = 3
IdHave ID = 4
IdBitfield ID = 5
IdRequest ID = 6
IdPiece ID = 7
IdCancel ID = 8
IdChoke ID = iota
IdUnchoke
IdInterested
IdNotinterested
IdHave
IdBitfield
IdRequest
IdPiece
IdCancel
)
/*type Message struct {
down/bt/torrent/peer_conn.go
+403
View File
@@ -0,0 +1,403 @@
package torrent
import (
"bufio"
"crypto/sha1"
"encoding/binary"
"errors"
"fmt"
"github.com/cenkalti/mse"
"github.com/monkeyWie/gopeed/down/bt/metainfo"
"github.com/monkeyWie/gopeed/down/bt/peer"
"github.com/monkeyWie/gopeed/down/bt/peer/message"
"io"
"math"
"net"
"os"
"path"
"sync"
"time"
)
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
pieceDownloadedCh chan bool
// block下载队列，官方推荐为5
pieceBlockQueue chan interface{}
// piece上所有的block下载状态
pieceBlockState sync.Map
}
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.pieceBlockQueue = make(chan interface{}, 5)
return nil
}
// 下载指定piece
func (pc *peerConn) downloadPiece(index int) error {
pc.pieceDownloadedCh = make(chan bool)
defer close(pc.pieceDownloadedCh)
pc.pieceBlockState = sync.Map{}
pieceLength := pc.torrent.MetaInfo.GetPieceSize(index)
lastOffset := uint64(0)
waitDuration := time.Second * 30
timer := time.NewTimer(waitDuration)
defer timer.Stop()
// 按块下载分片
for lastOffset < pieceLength {
blockLength := math.Min(2<<13, float64(pieceLength-lastOffset))
// 标记block开始下载
lastOffset32 := uint32(lastOffset)
pc.pieceBlockState.Store(lastOffset32, false)
// block下载排队，并做超时处理
timer.Reset(waitDuration)
select {
case pc.pieceBlockQueue <- nil:
case <-timer.C:
return errors.New("download timeout")
}
_, err := pc.conn.Write(message.NewRequest(uint32(index), lastOffset32, uint32(blockLength)).Encode())
if err != nil {
fmt.Println(err)
}
lastOffset += uint64(blockLength)
}
// 监听是否下载完成
<-pc.pieceDownloadedCh
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()
}
}
// 处理分片下载响应
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)
// 获取对应要的文件
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: f.Path[0],
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
}
}
}
// fmt.Println(fileBlocks)
for _, f := range fileBlocks {
func() {
name := path.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 {
fmt.Println(err)
}
}()
}
pc.pieceBlockState.Store(piece.Begin, true)
// 出队
<-pc.pieceBlockQueue
flag := true
// fmt.Printf("index %d: %v\n", piece.Index, pc.pieceBlockState)
pc.pieceBlockState.Range(func(key, value interface{}) bool {
// 如果有还没下载完的block
if !value.(bool) {
flag = false
return false
}
return true
})
// piece全部下载完
if flag {
// 计算piece对应的文件偏移
sha1 := sha1.New()
pieceLength := pc.torrent.MetaInfo.GetPieceSize(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 -= uint64(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.pieceDownloadedCh <- 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++ {
}
}
}
*/
down/bt/torrent/peer_conn_test.go
+147
View File
@@ -0,0 +1,147 @@
package torrent
import (
"fmt"
"github.com/monkeyWie/gopeed/down/bt/peer"
"github.com/monkeyWie/gopeed/down/bt/tracker"
"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)
}
fmt.Printf("Tracker end,peer count:%d\n", len(peers))
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")
}
down/bt/torrent/peer_pool.go
+82
View File
@@ -0,0 +1,82 @@
package torrent
import (
"github.com/monkeyWie/gopeed/down/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
}
down/bt/torrent/peer_pool_test.go
+40
View File
@@ -0,0 +1,40 @@
package torrent
import (
"fmt"
"github.com/monkeyWie/gopeed/down/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())
}
down/bt/torrent/peer_state.go
-292
View File
@@ -1,292 +0,0 @@
package torrent
import (
"bufio"
"encoding/binary"
"fmt"
"github.com/cenkalti/mse"
"github.com/monkeyWie/gopeed/down/bt/metainfo"
"github.com/monkeyWie/gopeed/down/bt/peer"
"github.com/monkeyWie/gopeed/down/bt/peer/message"
"io"
"math"
"net"
"os"
"path"
"time"
)
type peerState struct {
torrent *Torrent
*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
writeMsgCh chan []byte
bitfield *message.Bitfield
}
// 使用MSE加密来避免运营商对bt流量的封锁，基本上现在市面上BT客户端都默认开启了，不用MSE的话很多Peer拒绝连接
// see http://wiki.vuze.com/w/Message_Stream_Encryption
func (ps *peerState) dialMse() error {
conn, err := net.DialTimeout("tcp", ps.Address(), time.Second*time.Duration(30))
if err != nil {
return err
}
mseConn := mse.WrapConn(conn)
infoHash := ps.torrent.MetaInfo.GetInfoHash()
_, err = mseConn.HandshakeOutgoing(infoHash[:], mse.PlainText, nil)
if err != nil {
mseConn.Close()
return err
}
ps.conn = mseConn
return nil
}
// Handshake of Peer wire protocol
// see https://wiki.theory.org/index.php/BitTorrentSpecification#Handshake
func (ps *peerState) handshake() (*peer.Handshake, error) {
handshakeRes, err := func() (*peer.Handshake, error) {
handshakeReq := peer.NewHandshake([8]byte{}, ps.torrent.MetaInfo.GetInfoHash(), ps.torrent.PeerID)
_, err := ps.conn.Write(handshakeReq.Encode())
if err != nil {
return nil, err
}
var read [68]byte
_, err = io.ReadFull(ps.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 {
ps.conn.Close()
return nil, err
}
// init state
ps.amChoking = true
ps.amInterested = false
ps.peerChoking = true
ps.peerInterested = false
return handshakeRes, nil
}
func (ps *peerState) download() error {
// 异步写，防止读取阻塞
ps.writeMsgCh = make(chan []byte, 64)
go func() {
for buf := range ps.writeMsgCh {
_, err := ps.conn.Write(buf)
if err != nil {
fmt.Println(err)
ps.conn.Close()
}
}
}()
scanner := bufio.NewScanner(ps.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:
ps.handleUnchoke()
break
case message.IdInterested:
break
case message.IdNotinterested:
break
case message.IdHave:
break
case message.IdBitfield:
ps.handleBitfield(buf)
break
case message.IdRequest:
break
case message.IdPiece:
ps.handlePiece(buf)
break
case message.IdCancel:
break
}
}
}
return nil
}
func downloadPiece() {
}
func (ps *peerState) handleUnchoke() {
ps.peerChoking = false
// 如果客户端对peer感兴趣并且peer没有choked客户端，就可以开始下载了
if ps.amInterested {
have := ps.getHavePieces(ps.bitfield)
if len(have) > 0 {
// 获取分片的长度
for _, index := range have {
pieceLength := ps.torrent.MetaInfo.GetPieceSize(index)
lastOffset := uint64(0)
// 按块下载分片
for lastOffset < pieceLength {
blockLength := math.Min(2<<13, float64(pieceLength-lastOffset))
ps.write(message.NewRequest(uint32(index), uint32(lastOffset), uint32(blockLength)).Encode())
lastOffset += uint64(blockLength)
}
}
return
}
}
ps.conn.Close()
}
func (ps *peerState) handleBitfield(buf []byte) {
ps.bitfield = &message.Bitfield{}
ps.bitfield.Decode(buf)
have := ps.getHavePieces(ps.bitfield)
if len(have) > 0 {
// 表示对该peer感兴趣，并且不choked该peer
ps.write(message.NewInterested().Encode())
ps.amInterested = true
ps.write(message.NewUnchoke().Encode())
ps.amChoking = false
} else {
ps.conn.Close()
}
}
// 处理分片下载响应
func (ps *peerState) handlePiece(buf []byte) {
piece := &message.Piece{}
piece.Decode(buf)
info := ps.torrent.MetaInfo.Info
blockLength := int64(len(piece.Block))
pieceBegin := int64(piece.Index)*int64(info.PieceLength) + int64(piece.Begin)
// 获取对应要的文件
var fileBlocks []fileBlock
if len(info.Files) == 0 {
// 单文件
fileBlocks = append(fileBlocks, fileBlock{
filepath: info.Name,
fileSeek: pieceBegin,
blockBegin: 0,
blockEnd: blockLength,
})
} else {
// 获取要写入的第一个文件和偏移
writeIndex := getWriteFile(pieceBegin, ps.torrent.MetaInfo.GetFileDetails())
// block对应种子的偏移
blockBegin := pieceBegin
// block对应文件的偏移
var blockSeek int64 = 0
for _, f := range ps.torrent.MetaInfo.GetFileDetails()[writeIndex:] {
// 计算文件可剩余写入字节数
fileWritable := f.End - blockBegin
// 计算block剩余写入字节数
blockWritable := blockLength - blockSeek
fb := fileBlock{
filepath: f.Path[0],
fileSeek: blockBegin - 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)
blockBegin = f.End
blockSeek += fileWritable
}
}
}
fmt.Println(fileBlocks)
for _, f := range fileBlocks {
func() {
name := path.Join("e:/testbt/download", 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 {
fmt.Println(err)
}
}()
}
}
// 获取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 (ps *peerState) getHavePieces(bitfield *message.Bitfield) []int {
states := make([]bool, len(ps.torrent.PieceState))
for i := range states {
states[i] = ps.torrent.PieceState[i].complete
}
return bitfield.Have(states)
}
// 异步写入数据
func (ps *peerState) write(buf []byte) {
ps.writeMsgCh <- buf
}
// 获取要写入到的文件
/*func (ps *peerState) 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++ {
}
}
}
*/
down/bt/torrent/peer_state_test.go
-85
View File
@@ -1,85 +0,0 @@
package torrent
import (
"fmt"
"github.com/monkeyWie/gopeed/down/bt/metainfo"
"github.com/monkeyWie/gopeed/down/bt/peer"
"github.com/monkeyWie/gopeed/down/bt/tracker"
"testing"
)
func Test_peerState_handshake(t *testing.T) {
ps := getUsablePeer()
// 发出握手请求
handshake, err := ps.handshake()
fmt.Println("Handshake end")
if err != nil {
fmt.Println(ps.Address())
panic(err)
}
fmt.Println(handshake)
}
func Test_peerState_download(t *testing.T) {
ps := getUsablePeer()
// 发出握手请求
_, err := ps.handshake()
if err != nil {
panic(err)
}
fmt.Println("handshake end")
// 开始下载
err = ps.download()
}
func buildTorrent() *Torrent {
metaInfo, err := metainfo.ParseFromFile("../testdata/Game.of.Thrones.S08E05.720p.WEB.H264-MEMENTO.torrent")
if err != nil {
panic(err)
}
metaInfo.Announce = "udp://exodus.desync.com:6969/announce"
metaInfo.AnnounceList = [][]string{}
return NewTorrent(peer.GenPeerID(), metaInfo)
}
// 获取一个能连接上的peer
func getUsablePeer() *peerState {
torrent := buildTorrent()
tracker := &tracker.Tracker{
PeerID: torrent.PeerID,
MetaInfo: torrent.MetaInfo,
}
peers, err := tracker.Tracker()
if err != nil {
panic(err)
}
fmt.Println("Tracker end")
ch := make(chan *peerState, 1)
for i := range peers {
go peerTest(torrent, &peers[i], ch)
}
return <-ch
}
func peerTest(torrent *Torrent, peer *peer.Peer, ch chan *peerState) {
ps := &peerState{
torrent: torrent,
Peer: peer,
}
// 连接至peer
err := ps.dialMse()
if err != nil {
return
}
ch <- ps
}
func TestSome(t *testing.T) {
fmt.Println(11)
}
down/bt/torrent/piece_state.go
+53 -2
View File
@@ -1,5 +1,56 @@
package torrent
type pieceState struct {
complete bool
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)
}
down/bt/torrent/torrent.go
+85 -7
View File
@@ -1,22 +1,100 @@
package torrent
import (
"fmt"
"github.com/monkeyWie/gopeed/down/bt/metainfo"
"github.com/monkeyWie/gopeed/down/bt/tracker"
"sync"
"time"
)
type Torrent struct {
PeerID [20]byte
MetaInfo *metainfo.MetaInfo
PieceState []pieceState
PeerID [20]byte
MetaInfo *metainfo.MetaInfo
Peers []peerState
PiecesState *piecesState
peerPool *peerPool
Path string
}
func NewTorrent(peerID [20]byte, metaInfo *metainfo.MetaInfo) *Torrent {
torrent := &Torrent{
PeerID: peerID,
MetaInfo: metaInfo,
PieceState: make([]pieceState, len(metaInfo.Info.Pieces)),
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 {
fmt.Println("no peer")
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 {
t.PiecesState.setState(index, stateDownloading)
}
t.peerPool.release(peer)
}
<-taskCh
}(index)
}
}
// 获取可用的peer
func (t *Torrent) fetchPeers() {
tracker := &tracker.Tracker{
PeerID: t.PeerID,
MetaInfo: t.MetaInfo,
}
urls := []string{
"udp://exodus.desync.com:6969/announce",
"udp://tracker.openbittorrent.com:80/announce",
"udp://tracker.leechers-paradise.org:6969",
}
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.Println("tracker end:" + url)
ch <- nil
t.peerPool.put(peers)
}
}(url)
}
go func() {
wg.Wait()
close(ch)
}()
<-ch
}
down/bt/torrent/torrent_test.go
+32
View File
@@ -0,0 +1,32 @@
package torrent
import (
"github.com/monkeyWie/gopeed/down/bt/metainfo"
"github.com/monkeyWie/gopeed/down/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")
if err != nil {
panic(err)
}
// metaInfo.Announce = "udp://exodus.desync.com:6969/announce"
// metaInfo.Announce = "udp://tracker.openbittorrent.com:80/announce"
// metaInfo.Announce = "udp://tracker.cyberia.is:6969/announce"
// metaInfo.Announce = "udp://9.rarbg.me:2780/announce"
// metaInfo.Announce = "udp://9.rarbg.to:2760/announce"
// metaInfo.Announce = "udp://tracker.leechers-paradise.org:6969"
metaInfo.AnnounceList = [][]string{}
metaInfo.AnnounceList = append(metaInfo.AnnounceList, []string{
"udp://exodus.desync.com:6969/announce",
"udp://tracker.openbittorrent.com:80/announce",
"udp://tracker.leechers-paradise.org:6969",
})
return NewTorrent(peer.GenPeerID(), metaInfo)
}
down/bt/tracker/tracker.go
+3 -5
View File
@@ -246,7 +246,7 @@ func (tracker *Tracker) Tracker() (peers []peer.Peer, err error) {
for _, announceArr := range metaInfo.AnnounceList {
if len(announceArr) > 0 {
for _, announce := range announceArr {
peers, err = tracker.doTracker(announce)
peers, err = tracker.DoTracker(announce)
if err == nil {
return
}
@@ -254,18 +254,16 @@ func (tracker *Tracker) Tracker() (peers []peer.Peer, err error) {
}
}
} else {
peers, err = tracker.doTracker(metaInfo.Announce)
peers, err = tracker.DoTracker(metaInfo.Announce)
}
return
}
func (tracker *Tracker) doTracker(announce string) (peers []peer.Peer, err error) {
func (tracker *Tracker) DoTracker(announce string) (peers []peer.Peer, err error) {
if announce != "" {
url, _ := url.Parse(announce)
switch url.Scheme {
case "http", "https":
// TODO UDP test
break
return tracker.httpTracker(url)
case "udp", "udp4", "udp6":
return tracker.udpTracker(url)