Skip to content

Commit 72e81dc

Browse files
achingbrainachingbrain
authored
feat: add node.js/electron support for webrtc transport (#1905)
1 parent fdd8082 commit 72e81dc

17 files changed

+834
-230
lines changed

‎packages/transport-webrtc/.aegir.js

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2,9 +2,6 @@
22
/** @type {import('aegir').PartialOptions} */
33
export default {
44
build: {
5-
config: {
6-
platform: 'node'
7-
},
85
bundlesizeMax: '117KB'
96
},
107
test: {

‎packages/transport-webrtc/package.json

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,8 @@
3535
"scripts": {
3636
"generate": "protons src/private-to-private/pb/message.proto src/pb/message.proto",
3737
"build": "aegir build",
38-
"test": "aegir test -t browser",
38+
"test": "aegir test -t node -t browser -t electron-main",
39+
"test:node": "aegir test -t node --cov",
3940
"test:chrome": "aegir test -t browser --cov",
4041
"test:firefox": "aegir test -t browser -- --browser firefox",
4142
"lint": "aegir lint",
@@ -61,6 +62,7 @@
6162
"it-to-buffer": "^4.0.2",
6263
"multiformats": "^12.0.1",
6364
"multihashes": "^4.0.3",
65+
"node-datachannel": "^0.4.3",
6466
"p-defer": "^4.0.0",
6567
"p-event": "^6.0.0",
6668
"protons-runtime": "^5.0.0",
@@ -82,5 +84,8 @@
8284
"protons": "^7.0.2",
8385
"sinon": "^15.1.2",
8486
"sinon-ts": "^1.0.0"
87+
},
88+
"browser": {
89+
"./dist/src/webrtc/index.js": "./dist/src/webrtc/index.browser.js"
8590
}
8691
}

‎packages/transport-webrtc/src/private-to-private/handler.ts

Lines changed: 120 additions & 102 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,10 @@
1+
import { CodeError } from '@libp2p/interface/errors'
12
import { logger } from '@libp2p/logger'
23
import { abortableDuplex } from 'abortable-iterator'
34
import { pbStream } from 'it-protobuf-stream'
45
import pDefer, { type DeferredPromise } from 'p-defer'
56
import { DataChannelMuxerFactory } from '../muxer.js'
7+
import { RTCPeerConnection, RTCSessionDescription } from '../webrtc/index.js'
68
import { Message } from './pb/message.js'
79
import { readCandidatesUntilConnected, resolveOnConnected } from './util.js'
810
import type { DataChannelOpts } from '../stream.js'
@@ -20,66 +22,75 @@ export async function handleIncomingStream ({ rtcConfiguration, dataChannelOptio
2022
const signal = AbortSignal.timeout(DEFAULT_TIMEOUT)
2123
const stream = pbStream(abortableDuplex(rawStream, signal)).pb(Message)
2224
const pc = new RTCPeerConnection(rtcConfiguration)
23-
const muxerFactory = new DataChannelMuxerFactory({ peerConnection: pc, dataChannelOptions })
24-
const connectedPromise: DeferredPromise<void> = pDefer()
25-
const answerSentPromise: DeferredPromise<void> = pDefer()
26-
27-
signal.onabort = () => { connectedPromise.reject() }
28-
// candidate callbacks
29-
pc.onicecandidate = ({ candidate }) => {
30-
answerSentPromise.promise.then(
31-
async () => {
32-
await stream.write({
33-
type: Message.Type.ICE_CANDIDATE,
34-
data: (candidate != null) ? JSON.stringify(candidate.toJSON()) : ''
35-
})
36-
},
37-
(err) => {
38-
log.error('cannot set candidate since sending answer failed', err)
39-
}
40-
)
41-
}
4225

43-
resolveOnConnected(pc, connectedPromise)
26+
try {
27+
const muxerFactory = new DataChannelMuxerFactory({ peerConnection: pc, dataChannelOptions })
28+
const connectedPromise: DeferredPromise<void> = pDefer()
29+
const answerSentPromise: DeferredPromise<void> = pDefer()
30+
31+
signal.onabort = () => {
32+
connectedPromise.reject(new CodeError('Timed out while trying to connect', 'ERR_TIMEOUT'))
33+
}
34+
// candidate callbacks
35+
pc.onicecandidate = ({ candidate }) => {
36+
answerSentPromise.promise.then(
37+
async () => {
38+
await stream.write({
39+
type: Message.Type.ICE_CANDIDATE,
40+
data: (candidate != null) ? JSON.stringify(candidate.toJSON()) : ''
41+
})
42+
},
43+
(err) => {
44+
log.error('cannot set candidate since sending answer failed', err)
45+
connectedPromise.reject(err)
46+
}
47+
)
48+
}
49+
50+
resolveOnConnected(pc, connectedPromise)
51+
52+
// read an SDP offer
53+
const pbOffer = await stream.read()
54+
if (pbOffer.type !== Message.Type.SDP_OFFER) {
55+
throw new Error(`expected message type SDP_OFFER, received: ${pbOffer.type ?? 'undefined'} `)
56+
}
57+
const offer = new RTCSessionDescription({
58+
type: 'offer',
59+
sdp: pbOffer.data
60+
})
61+
62+
await pc.setRemoteDescription(offer).catch(err => {
63+
log.error('could not execute setRemoteDescription', err)
64+
throw new Error('Failed to set remoteDescription')
65+
})
66+
67+
// create and write an SDP answer
68+
const answer = await pc.createAnswer().catch(err => {
69+
log.error('could not execute createAnswer', err)
70+
answerSentPromise.reject(err)
71+
throw new Error('Failed to create answer')
72+
})
73+
// write the answer to the remote
74+
await stream.write({ type: Message.Type.SDP_ANSWER, data: answer.sdp })
75+
76+
await pc.setLocalDescription(answer).catch(err => {
77+
log.error('could not execute setLocalDescription', err)
78+
answerSentPromise.reject(err)
79+
throw new Error('Failed to set localDescription')
80+
})
81+
82+
answerSentPromise.resolve()
83+
84+
// wait until candidates are connected
85+
await readCandidatesUntilConnected(connectedPromise, pc, stream)
4486

45-
// read an SDP offer
46-
const pbOffer = await stream.read()
47-
if (pbOffer.type !== Message.Type.SDP_OFFER) {
48-
throw new Error(`expected message type SDP_OFFER, received: ${pbOffer.type ?? 'undefined'} `)
87+
const remoteAddress = parseRemoteAddress(pc.currentRemoteDescription?.sdp ?? '')
88+
89+
return { pc, muxerFactory, remoteAddress }
90+
} catch (err) {
91+
pc.close()
92+
throw err
4993
}
50-
const offer = new RTCSessionDescription({
51-
type: 'offer',
52-
sdp: pbOffer.data
53-
})
54-
55-
await pc.setRemoteDescription(offer).catch(err => {
56-
log.error('could not execute setRemoteDescription', err)
57-
throw new Error('Failed to set remoteDescription')
58-
})
59-
60-
// create and write an SDP answer
61-
const answer = await pc.createAnswer().catch(err => {
62-
log.error('could not execute createAnswer', err)
63-
answerSentPromise.reject(err)
64-
throw new Error('Failed to create answer')
65-
})
66-
// write the answer to the remote
67-
await stream.write({ type: Message.Type.SDP_ANSWER, data: answer.sdp })
68-
69-
await pc.setLocalDescription(answer).catch(err => {
70-
log.error('could not execute setLocalDescription', err)
71-
answerSentPromise.reject(err)
72-
throw new Error('Failed to set localDescription')
73-
})
74-
75-
answerSentPromise.resolve()
76-
77-
// wait until candidates are connected
78-
await readCandidatesUntilConnected(connectedPromise, pc, stream)
79-
80-
const remoteAddress = parseRemoteAddress(pc.currentRemoteDescription?.sdp ?? '')
81-
82-
return { pc, muxerFactory, remoteAddress }
8394
}
8495

8596
export interface ConnectOptions {
@@ -93,56 +104,63 @@ export async function initiateConnection ({ rtcConfiguration, dataChannelOptions
93104
const stream = pbStream(abortableDuplex(rawStream, signal)).pb(Message)
94105
// setup peer connection
95106
const pc = new RTCPeerConnection(rtcConfiguration)
96-
const muxerFactory = new DataChannelMuxerFactory({ peerConnection: pc, dataChannelOptions })
97-
98-
const connectedPromise: DeferredPromise<void> = pDefer()
99-
resolveOnConnected(pc, connectedPromise)
100-
101-
// reject the connectedPromise if the signal aborts
102-
signal.onabort = connectedPromise.reject
103-
// we create the channel so that the peerconnection has a component for which
104-
// to collect candidates. The label is not relevant to connection initiation
105-
// but can be useful for debugging
106-
const channel = pc.createDataChannel('init')
107-
// setup callback to write ICE candidates to the remote
108-
// peer
109-
pc.onicecandidate = ({ candidate }) => {
110-
void stream.write({
111-
type: Message.Type.ICE_CANDIDATE,
112-
data: (candidate != null) ? JSON.stringify(candidate.toJSON()) : ''
113-
})
114-
.catch(err => {
115-
log.error('error sending ICE candidate', err)
107+
108+
try {
109+
const muxerFactory = new DataChannelMuxerFactory({ peerConnection: pc, dataChannelOptions })
110+
111+
const connectedPromise: DeferredPromise<void> = pDefer()
112+
resolveOnConnected(pc, connectedPromise)
113+
114+
// reject the connectedPromise if the signal aborts
115+
signal.onabort = connectedPromise.reject
116+
// we create the channel so that the peerconnection has a component for which
117+
// to collect candidates. The label is not relevant to connection initiation
118+
// but can be useful for debugging
119+
const channel = pc.createDataChannel('init')
120+
// setup callback to write ICE candidates to the remote
121+
// peer
122+
pc.onicecandidate = ({ candidate }) => {
123+
void stream.write({
124+
type: Message.Type.ICE_CANDIDATE,
125+
data: (candidate != null) ? JSON.stringify(candidate.toJSON()) : ''
116126
})
117-
}
118-
// create an offer
119-
const offerSdp = await pc.createOffer()
120-
// write the offer to the stream
121-
await stream.write({ type: Message.Type.SDP_OFFER, data: offerSdp.sdp })
122-
// set offer as local description
123-
await pc.setLocalDescription(offerSdp).catch(err => {
124-
log.error('could not execute setLocalDescription', err)
125-
throw new Error('Failed to set localDescription')
126-
})
127-
128-
// read answer
129-
const answerMessage = await stream.read()
130-
if (answerMessage.type !== Message.Type.SDP_ANSWER) {
131-
throw new Error('remote should send an SDP answer')
132-
}
127+
.catch(err => {
128+
log.error('error sending ICE candidate', err)
129+
})
130+
}
131+
132+
// create an offer
133+
const offerSdp = await pc.createOffer()
134+
// write the offer to the stream
135+
await stream.write({ type: Message.Type.SDP_OFFER, data: offerSdp.sdp })
136+
// set offer as local description
137+
await pc.setLocalDescription(offerSdp).catch(err => {
138+
log.error('could not execute setLocalDescription', err)
139+
throw new Error('Failed to set localDescription')
140+
})
133141

134-
const answerSdp = new RTCSessionDescription({ type: 'answer', sdp: answerMessage.data })
135-
await pc.setRemoteDescription(answerSdp).catch(err => {
136-
log.error('could not execute setRemoteDescription', err)
137-
throw new Error('Failed to set remoteDescription')
138-
})
142+
// read answer
143+
const answerMessage = await stream.read()
144+
if (answerMessage.type !== Message.Type.SDP_ANSWER) {
145+
throw new Error('remote should send an SDP answer')
146+
}
139147

140-
await readCandidatesUntilConnected(connectedPromise, pc, stream)
141-
channel.close()
148+
const answerSdp = new RTCSessionDescription({ type: 'answer', sdp: answerMessage.data })
149+
await pc.setRemoteDescription(answerSdp).catch(err => {
150+
log.error('could not execute setRemoteDescription', err)
151+
throw new Error('Failed to set remoteDescription')
152+
})
142153

143-
const remoteAddress = parseRemoteAddress(pc.currentRemoteDescription?.sdp ?? '')
154+
await readCandidatesUntilConnected(connectedPromise, pc, stream)
155+
channel.close()
144156

145-
return { pc, muxerFactory, remoteAddress }
157+
const remoteAddress = parseRemoteAddress(pc.currentRemoteDescription?.sdp ?? '')
158+
159+
return { pc, muxerFactory, remoteAddress }
160+
} catch (err) {
161+
pc.close()
162+
throw err
163+
}
146164
}
147165

148166
function parseRemoteAddress (sdp: string): string {

‎packages/transport-webrtc/src/private-to-private/transport.ts

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import { peerIdFromString } from '@libp2p/peer-id'
55
import { multiaddr, type Multiaddr, protocols } from '@multiformats/multiaddr'
66
import { codes } from '../error.js'
77
import { WebRTCMultiaddrConnection } from '../maconn.js'
8+
import { cleanup } from '../webrtc/index.js'
89
import { initiateConnection, handleIncomingStream } from './handler.js'
910
import { WebRTCPeerListener } from './listener.js'
1011
import type { DataChannelOpts } from '../stream.js'
@@ -57,6 +58,7 @@ export class WebRTCTransport implements Transport, Startable {
5758

5859
async stop (): Promise<void> {
5960
await this.components.registrar.unhandle(SIGNALING_PROTO_ID)
61+
cleanup()
6062
this._started = false
6163
}
6264

‎packages/transport-webrtc/src/private-to-private/util.ts

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
import { logger } from '@libp2p/logger'
22
import { isFirefox } from '../util.js'
3+
import { RTCIceCandidate } from '../webrtc/index.js'
34
import { Message } from './pb/message.js'
45
import type { DeferredPromise } from 'p-defer'
56

0 commit comments

Comments
 (0)