10 "github.com/anacrolix/log"
12 "github.com/anacrolix/torrent/tracker"
13 "github.com/gorilla/websocket"
14 "github.com/pion/datachannel"
15 "github.com/pion/webrtc/v3"
18 type TrackerClientStats struct {
20 ConvertedInboundConns int64
21 ConvertedOutboundConns int64
24 // Client represents the webtorrent client
25 type TrackerClient struct {
27 GetAnnounceRequest func(_ tracker.AnnounceEvent, infoHash [20]byte) (tracker.AnnounceRequest, error)
29 OnConn onDataChannelOpen
34 outboundOffers map[string]outboundOffer // OfferID to outboundOffer
35 wsConn *websocket.Conn
37 stats TrackerClientStats
38 pingTicker *time.Ticker
41 func (me *TrackerClient) Stats() TrackerClientStats {
47 func (me *TrackerClient) peerIdBinary() string {
48 return binaryToJsonString(me.PeerId[:])
51 // outboundOffer represents an outstanding offer.
52 type outboundOffer struct {
53 originalOffer webrtc.SessionDescription
54 peerConnection *wrappedPeerConnection
55 dataChannel *webrtc.DataChannel
59 type DataChannelContext struct {
60 Local, Remote webrtc.SessionDescription
66 type onDataChannelOpen func(_ datachannel.ReadWriteCloser, dcc DataChannelContext)
68 func (tc *TrackerClient) doWebsocket() error {
69 metrics.Add("websocket dials", 1)
71 c, _, err := websocket.DefaultDialer.Dial(tc.Url, nil)
73 return fmt.Errorf("dialing tracker: %w", err)
76 tc.Logger.WithDefaultLevel(log.Info).Printf("connected")
82 closeChan := make(chan struct{})
86 case <-tc.pingTicker.C:
88 err := c.WriteMessage(websocket.PingMessage, []byte{})
99 err = tc.trackerReadLoop(tc.wsConn)
107 func (tc *TrackerClient) Run() error {
108 tc.pingTicker = time.NewTicker(60 * time.Second)
113 err := tc.doWebsocket()
120 tc.Logger.WithDefaultLevel(level).Printf("websocket instance ended: %v", err)
121 time.Sleep(time.Minute)
128 func (tc *TrackerClient) Close() error {
131 if tc.wsConn != nil {
134 tc.closeUnusedOffers()
141 func (tc *TrackerClient) announceOffers() {
143 // tc.Announce grabs a lock on tc.outboundOffers. It also handles the case where outboundOffers
144 // is nil. Take ownership of outboundOffers here.
146 offers := tc.outboundOffers
147 tc.outboundOffers = nil
154 // Iterate over our locally-owned offers, close any existing "invalid" ones from before the
155 // socket reconnected, reannounce the infohash, adding it back into the tc.outboundOffers.
156 tc.Logger.WithDefaultLevel(log.Info).Printf("reannouncing %d infohashes after restart", len(offers))
157 for _, offer := range offers {
158 // TODO: Capture the errors? Are we even in a position to do anything with them?
159 offer.peerConnection.Close()
160 // Use goroutine here to allow read loop to start and ensure the buffer drains.
161 go tc.Announce(tracker.Started, offer.infoHash)
165 func (tc *TrackerClient) closeUnusedOffers() {
166 for _, offer := range tc.outboundOffers {
167 offer.peerConnection.Close()
169 tc.outboundOffers = nil
172 func (tc *TrackerClient) Announce(event tracker.AnnounceEvent, infoHash [20]byte) error {
173 metrics.Add("outbound announces", 1)
174 var randOfferId [20]byte
175 _, err := rand.Read(randOfferId[:])
177 return fmt.Errorf("generating offer_id bytes: %w", err)
179 offerIDBinary := binaryToJsonString(randOfferId[:])
181 pc, dc, offer, err := newOffer()
183 return fmt.Errorf("creating offer: %w", err)
186 request, err := tc.GetAnnounceRequest(event, infoHash)
189 return fmt.Errorf("getting announce parameters: %w", err)
192 req := AnnounceRequest{
193 Numwant: 1, // If higher we need to create equal amount of offers.
194 Uploaded: request.Uploaded,
195 Downloaded: request.Downloaded,
197 Event: request.Event.String(),
199 InfoHash: binaryToJsonString(infoHash[:]),
200 PeerID: tc.peerIdBinary(),
202 OfferID: offerIDBinary,
207 data, err := json.Marshal(req)
210 return fmt.Errorf("marshalling request: %w", err)
215 err = tc.writeMessage(data)
218 return fmt.Errorf("write AnnounceRequest: %w", err)
220 if tc.outboundOffers == nil {
221 tc.outboundOffers = make(map[string]outboundOffer)
223 tc.outboundOffers[offerIDBinary] = outboundOffer{
226 originalOffer: offer,
232 func (tc *TrackerClient) writeMessage(data []byte) error {
233 for tc.wsConn == nil {
235 return fmt.Errorf("%T closed", tc)
239 return tc.wsConn.WriteMessage(websocket.TextMessage, data)
242 func (tc *TrackerClient) trackerReadLoop(tracker *websocket.Conn) error {
244 _, message, err := tracker.ReadMessage()
246 return fmt.Errorf("read message error: %w", err)
248 //tc.Logger.WithDefaultLevel(log.Debug).Printf("received message from tracker: %q", message)
250 var ar AnnounceResponse
251 if err := json.Unmarshal(message, &ar); err != nil {
252 tc.Logger.WithDefaultLevel(log.Warning).Printf("error unmarshalling announce response: %v", err)
256 case ar.Offer != nil:
257 ih, err := jsonStringToInfoHash(ar.InfoHash)
259 tc.Logger.WithDefaultLevel(log.Warning).Printf("error decoding info_hash in offer: %v", err)
262 tc.handleOffer(*ar.Offer, ar.OfferID, ih, ar.PeerID)
263 case ar.Answer != nil:
264 tc.handleAnswer(ar.OfferID, *ar.Answer)
269 func (tc *TrackerClient) handleOffer(
270 offer webrtc.SessionDescription,
275 peerConnection, answer, err := newAnsweringPeerConnection(offer)
277 return fmt.Errorf("write AnnounceResponse: %w", err)
279 response := AnnounceResponse{
281 InfoHash: binaryToJsonString(infoHash[:]),
282 PeerID: tc.peerIdBinary(),
287 data, err := json.Marshal(response)
289 peerConnection.Close()
290 return fmt.Errorf("marshalling response: %w", err)
294 if err := tc.writeMessage(data); err != nil {
295 peerConnection.Close()
296 return fmt.Errorf("writing response: %w", err)
298 timer := time.AfterFunc(30*time.Second, func() {
299 metrics.Add("answering peer connections timed out", 1)
300 peerConnection.Close()
302 peerConnection.OnDataChannel(func(d *webrtc.DataChannel) {
303 setDataChannelOnOpen(d, peerConnection, func(dc datachannel.ReadWriteCloser) {
305 metrics.Add("answering peer connection conversions", 1)
307 tc.stats.ConvertedInboundConns++
309 tc.OnConn(dc, DataChannelContext{
321 func (tc *TrackerClient) handleAnswer(offerId string, answer webrtc.SessionDescription) {
324 offer, ok := tc.outboundOffers[offerId]
326 tc.Logger.WithDefaultLevel(log.Warning).Printf("could not find offer for id %q", offerId)
329 //tc.Logger.WithDefaultLevel(log.Debug).Printf("offer %q got answer %v", offerId, answer)
330 metrics.Add("outbound offers answered", 1)
331 err := offer.setAnswer(answer, func(dc datachannel.ReadWriteCloser) {
332 metrics.Add("outbound offers answered with datachannel", 1)
334 tc.stats.ConvertedOutboundConns++
336 tc.OnConn(dc, DataChannelContext{
337 Local: offer.originalOffer,
341 InfoHash: offer.infoHash,
345 tc.Logger.WithDefaultLevel(log.Warning).Printf("error using outbound offer answer: %v", err)
348 delete(tc.outboundOffers, offerId)
349 go tc.Announce(tracker.None, offer.infoHash)