webtorrent/client.go

   1 package main
   2
   3 import (
   4         "encoding/json"
   5         "fmt"
   6         "io"
   7         "log"
   8         "sync"
   9
  10         "github.com/anacrolix/torrent/metainfo"
  11         "github.com/anacrolix/torrent/webtorrent/buffer"
  12         "github.com/gorilla/websocket"
  13         "github.com/pion/datachannel"
  14         "github.com/pion/webrtc/v2"
  15 )
  16
  17 const (
  18         trackerURL = `wss://tracker.openwebtorrent.com/` // For simplicity
  19 )
  20
  21 // Client represents the webtorrent client
  22 type Client struct {
  23         peerID       string
  24         peerIDBinary string
  25
  26         infoHash       string
  27         infoHashBinary string
  28         totalLength    int
  29
  30         offeredPeers map[string]Peer // OfferID to Peer
  31
  32         tracker *websocket.Conn
  33
  34         lock *sync.Mutex
  35 }
  36
  37 // Peer represents a remote peer
  38 type Peer struct {
  39         peerID    string
  40         transport *Transport
  41 }
  42
  43 func NewClient() (*Client, error) {
  44         c := &Client{
  45                 offeredPeers: make(map[string]Peer),
  46                 lock:         &sync.Mutex{},
  47         }
  48
  49         randPeerID, err := buffer.RandomBytes(9)
  50         if err != nil {
  51                 return nil, fmt.Errorf("failed to generate bytes: %v", err)
  52         }
  53         peerIDBuffer := buffer.From("-WW0007-" + randPeerID.ToStringBase64())
  54         c.peerID = peerIDBuffer.ToStringHex()
  55         c.peerIDBinary = peerIDBuffer.ToStringLatin1()
  56
  57         return c, nil
  58 }
  59
  60 func (c *Client) LoadFile(p string) error {
  61         meta, err := metainfo.LoadFromFile(p)
  62         if err != nil {
  63                 return fmt.Errorf("failed to load meta info: %v\n", err)
  64         }
  65
  66         info, err := meta.UnmarshalInfo()
  67         if err != nil {
  68                 return fmt.Errorf("failed to unmarshal info: %v\n", err)
  69         }
  70         c.totalLength = int(info.TotalLength())
  71
  72         c.infoHash = meta.HashInfoBytes().String()
  73         b, err := buffer.FromHex(c.infoHash)
  74         if err != nil {
  75                 return fmt.Errorf("failed to create buffer: %v\n", err)
  76         }
  77         c.infoHashBinary = b.ToStringLatin1()
  78
  79         return nil
  80 }
  81
  82 func (c *Client) Run() error {
  83         t, _, err := websocket.DefaultDialer.Dial(trackerURL, nil)
  84         if err != nil {
  85                 return fmt.Errorf("failed to dial tracker: %v", err)
  86         }
  87         defer t.Close()
  88         c.tracker = t
  89
  90         go c.announce()
  91         c.trackerReadLoop()
  92
  93         return nil
  94 }
  95
  96 func (c *Client) announce() {
  97         transpot, offer, err := NewTransport()
  98         if err != nil {
  99                 log.Fatalf("failed to create transport: %v\n", err)
 100         }
 101
 102         randOfferID, err := buffer.RandomBytes(20)
 103         if err != nil {
 104                 log.Fatalf("failed to generate bytes: %v\n", err)
 105         }
 106         // OfferID := randOfferID.ToStringHex()
 107         offerIDBinary := randOfferID.ToStringLatin1()
 108
 109         c.lock.Lock()
 110         c.offeredPeers[offerIDBinary] = Peer{transport: transpot}
 111         c.lock.Unlock()
 112
 113         req := AnnounceRequest{
 114                 Numwant:    1, // If higher we need to create equal amount of offers
 115                 Uploaded:   0,
 116                 Downloaded: 0,
 117                 Left:       int(c.totalLength),
 118                 Event:      "started",
 119                 Action:     "announce",
 120                 InfoHash:   c.infoHashBinary,
 121                 PeerID:     c.peerIDBinary,
 122                 Offers: []Offer{
 123                         {
 124                                 OfferID: offerIDBinary,
 125                                 Offer:   offer,
 126                         }},
 127         }
 128
 129         data, err := json.Marshal(req)
 130         if err != nil {
 131                 log.Fatal("failed to marshal request:", err)
 132         }
 133         c.lock.Lock()
 134         tracker := c.tracker
 135         err = tracker.WriteMessage(websocket.TextMessage, data)
 136         if err != nil {
 137                 log.Fatal("write AnnounceRequest:", err)
 138                 c.lock.Unlock()
 139         }
 140         c.lock.Unlock()
 141 }
 142
 143 func (c *Client) trackerReadLoop() {
 144
 145         c.lock.Lock()
 146         tracker := c.tracker
 147         c.lock.Unlock()
 148         for {
 149                 _, message, err := tracker.ReadMessage()
 150                 if err != nil {
 151                         log.Fatalf("read error: %v", err)
 152                 }
 153                 log.Printf("recv: %s", message)
 154
 155                 var ar AnnounceResponse
 156                 if err := json.Unmarshal(message, &ar); err != nil {
 157                         log.Printf("error unmarshaling announce response: %v", err)
 158                         continue
 159                 }
 160                 if ar.InfoHash != c.infoHashBinary {
 161                         log.Printf("announce response for different hash: %s", ar.InfoHash)
 162                         continue
 163                 }
 164                 switch {
 165                 case ar.Offer != nil:
 166                         t, answer, err := NewTransportFromOffer(*ar.Offer, c.handleDataChannel)
 167                         if err != nil {
 168                                 log.Fatal("write AnnounceResponse:", err)
 169                         }
 170
 171                         req := AnnounceResponse{
 172                                 Action:   "announce",
 173                                 InfoHash: c.infoHashBinary,
 174                                 PeerID:   c.peerIDBinary,
 175                                 ToPeerID: ar.PeerID,
 176                                 Answer:   &answer,
 177                                 OfferID:  ar.OfferID,
 178                         }
 179                         data, err := json.Marshal(req)
 180                         if err != nil {
 181                                 log.Fatal("failed to marshal request:", err)
 182                         }
 183
 184                         c.lock.Lock()
 185                         err = tracker.WriteMessage(websocket.TextMessage, data)
 186                         if err != nil {
 187                                 log.Fatal("write AnnounceResponse:", err)
 188                                 c.lock.Unlock()
 189                         }
 190                         c.lock.Unlock()
 191
 192                         // Do something with the peer
 193                         _ = Peer{peerID: ar.PeerID, transport: t}
 194                 case ar.Answer != nil:
 195                         c.lock.Lock()
 196                         peer, ok := c.offeredPeers[ar.OfferID]
 197                         c.lock.Unlock()
 198                         if !ok {
 199                                 fmt.Printf("could not find peer for offer %s", ar.OfferID)
 200                                 continue
 201                         }
 202                         err = peer.transport.SetAnswer(*ar.Answer, c.handleDataChannel)
 203                         if err != nil {
 204                                 log.Fatalf("failed to sent answer: %v", err)
 205                         }
 206                 }
 207         }
 208 }
 209
 210 func (c *Client) handleDataChannel(dc datachannel.ReadWriteCloser) {
 211         go c.dcReadLoop(dc)
 212         //go c.dcWriteLoop(dc)
 213 }
 214
 215 func (c *Client) dcReadLoop(d io.Reader) {
 216         for {
 217                 buffer := make([]byte, 1024)
 218                 n, err := d.Read(buffer)
 219                 if err != nil {
 220                         log.Fatal("Datachannel closed; Exit the readloop:", err)
 221                 }
 222
 223                 fmt.Printf("Message from DataChannel: %s\n", string(buffer[:n]))
 224         }
 225 }
 226
 227 type AnnounceRequest struct {
 228         Numwant    int     `json:"numwant"`
 229         Uploaded   int     `json:"uploaded"`
 230         Downloaded int     `json:"downloaded"`
 231         Left       int     `json:"left"`
 232         Event      string  `json:"event"`
 233         Action     string  `json:"action"`
 234         InfoHash   string  `json:"info_hash"`
 235         PeerID     string  `json:"peer_id"`
 236         Offers     []Offer `json:"offers"`
 237 }
 238
 239 type Offer struct {
 240         OfferID string                    `json:"offer_id"`
 241         Offer   webrtc.SessionDescription `json:"offer"`
 242 }
 243
 244 type AnnounceResponse struct {
 245         InfoHash   string                     `json:"info_hash"`
 246         Action     string                     `json:"action"`
 247         Interval   *int                       `json:"interval,omitempty"`
 248         Complete   *int                       `json:"complete,omitempty"`
 249         Incomplete *int                       `json:"incomplete,omitempty"`
 250         PeerID     string                     `json:"peer_id,omitempty"`
 251         ToPeerID   string                     `json:"to_peer_id,omitempty"`
 252         Answer     *webrtc.SessionDescription `json:"answer,omitempty"`
 253         Offer      *webrtc.SessionDescription `json:"offer,omitempty"`
 254         OfferID    string                     `json:"offer_id,omitempty"`
 255 }