7 "github.com/anacrolix/log"
8 "github.com/anacrolix/missinggo/v2/pproffd"
9 "github.com/pion/datachannel"
10 "github.com/pion/webrtc/v3"
11 "go.opentelemetry.io/otel"
12 "go.opentelemetry.io/otel/attribute"
13 "go.opentelemetry.io/otel/codes"
14 "go.opentelemetry.io/otel/trace"
21 dataChannelLabel = "webrtc-datachannel"
25 metrics = expvar.NewMap("webtorrent")
26 api = func() *webrtc.API {
27 // Enable the detach API (since it's non-standard but more idiomatic).
28 s.DetachDataChannels()
29 return webrtc.NewAPI(webrtc.WithSettingEngine(s))
31 config = webrtc.Configuration{ICEServers: []webrtc.ICEServer{{URLs: []string{"stun:stun.l.google.com:19302"}}}}
32 newPeerConnectionMu sync.Mutex
35 type wrappedPeerConnection struct {
36 *webrtc.PeerConnection
43 func (me *wrappedPeerConnection) Close() error {
45 defer me.closeMu.Unlock()
46 err := me.CloseWrapper.Close()
51 func newPeerConnection(logger log.Logger) (*wrappedPeerConnection, error) {
52 newPeerConnectionMu.Lock()
53 defer newPeerConnectionMu.Unlock()
54 ctx, span := otel.Tracer(tracerName).Start(context.Background(), "PeerConnection")
55 pc, err := api.NewPeerConnection(config)
57 span.SetStatus(codes.Error, err.Error())
62 wpc := &wrappedPeerConnection{
64 CloseWrapper: pproffd.NewCloseWrapper(pc),
68 // If the state change handler intends to call Close, it should call it on the wrapper.
69 wpc.OnConnectionStateChange(func(state webrtc.PeerConnectionState) {
70 logger.Levelf(log.Warning, "webrtc PeerConnection state changed to %v", state)
71 span.AddEvent("connection state changed", trace.WithAttributes(attribute.String("state", state.String())))
76 func setAndGatherLocalDescription(peerConnection *wrappedPeerConnection, sdp webrtc.SessionDescription) (_ webrtc.SessionDescription, err error) {
77 gatherComplete := webrtc.GatheringCompletePromise(peerConnection.PeerConnection)
78 peerConnection.span.AddEvent("setting local description")
79 err = peerConnection.SetLocalDescription(sdp)
81 err = fmt.Errorf("setting local description: %w", err)
85 peerConnection.span.AddEvent("gathering complete")
86 return *peerConnection.LocalDescription(), nil
89 // newOffer creates a transport and returns a WebRTC offer to be announced. See
90 // https://github.com/pion/webrtc/blob/master/examples/data-channels/jsfiddle/main.go for what this is modelled on.
91 func (tc *TrackerClient) newOffer(
96 peerConnection *wrappedPeerConnection,
97 dataChannel *webrtc.DataChannel,
98 offer webrtc.SessionDescription,
101 peerConnection, err = newPeerConnection(logger)
106 peerConnection.span.SetAttributes(attribute.String(webrtcConnTypeKey, "offer"))
108 dataChannel, err = peerConnection.CreateDataChannel(dataChannelLabel, nil)
110 err = fmt.Errorf("creating data channel: %w", err)
111 peerConnection.Close()
113 initDataChannel(dataChannel, peerConnection, func(dc datachannel.ReadWriteCloser, dcCtx context.Context, dcSpan trace.Span) {
114 metrics.Add("outbound offers answered with datachannel", 1)
116 tc.stats.ConvertedOutboundConns++
118 tc.OnConn(dc, DataChannelContext{
122 peerConnection: peerConnection,
128 offer, err = peerConnection.CreateOffer(nil)
130 peerConnection.Close()
134 offer, err = setAndGatherLocalDescription(peerConnection, offer)
136 peerConnection.Close()
141 type onDetachedDataChannelFunc func(detached datachannel.ReadWriteCloser, ctx context.Context, span trace.Span)
143 func (tc *TrackerClient) initAnsweringPeerConnection(
144 peerConn *wrappedPeerConnection,
145 offerContext offerContext,
146 ) (answer webrtc.SessionDescription, err error) {
147 peerConn.span.SetAttributes(attribute.String(webrtcConnTypeKey, "answer"))
149 timer := time.AfterFunc(30*time.Second, func() {
150 peerConn.span.SetStatus(codes.Error, "answer timeout")
151 metrics.Add("answering peer connections timed out", 1)
154 peerConn.OnDataChannel(func(d *webrtc.DataChannel) {
155 initDataChannel(d, peerConn, func(detached datachannel.ReadWriteCloser, ctx context.Context, span trace.Span) {
157 metrics.Add("answering peer connection conversions", 1)
159 tc.stats.ConvertedInboundConns++
161 tc.OnConn(detached, DataChannelContext{
162 OfferId: offerContext.Id,
164 InfoHash: offerContext.InfoHash,
165 peerConnection: peerConn,
172 err = peerConn.SetRemoteDescription(offerContext.SessDesc)
176 answer, err = peerConn.CreateAnswer(nil)
181 answer, err = setAndGatherLocalDescription(peerConn, answer)
185 // newAnsweringPeerConnection creates a transport from a WebRTC offer and returns a WebRTC answer to be announced.
186 func (tc *TrackerClient) newAnsweringPeerConnection(
187 offerContext offerContext,
189 peerConn *wrappedPeerConnection, answer webrtc.SessionDescription, err error,
191 peerConn, err = newPeerConnection(tc.Logger)
193 err = fmt.Errorf("failed to create new connection: %w", err)
196 answer, err = tc.initAnsweringPeerConnection(peerConn, offerContext)
198 peerConn.span.RecordError(err)
204 type datachannelReadWriter interface {
211 type ioCloserFunc func() error
213 func (me ioCloserFunc) Close() error {
217 func initDataChannel(
218 dc *webrtc.DataChannel,
219 pc *wrappedPeerConnection,
220 onOpen onDetachedDataChannelFunc,
227 pc.span.AddEvent("data channel opened")
228 var ctx context.Context
229 ctx, span = otel.Tracer(tracerName).Start(pc.ctx, "DataChannel")
230 raw, err := dc.Detach()
232 // This shouldn't happen if the API is configured correctly, and we call from OnOpen.
235 onOpen(hookDataChannelCloser(raw, pc, span, dc), ctx, span)
239 // Hooks the datachannel's Close to Close the owning PeerConnection. The datachannel takes ownership
240 // and responsibility for the PeerConnection.
241 func hookDataChannelCloser(
242 dcrwc datachannel.ReadWriteCloser,
243 pc *wrappedPeerConnection,
244 dataChannelSpan trace.Span,
245 originalDataChannel *webrtc.DataChannel,
246 ) datachannel.ReadWriteCloser {
248 datachannelReadWriter
252 ioCloserFunc(func() error {
255 originalDataChannel.Close()
256 dataChannelSpan.End()