webtorrent/transport.go

   1 package webtorrent
   2
   3 import (
   4         "expvar"
   5         "fmt"
   6         "io"
   7         "sync"
   8
   9         "github.com/anacrolix/missinggo/v2/pproffd"
  10         "github.com/pion/datachannel"
  11
  12         "github.com/pion/webrtc/v3"
  13 )
  14
  15 var (
  16         metrics = expvar.NewMap("webtorrent")
  17         api     = func() *webrtc.API {
  18                 // Enable the detach API (since it's non-standard but more idiomatic).
  19                 s := webrtc.SettingEngine{}
  20                 s.DetachDataChannels()
  21                 return webrtc.NewAPI(webrtc.WithSettingEngine(s))
  22         }()
  23         config              = webrtc.Configuration{ICEServers: []webrtc.ICEServer{{URLs: []string{"stun:stun.l.google.com:19302"}}}}
  24         newPeerConnectionMu sync.Mutex
  25 )
  26
  27 type wrappedPeerConnection struct {
  28         *webrtc.PeerConnection
  29         closeMu sync.Mutex
  30         pproffd.CloseWrapper
  31 }
  32
  33 func (me *wrappedPeerConnection) Close() error {
  34         me.closeMu.Lock()
  35         defer me.closeMu.Unlock()
  36         return me.CloseWrapper.Close()
  37 }
  38
  39 func newPeerConnection() (*wrappedPeerConnection, error) {
  40         newPeerConnectionMu.Lock()
  41         defer newPeerConnectionMu.Unlock()
  42         pc, err := api.NewPeerConnection(config)
  43         if err != nil {
  44                 return nil, err
  45         }
  46         return &wrappedPeerConnection{
  47                 PeerConnection: pc,
  48                 CloseWrapper:   pproffd.NewCloseWrapper(pc),
  49         }, nil
  50 }
  51
  52 // newOffer creates a transport and returns a WebRTC offer to be announced
  53 func newOffer() (
  54         peerConnection *wrappedPeerConnection,
  55         dataChannel *webrtc.DataChannel,
  56         offer webrtc.SessionDescription,
  57         err error,
  58 ) {
  59         peerConnection, err = newPeerConnection()
  60         if err != nil {
  61                 return
  62         }
  63         dataChannel, err = peerConnection.CreateDataChannel("webrtc-datachannel", nil)
  64         if err != nil {
  65                 peerConnection.Close()
  66                 return
  67         }
  68         offer, err = peerConnection.CreateOffer(nil)
  69         if err != nil {
  70                 peerConnection.Close()
  71                 return
  72         }
  73
  74         gatherComplete := webrtc.GatheringCompletePromise(peerConnection.PeerConnection)
  75         err = peerConnection.SetLocalDescription(offer)
  76         if err != nil {
  77                 peerConnection.Close()
  78                 return
  79         }
  80         <-gatherComplete
  81
  82         offer = *peerConnection.LocalDescription()
  83         return
  84 }
  85
  86 func initAnsweringPeerConnection(
  87         peerConnection *wrappedPeerConnection,
  88         offer webrtc.SessionDescription,
  89 ) (answer webrtc.SessionDescription, err error) {
  90         err = peerConnection.SetRemoteDescription(offer)
  91         if err != nil {
  92                 return
  93         }
  94         answer, err = peerConnection.CreateAnswer(nil)
  95         if err != nil {
  96                 return
  97         }
  98
  99         gatherComplete := webrtc.GatheringCompletePromise(peerConnection.PeerConnection)
 100         err = peerConnection.SetLocalDescription(answer)
 101         if err != nil {
 102                 return
 103         }
 104         <-gatherComplete
 105
 106         answer = *peerConnection.LocalDescription()
 107         return
 108 }
 109
 110 // newAnsweringPeerConnection creates a transport from a WebRTC offer and and returns a WebRTC answer to be
 111 // announced.
 112 func newAnsweringPeerConnection(offer webrtc.SessionDescription) (
 113         peerConn *wrappedPeerConnection, answer webrtc.SessionDescription, err error,
 114 ) {
 115         peerConn, err = newPeerConnection()
 116         if err != nil {
 117                 err = fmt.Errorf("failed to create new connection: %w", err)
 118                 return
 119         }
 120         answer, err = initAnsweringPeerConnection(peerConn, offer)
 121         if err != nil {
 122                 peerConn.Close()
 123         }
 124         return
 125 }
 126
 127 func (t *outboundOffer) setAnswer(answer webrtc.SessionDescription, onOpen func(datachannel.ReadWriteCloser)) error {
 128         setDataChannelOnOpen(t.dataChannel, t.peerConnection, onOpen)
 129         err := t.peerConnection.SetRemoteDescription(answer)
 130         return err
 131 }
 132
 133 type datachannelReadWriter interface {
 134         datachannel.Reader
 135         datachannel.Writer
 136         io.Reader
 137         io.Writer
 138 }
 139
 140 type ioCloserFunc func() error
 141
 142 func (me ioCloserFunc) Close() error {
 143         return me()
 144 }
 145
 146 func setDataChannelOnOpen(
 147         dc *webrtc.DataChannel,
 148         pc *wrappedPeerConnection,
 149         onOpen func(closer datachannel.ReadWriteCloser),
 150 ) {
 151         dc.OnOpen(func() {
 152                 raw, err := dc.Detach()
 153                 if err != nil {
 154                         // This shouldn't happen if the API is configured correctly, and we call from OnOpen.
 155                         panic(err)
 156                 }
 157                 onOpen(hookDataChannelCloser(raw, pc))
 158         })
 159 }
 160
 161 // Hooks the datachannel's Close to Close the owning PeerConnection. The datachannel takes ownership
 162 // and responsibility for the PeerConnection.
 163 func hookDataChannelCloser(dcrwc datachannel.ReadWriteCloser, pc *wrappedPeerConnection) datachannel.ReadWriteCloser {
 164         return struct {
 165                 datachannelReadWriter
 166                 io.Closer
 167         }{
 168                 dcrwc,
 169                 ioCloserFunc(pc.Close),
 170         }
 171 }