3 // get_peers and announce_peers.
9 "github.com/anacrolix/sync"
10 "github.com/willf/bloom"
12 "github.com/anacrolix/torrent/logonce"
13 "github.com/anacrolix/torrent/util"
16 type peerDiscovery struct {
18 triedAddrs *bloom.BloomFilter
24 announcePortImplied bool
27 func (pd *peerDiscovery) NumContacted() int {
30 return pd.numContacted
33 func (s *Server) Announce(infoHash string, port int, impliedPort bool) (*peerDiscovery, error) {
35 startAddrs := func() (ret []dHTAddr) {
36 for _, n := range s.closestGoodNodes(160, infoHash) {
37 ret = append(ret, n.addr)
42 if len(startAddrs) == 0 {
43 addrs, err := bootstrapAddrs()
47 for _, addr := range addrs {
48 startAddrs = append(startAddrs, newDHTAddr(addr))
51 disc := &peerDiscovery{
52 peerStream: &peerStream{
53 Values: make(chan peerStreamValue, 100),
54 stop: make(chan struct{}),
55 values: make(chan peerStreamValue),
57 triedAddrs: bloom.NewWithEstimates(1000, 0.5),
61 announcePortImplied: impliedPort,
63 // Function ferries from values to Values until discovery is halted.
65 defer close(disc.Values)
68 case psv := <-disc.values:
70 case disc.Values <- psv:
79 for i, addr := range startAddrs {
81 time.Sleep(time.Millisecond)
90 func (me *peerDiscovery) gotNodeAddr(addr dHTAddr) {
91 if util.AddrPort(addr) == 0 {
92 // Not a contactable address.
95 if me.triedAddrs.Test([]byte(addr.String())) {
98 if me.server.ipBlocked(util.AddrIP(addr)) {
104 func (me *peerDiscovery) contact(addr dHTAddr) {
106 me.triedAddrs.Add([]byte(addr.String()))
107 if err := me.getPeers(addr); err != nil {
108 log.Printf("error sending get_peers request to %s: %#v", addr, err)
114 func (me *peerDiscovery) transactionClosed() {
122 func (me *peerDiscovery) responseNode(node NodeInfo) {
123 me.gotNodeAddr(node.Addr)
126 func (me *peerDiscovery) closingCh() chan struct{} {
127 return me.peerStream.stop
130 func (me *peerDiscovery) announcePeer(to dHTAddr, token string) {
132 err := me.server.announcePeer(to, me.infoHash, me.announcePort, token, me.announcePortImplied)
133 me.server.mu.Unlock()
135 logonce.Stderr.Printf("error announcing peer: %s", err)
139 func (me *peerDiscovery) getPeers(addr dHTAddr) error {
141 defer me.server.mu.Unlock()
142 t, err := me.server.getPeers(addr, me.infoHash)
146 t.SetResponseHandler(func(m Msg) {
147 // Register suggested nodes closer to the target info-hash.
149 for _, n := range m.Nodes() {
154 if vs := m.Values(); vs != nil {
155 nodeInfo := NodeInfo{
158 copy(nodeInfo.ID[:], m.ID())
160 case me.peerStream.values <- peerStreamValue{
164 case <-me.peerStream.stop:
168 if at, ok := m.AnnounceToken(); ok {
169 me.announcePeer(addr, at)
173 me.transactionClosed()
179 type peerStreamValue struct {
180 Peers []util.CompactPeer // Peers given in get_peers response.
181 NodeInfo // The node that gave the response.
184 // TODO: This was to be the shared publicly accessible part returned by DHT
185 // functions that stream peers. Possibly not necessary anymore.
186 type peerStream struct {
188 Values chan peerStreamValue
189 // Inner chan is set to nil when on close.
190 values chan peerStreamValue
194 func (ps *peerStream) Close() {
200 func (ps *peerStream) close() {