7 "go.opentelemetry.io/otel/attribute"
8 "go.opentelemetry.io/otel/codes"
9 "go.opentelemetry.io/otel/trace"
13 "github.com/anacrolix/log"
14 "github.com/anacrolix/missinggo/v2/pproffd"
15 "github.com/pion/datachannel"
16 "github.com/pion/webrtc/v3"
17 "go.opentelemetry.io/otel"
21 metrics = expvar.NewMap("webtorrent")
22 api = func() *webrtc.API {
23 // Enable the detach API (since it's non-standard but more idiomatic).
24 s.DetachDataChannels()
25 return webrtc.NewAPI(webrtc.WithSettingEngine(s))
27 config = webrtc.Configuration{ICEServers: []webrtc.ICEServer{{URLs: []string{"stun:stun.l.google.com:19302"}}}}
28 newPeerConnectionMu sync.Mutex
31 type wrappedPeerConnection struct {
32 *webrtc.PeerConnection
39 func (me *wrappedPeerConnection) Close() error {
41 defer me.closeMu.Unlock()
42 err := me.CloseWrapper.Close()
47 func newPeerConnection(logger log.Logger) (*wrappedPeerConnection, error) {
48 newPeerConnectionMu.Lock()
49 defer newPeerConnectionMu.Unlock()
50 ctx, span := otel.Tracer(tracerName).Start(context.Background(), "PeerConnection")
51 pc, err := api.NewPeerConnection(config)
53 span.SetStatus(codes.Error, err.Error())
58 wpc := &wrappedPeerConnection{
60 CloseWrapper: pproffd.NewCloseWrapper(pc),
64 // If the state change handler intends to call Close, it should call it on the wrapper.
65 wpc.OnConnectionStateChange(func(state webrtc.PeerConnectionState) {
66 logger.Levelf(log.Warning, "webrtc PeerConnection state changed to %v", state)
67 span.AddEvent("connection state changed", trace.WithAttributes(attribute.String("state", state.String())))
72 func setAndGatherLocalDescription(peerConnection *wrappedPeerConnection, sdp webrtc.SessionDescription) (_ webrtc.SessionDescription, err error) {
73 gatherComplete := webrtc.GatheringCompletePromise(peerConnection.PeerConnection)
74 peerConnection.span.AddEvent("setting local description")
75 err = peerConnection.SetLocalDescription(sdp)
77 err = fmt.Errorf("setting local description: %w", err)
81 peerConnection.span.AddEvent("gathering complete")
82 return *peerConnection.LocalDescription(), nil
85 // newOffer creates a transport and returns a WebRTC offer to be announced
89 peerConnection *wrappedPeerConnection,
90 offer webrtc.SessionDescription,
93 peerConnection, err = newPeerConnection(logger)
98 peerConnection.span.SetAttributes(attribute.String(webrtcConnTypeKey, "offer"))
100 offer, err = peerConnection.CreateOffer(nil)
102 peerConnection.Close()
106 offer, err = setAndGatherLocalDescription(peerConnection, offer)
108 peerConnection.Close()
113 func initAnsweringPeerConnection(
114 peerConnection *wrappedPeerConnection,
115 offer webrtc.SessionDescription,
116 ) (answer webrtc.SessionDescription, err error) {
117 peerConnection.span.SetAttributes(attribute.String(webrtcConnTypeKey, "answer"))
119 err = peerConnection.SetRemoteDescription(offer)
123 answer, err = peerConnection.CreateAnswer(nil)
128 answer, err = setAndGatherLocalDescription(peerConnection, answer)
132 // newAnsweringPeerConnection creates a transport from a WebRTC offer and returns a WebRTC answer to be announced.
133 func newAnsweringPeerConnection(
135 offer webrtc.SessionDescription,
137 peerConn *wrappedPeerConnection, answer webrtc.SessionDescription, err error,
139 peerConn, err = newPeerConnection(logger)
141 err = fmt.Errorf("failed to create new connection: %w", err)
144 answer, err = initAnsweringPeerConnection(peerConn, offer)
146 peerConn.span.RecordError(err)
152 type datachannelReadWriter interface {
159 type ioCloserFunc func() error
161 func (me ioCloserFunc) Close() error {
165 func setDataChannelOnOpen(
167 dc *webrtc.DataChannel,
168 pc *wrappedPeerConnection,
169 onOpen func(closer datachannel.ReadWriteCloser),
172 trace.SpanFromContext(ctx).AddEvent("opened")
173 raw, err := dc.Detach()
175 // This shouldn't happen if the API is configured correctly, and we call from OnOpen.
179 onOpen(hookDataChannelCloser(raw, pc))
183 // Hooks the datachannel's Close to Close the owning PeerConnection. The datachannel takes ownership
184 // and responsibility for the PeerConnection.
185 func hookDataChannelCloser(dcrwc datachannel.ReadWriteCloser, pc *wrappedPeerConnection) datachannel.ReadWriteCloser {
187 datachannelReadWriter
191 ioCloserFunc(pc.Close),