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