webtorrent/transport.go

   1 package webtorrent
   2
   3 import (
   4         "fmt"
   5         "log"
   6         "sync"
   7
   8         "github.com/pion/datachannel"
   9
  10         "github.com/pion/webrtc/v2"
  11 )
  12
  13 var (
  14         api = func() *webrtc.API {
  15                 // Enable the detach API (since it's non-standard but more idiomatic).
  16                 s := webrtc.SettingEngine{}
  17                 s.DetachDataChannels()
  18                 return webrtc.NewAPI(webrtc.WithSettingEngine(s))
  19         }()
  20         config              = webrtc.Configuration{ICEServers: []webrtc.ICEServer{{URLs: []string{"stun:stun.l.google.com:19302"}}}}
  21         newPeerConnectionMu sync.Mutex
  22 )
  23
  24 func newPeerConnection() (*webrtc.PeerConnection, error) {
  25         newPeerConnectionMu.Lock()
  26         defer newPeerConnectionMu.Unlock()
  27         return api.NewPeerConnection(config)
  28 }
  29
  30 type transport struct {
  31         pc *webrtc.PeerConnection
  32         dc *webrtc.DataChannel
  33
  34         lock sync.Mutex
  35 }
  36
  37 // newTransport creates a transport and returns a WebRTC offer to be announced
  38 func newTransport() (*transport, webrtc.SessionDescription, error) {
  39         peerConnection, err := newPeerConnection()
  40         if err != nil {
  41                 return nil, webrtc.SessionDescription{}, fmt.Errorf("failed to peer connection: %v\n", err)
  42         }
  43         dataChannel, err := peerConnection.CreateDataChannel("webrtc-datachannel", nil)
  44         if err != nil {
  45                 return nil, webrtc.SessionDescription{}, fmt.Errorf("failed to data channel: %v\n", err)
  46         }
  47         peerConnection.OnICEConnectionStateChange(func(connectionState webrtc.ICEConnectionState) {
  48                 fmt.Printf("ICE Connection State has changed: %s\n", connectionState.String())
  49         })
  50
  51         dataChannel.OnMessage(func(msg webrtc.DataChannelMessage) {
  52                 fmt.Printf("Message from DataChannel '%s': '%s'\n", dataChannel.Label(), string(msg.Data))
  53         })
  54         offer, err := peerConnection.CreateOffer(nil)
  55         if err != nil {
  56                 return nil, webrtc.SessionDescription{}, fmt.Errorf("failed to create offer: %v\n", err)
  57         }
  58         err = peerConnection.SetLocalDescription(offer)
  59         if err != nil {
  60                 return nil, webrtc.SessionDescription{}, fmt.Errorf("failed to set local description: %v\n", err)
  61         }
  62
  63         t := &transport{pc: peerConnection, dc: dataChannel}
  64         return t, offer, nil
  65 }
  66
  67 // newTransportFromOffer creates a transport from a WebRTC offer and and returns a WebRTC answer to
  68 // be announced.
  69 func newTransportFromOffer(offer webrtc.SessionDescription, onOpen onDataChannelOpen, offerId string) (*transport, webrtc.SessionDescription, error) {
  70         peerConnection, err := newPeerConnection()
  71         if err != nil {
  72                 return nil, webrtc.SessionDescription{}, fmt.Errorf("failed to peer connection: %v", err)
  73         }
  74         peerConnection.OnICEConnectionStateChange(func(connectionState webrtc.ICEConnectionState) {
  75                 fmt.Printf("ICE Connection State has changed: %s\n", connectionState.String())
  76         })
  77
  78         t := &transport{pc: peerConnection}
  79
  80         err = peerConnection.SetRemoteDescription(offer)
  81         if err != nil {
  82                 return nil, webrtc.SessionDescription{}, fmt.Errorf("%v", err)
  83         }
  84         answer, err := peerConnection.CreateAnswer(nil)
  85         if err != nil {
  86                 return nil, webrtc.SessionDescription{}, fmt.Errorf("%v", err)
  87         }
  88         peerConnection.OnDataChannel(func(d *webrtc.DataChannel) {
  89                 fmt.Printf("New DataChannel %s %d\n", d.Label(), d.ID())
  90                 t.lock.Lock()
  91                 t.dc = d
  92                 t.lock.Unlock()
  93                 t.handleOpen(func(dc datachannel.ReadWriteCloser) {
  94                         onOpen(dc, DataChannelContext{answer, offer, offerId, false})
  95                 })
  96         })
  97         err = peerConnection.SetLocalDescription(answer)
  98         if err != nil {
  99                 return nil, webrtc.SessionDescription{}, fmt.Errorf("%v", err)
 100         }
 101
 102         return t, answer, nil
 103 }
 104
 105 // SetAnswer sets the WebRTC answer
 106 func (t *transport) SetAnswer(answer webrtc.SessionDescription, onOpen func(datachannel.ReadWriteCloser)) error {
 107         t.handleOpen(onOpen)
 108
 109         err := t.pc.SetRemoteDescription(answer)
 110         if err != nil {
 111                 return err
 112         }
 113         return nil
 114 }
 115
 116 func (t *transport) handleOpen(onOpen func(datachannel.ReadWriteCloser)) {
 117         t.lock.Lock()
 118         dc := t.dc
 119         t.lock.Unlock()
 120         dc.OnOpen(func() {
 121                 fmt.Printf("Data channel '%s'-'%d' open.\n", dc.Label(), dc.ID())
 122
 123                 // Detach the data channel
 124                 raw, err := dc.Detach()
 125                 if err != nil {
 126                         log.Fatalf("failed to detach: %v", err) // TODO: Error handling
 127                 }
 128
 129                 onOpen(raw)
 130         })
 131 }