<commit_before>package rain

import (
	"bytes"
	"encoding/binary"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"net"
	"time"

	"github.com/cenkalti/log"
	"github.com/zeebo/bencode"

	"github.com/cenkalti/rain/internal/magnet"
	"github.com/cenkalti/rain/internal/protocol"
	"github.com/cenkalti/rain/internal/torrent"
	"github.com/cenkalti/rain/internal/tracker"
)

const (
	metadataPieceSize           = 16 * 1024
	concurrentMetadataDownloads = 1000
	metadataNetworkTimeout      = 2 * time.Minute
)

// Extension IDs
const (
	extensionHandshakeID = iota
	extensionMetadataID
)

// Metadata Extension Message Types
const (
	metadataRequest = iota
	metadataData
	metadataReject
)

type MetadataDownloader struct {
	magnet    *magnet.Magnet
	tracker   tracker.Tracker
	announceC chan *tracker.AnnounceResponse
	peerC     chan tracker.Peer
	Result    chan *torrent.Info
	cancel    chan struct{}
}

func NewMetadataDownloader(m *magnet.Magnet) (*MetadataDownloader, error) {
	if len(m.Trackers) == 0 {
		return nil, errors.New("magnet link does not contain a tracker")
	}
	c, err := newDummyClient()
	if err != nil {
		return nil, err
	}
	tr, err := tracker.New(m.Trackers[0], c)
	if err != nil {
		return nil, err
	}
	return &MetadataDownloader{
		magnet:    m,
		tracker:   tr,
		announceC: make(chan *tracker.AnnounceResponse),
		peerC:     make(chan tracker.Peer),
		Result:    make(chan *torrent.Info, 1),
		cancel:    make(chan struct{}),
	}, nil
}

func (m *MetadataDownloader) Run() {
	t := emptyTransfer(m.magnet.InfoHash)
	go m.tracker.Announce(&t, m.cancel, nil, m.announceC)
	for i := 0; i < concurrentMetadataDownloads; i++ {
		go m.worker()
	}
	for resp := range m.announceC {
		for _, p := range resp.Peers {
			m.peerC <- p
		}
	}
}

func (m *MetadataDownloader) worker() {
	for {
		select {
		case peer := <-m.peerC:
			conn, err := net.DialTCP("tcp4", nil, peer.TCPAddr())
			if err != nil {
				log.Error(err)
				continue
			}

			p := newPeer(conn)
			p.log.Debug("tcp connection is opened")

			info, err := downloadMetadataFromPeer(m.magnet, p)
			conn.Close()
			if err != nil {
				p.log.Error(err)
				continue
			}

			select {
			case m.Result <- info:
				close(m.cancel) // will stop other workers
			case <-m.cancel:
				return
			}
		case <-m.cancel:
			return
		}
	}
}

func downloadMetadataFromPeer(m *magnet.Magnet, p *peer) (*torrent.Info, error) {<commit_msg>no peerC<commit_after>package rain

import (
	"bytes"
	"encoding/binary"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"net"
	"time"

	"github.com/cenkalti/log"
	"github.com/zeebo/bencode"

	"github.com/cenkalti/rain/internal/magnet"
	"github.com/cenkalti/rain/internal/protocol"
	"github.com/cenkalti/rain/internal/torrent"
	"github.com/cenkalti/rain/internal/tracker"
)

const (
	metadataPieceSize           = 16 * 1024
	concurrentMetadataDownloads = 1000
	metadataNetworkTimeout      = 2 * time.Minute
)

// Extension IDs
const (
	extensionHandshakeID = iota
	extensionMetadataID
)

// Metadata Extension Message Types
const (
	metadataRequest = iota
	metadataData
	metadataReject
)

type MetadataDownloader struct {
	magnet    *magnet.Magnet
	tracker   tracker.Tracker
	announceC chan *tracker.AnnounceResponse
	Result    chan *torrent.Info
	cancel    chan struct{}
}

func NewMetadataDownloader(m *magnet.Magnet) (*MetadataDownloader, error) {
	if len(m.Trackers) == 0 {
		return nil, errors.New("magnet link does not contain a tracker")
	}
	c, err := newDummyClient()
	if err != nil {
		return nil, err
	}
	tr, err := tracker.New(m.Trackers[0], c)
	if err != nil {
		return nil, err
	}
	return &MetadataDownloader{
		magnet:    m,
		tracker:   tr,
		announceC: make(chan *tracker.AnnounceResponse),
		Result:    make(chan *torrent.Info, 1),
		cancel:    make(chan struct{}),
	}, nil
}

func (m *MetadataDownloader) Run() {
	t := emptyTransfer(m.magnet.InfoHash)
	go m.tracker.Announce(&t, m.cancel, nil, m.announceC)
	for resp := range m.announceC {
		for _, p := range resp.Peers {
			go m.worker(p)
		}
	}
}

func (m *MetadataDownloader) worker(peer tracker.Peer) {
	conn, err := net.DialTCP("tcp4", nil, peer.TCPAddr())
	if err != nil {
		log.Error(err)
	}

	p := newPeer(conn)
	p.log.Debug("tcp connection is opened")

	info, err := downloadMetadataFromPeer(m.magnet, p)
	conn.Close()
	if err != nil {
		p.log.Error(err)
	}

	select {
	case m.Result <- info:
		close(m.cancel) // will stop other workers
	case <-m.cancel:
		return
	}
}

func downloadMetadataFromPeer(m *magnet.Magnet, p *peer) (*torrent.Info, error) {
{'fraction_non_alphanumeric': 0.13349622390048868, 'fraction_numerical': 0.006885828520657486, 'mean_word_length': 3.4940119760479043, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 19, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2517386', 'n_tokens_mistral': 1837, 'n_tokens_neox': 1687, 'n_words': 382}