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/v2"
  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         pproffd.CloseWrapper
  30 }
  31
  32 func (me wrappedPeerConnection) Close() error {
  33         return me.CloseWrapper.Close()
  34 }
  35
  36 func newPeerConnection() (wrappedPeerConnection, error) {
  37         newPeerConnectionMu.Lock()
  38         defer newPeerConnectionMu.Unlock()
  39         pc, err := api.NewPeerConnection(config)
  40         if err != nil {
  41                 return wrappedPeerConnection{}, err
  42         }
  43         return wrappedPeerConnection{
  44                 pc,
  45                 pproffd.NewCloseWrapper(pc),
  46         }, nil
  47 }
  48
  49 // newOffer creates a transport and returns a WebRTC offer to be announced
  50 func newOffer() (
  51         peerConnection wrappedPeerConnection,
  52         dataChannel *webrtc.DataChannel,
  53         offer webrtc.SessionDescription,
  54         err error,
  55 ) {
  56         peerConnection, err = newPeerConnection()
  57         if err != nil {
  58                 return
  59         }
  60         dataChannel, err = peerConnection.CreateDataChannel("webrtc-datachannel", nil)
  61         if err != nil {
  62                 peerConnection.Close()
  63                 return
  64         }
  65         offer, err = peerConnection.CreateOffer(nil)
  66         if err != nil {
  67                 peerConnection.Close()
  68                 return
  69         }
  70         err = peerConnection.SetLocalDescription(offer)
  71         if err != nil {
  72                 peerConnection.Close()
  73                 return
  74         }
  75         return
  76 }
  77
  78 func initAnsweringPeerConnection(
  79         peerConnection wrappedPeerConnection,
  80         offer webrtc.SessionDescription,
  81 ) (answer webrtc.SessionDescription, err error) {
  82         err = peerConnection.SetRemoteDescription(offer)
  83         if err != nil {
  84                 return
  85         }
  86         answer, err = peerConnection.CreateAnswer(nil)
  87         if err != nil {
  88                 return
  89         }
  90         err = peerConnection.SetLocalDescription(answer)
  91         if err != nil {
  92                 return
  93         }
  94         return
  95 }
  96
  97 // newAnsweringPeerConnection creates a transport from a WebRTC offer and and returns a WebRTC answer to be
  98 // announced.
  99 func newAnsweringPeerConnection(offer webrtc.SessionDescription) (
 100         peerConn wrappedPeerConnection, answer webrtc.SessionDescription, err error,
 101 ) {
 102         peerConn, err = newPeerConnection()
 103         if err != nil {
 104                 err = fmt.Errorf("failed to create new connection: %w", err)
 105                 return
 106         }
 107         answer, err = initAnsweringPeerConnection(peerConn, offer)
 108         if err != nil {
 109                 peerConn.Close()
 110         }
 111         return
 112 }
 113
 114 func (t *outboundOffer) setAnswer(answer webrtc.SessionDescription, onOpen func(datachannel.ReadWriteCloser)) error {
 115         setDataChannelOnOpen(t.dataChannel, t.peerConnection, onOpen)
 116         err := t.peerConnection.SetRemoteDescription(answer)
 117         return err
 118 }
 119
 120 type datachannelReadWriter interface {
 121         datachannel.Reader
 122         datachannel.Writer
 123         io.Reader
 124         io.Writer
 125 }
 126
 127 type ioCloserFunc func() error
 128
 129 func (me ioCloserFunc) Close() error {
 130         return me()
 131 }
 132
 133 func setDataChannelOnOpen(
 134         dc *webrtc.DataChannel,
 135         pc wrappedPeerConnection,
 136         onOpen func(closer datachannel.ReadWriteCloser),
 137 ) {
 138         dc.OnOpen(func() {
 139                 raw, err := dc.Detach()
 140                 if err != nil {
 141                         // This shouldn't happen if the API is configured correctly, and we call from OnOpen.
 142                         panic(err)
 143                 }
 144                 onOpen(hookDataChannelCloser(raw, pc))
 145         })
 146 }
 147
 148 func hookDataChannelCloser(dcrwc datachannel.ReadWriteCloser, pc wrappedPeerConnection) datachannel.ReadWriteCloser {
 149         return struct {
 150                 datachannelReadWriter
 151                 io.Closer
 152         }{
 153                 dcrwc,
 154                 ioCloserFunc(pc.Close),
 155         }
 156 }