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