diff --git a/common/src/rtsp-server.ts b/common/src/rtsp-server.ts index a36ec3aa2..e6ec4924c 100644 --- a/common/src/rtsp-server.ts +++ b/common/src/rtsp-server.ts @@ -2,6 +2,7 @@ import { readLength, readLine } from './read-stream'; import { Duplex } from 'stream'; import { randomBytes } from 'crypto'; import { StreamChunk, StreamParser } from './stream-parser'; +import dgram from 'dgram'; interface Headers { [header: string]: string @@ -67,17 +68,23 @@ export class RtspServer { videoChannel = 0; audioChannel = 2; session: string; + console: Console; + udpPorts = { + video: 0, + audio: 0, + }; - constructor(public duplex: Duplex, public sdp?: string) { + constructor(public client: Duplex, public sdp?: string, public udp?: dgram.Socket) { this.session = randomBytes(4).toString('hex'); } async handleSetup() { let currentHeaders: string[] = []; while (true) { - let line = await readLine(this.duplex); + let line = await readLine(this.client); line = line.trim(); if (!line) { + this.console?.log(currentHeaders.join('\n')) if (!await this.headers(currentHeaders)) break; currentHeaders = []; @@ -85,7 +92,6 @@ export class RtspServer { } currentHeaders.push(line); } - } async handlePlayback() { @@ -99,9 +105,9 @@ export class RtspServer { packet: Buffer, }> { while (true) { - const header = await readLength(this.duplex, 4); + const header = await readLength(this.client, 4); const length = header.readUInt16BE(2); - const packet = await readLength(this.duplex, length); + const packet = await readLength(this.client, length); const id = header.readUInt8(1); yield { type: id - (id % 2) === this.videoChannel ? 'video' : 'audio', @@ -118,16 +124,31 @@ export class RtspServer { header.writeUInt8(channel, 1); header.writeUInt16BE(rtp.length, 2); - this.duplex.write(header); - this.duplex.write(Buffer.from(rtp)); + this.client.write(header); + this.client.write(Buffer.from(rtp)); + } + + sendUdp(port: number, packet: Buffer, rtcp: boolean) { + // todo: support non local host? + this.udp.send(packet, rtcp ? port + 1 : port, '127.0.0.1'); } sendVideo(packet: Buffer, rtcp: boolean) { - this.send(packet, rtcp ? this.videoChannel + 1 : this.videoChannel); + if (this.udp) { + this.sendUdp(this.udpPorts.video, packet, rtcp) + } + else { + this.send(packet, rtcp ? this.videoChannel + 1 : this.videoChannel); + } } sendAudio(packet: Buffer, rtcp: boolean) { - this.send(packet, rtcp ? this.audioChannel + 1 : this.audioChannel); + if (this.udp) { + this.sendUdp(this.udpPorts.audio, packet, rtcp) + } + else { + this.send(packet, rtcp ? this.audioChannel + 1 : this.audioChannel); + } } options(url: string, requestHeaders: Headers) { @@ -146,8 +167,17 @@ export class RtspServer { setup(url: string, requestHeaders: Headers) { const headers: Headers = {}; + const transport = requestHeaders['transport']; headers['Transport'] = requestHeaders['transport']; headers['Session'] = this.session; + if (transport.includes('UDP')) { + const match = transport.match(/.*?client_port=([0-9]+)-([0-9]+)/) + const [_, rtp, rtcp] = match; + if (url.includes('audio')) + this.udpPorts.audio = parseInt(rtp); + else + this.udpPorts.video = parseInt(rtp); + } this.respond(200, 'OK', requestHeaders, headers) } @@ -161,7 +191,7 @@ export class RtspServer { async announce(url: string, requestHeaders: Headers) { const contentLength = parseInt(requestHeaders['content-length']); - const sdpBuffer = await readLength(this.duplex, contentLength); + const sdpBuffer = await readLength(this.client, contentLength); this.sdp = sdpBuffer.toString(); const headers: Headers = {}; headers['Session'] = this.session; @@ -198,8 +228,8 @@ export class RtspServer { response += `${key}: ${value}\r\n`; } response += '\r\n'; - this.duplex.write(response); + this.client.write(response); if (buffer) - this.duplex.write(buffer); + this.client.write(buffer); } } diff --git a/common/src/wrtc-ffmpeg-source.ts b/common/src/wrtc-ffmpeg-source.ts index be9124d70..856cc89d4 100644 --- a/common/src/wrtc-ffmpeg-source.ts +++ b/common/src/wrtc-ffmpeg-source.ts @@ -12,11 +12,13 @@ s=- c=IN IP4 127.0.0.1 t=0 0 m=audio ${audioPort} UDP 96 +a=control:trackID=audio a=rtpmap:96 opus/48000/2 a=fmtp:96 minptime=10;useinbandfec=1 a=rtcp-fb:96 transport-cc a=sendrecv m=video ${videoPort} UDP 97 +a=control:trackID=video a=rtpmap:97 H264/90000 a=rtcp-fb:97 ccm fir a=rtcp-fb:97 nack @@ -48,22 +50,29 @@ export async function createRTCPeerConnectionSource(avsource: RTCAVSignalingOffe const { clientPromise, port } = await listenZeroSingleClient(); - let ai: NodeJS.Timeout; let vi: NodeJS.Timeout; let pc: RTCPeerConnection; let socket: Socket; + // rtsp server must operate in udp forwarding mode to accomodate packet reordering. + let udp = dgram.createSocket('udp4'); const cleanup = () => { pc?.close(); socket?.destroy(); clearInterval(ai); clearInterval(vi); + try { + udp.close(); + } + catch(e) { + } }; clientPromise.then(async (client) => { socket = client; - const rtspServer = new RtspServer(socket, createSdpInput(audioPort, videoPort)); + const rtspServer = new RtspServer(socket, createSdpInput(audioPort, videoPort), udp); + rtspServer.console = console; rtspServer.audioChannel = 0; rtspServer.videoChannel = 2; await rtspServer.handleSetup(); @@ -179,7 +188,7 @@ export async function createRTCPeerConnectionSource(avsource: RTCAVSignalingOffe url, mediaStreamOptions: getRTCMediaStreamOptions(id, name, 'rtsp'), inputArguments: [ - "-rtsp_transport", "tcp", + "-rtsp_transport", "udp", "-max_delay", "1000000", '-i', url, ] diff --git a/plugins/google-device-access/package-lock.json b/plugins/google-device-access/package-lock.json index fd3148fcf..9d0f78af4 100644 --- a/plugins/google-device-access/package-lock.json +++ b/plugins/google-device-access/package-lock.json @@ -1,12 +1,12 @@ { "name": "@scrypted/google-device-access", - "version": "0.0.82", + "version": "0.0.83", "lockfileVersion": 2, "requires": true, "packages": { "": { "name": "@scrypted/google-device-access", - "version": "0.0.82", + "version": "0.0.83", "dependencies": { "@googleapis/smartdevicemanagement": "^0.2.0", "axios": "^0.21.1", diff --git a/plugins/google-device-access/package.json b/plugins/google-device-access/package.json index cf8a81ab9..fa6c0a87d 100644 --- a/plugins/google-device-access/package.json +++ b/plugins/google-device-access/package.json @@ -45,5 +45,5 @@ "@types/node": "^14.17.11", "@types/url-parse": "^1.4.3" }, - "version": "0.0.82" + "version": "0.0.83" } diff --git a/plugins/google-device-access/src/main.ts b/plugins/google-device-access/src/main.ts index 7ba31f41f..14d0eb81d 100644 --- a/plugins/google-device-access/src/main.ts +++ b/plugins/google-device-access/src/main.ts @@ -8,6 +8,7 @@ import throttle from 'lodash/throttle'; import { createRTCPeerConnectionSource, getRTCMediaStreamOptions as getRtcMediaStreamOptions } from '../../../common/src/wrtc-ffmpeg-source'; import { sleep } from '../../../common/src/sleep'; import fs from 'fs'; +import { randomBytes } from 'crypto'; const { deviceManager, mediaManager, endpointManager, systemManager } = sdk; @@ -100,6 +101,7 @@ class NestCamera extends ScryptedDeviceBase implements Readme, Camera, VideoCame signalingMime: string; lastMotionEventId: string; lastImage: Promise; + streams = new Map(); constructor(public provider: GoogleSmartDeviceAccess, public device: any) { super(device.name.split('/').pop()); @@ -113,6 +115,10 @@ class NestCamera extends ScryptedDeviceBase implements Readme, Camera, VideoCame this.toMimeType = this.signalingMime; } + trackStream(id: string, result: any) { + this.streams.set(id, result); + } + async getReadmeMarkdown(): Promise { return this.isWebRtc ? readmeV2 : readmeV1; } @@ -205,25 +211,18 @@ class NestCamera extends ScryptedDeviceBase implements Readme, Camera, VideoCame return { result, answer }; } - addRefreshOptions(result: any, mso: MediaStreamOptions): MediaStreamOptions { - const { expiresAt, streamToken, streamExtensionToken, mediaSessionId } = result.data.results; - const expirationDate = new Date(expiresAt); - const refreshAt = expirationDate.getTime(); + addRefreshOptions(trackerId: string, mso: MediaStreamOptions): MediaStreamOptions { return Object.assign(mso, { - refreshAt, + refreshAt: Date.now() + 4 * 60 * 1000, metadata: { - expiresAt, - streamToken, - streamExtensionToken, - mediaSessionId, + trackerId, }, }); } - createFFmpegMediaObject(result: any) { + createFFmpegMediaObject(trackerId: string, result: any) { const u = result.data.results.streamUrls.rtspUrl; - this.console.log('rtsp url', u); - + this.trackStream(trackerId, result); return mediaManager.createFFmpegMediaObject({ url: u, mediaStreamOptions: this.addRefreshOptions(result, getSdmRtspMediaStreamOptions()), @@ -240,8 +239,9 @@ class NestCamera extends ScryptedDeviceBase implements Readme, Camera, VideoCame } async getVideoStream(options?: RequestMediaStreamOptions): Promise { - if (options?.metadata?.streamExtensionToken || options?.metadata?.mediaSessionId) { - const { streamExtensionToken, mediaSessionId } = options?.metadata; + if (options?.metadata?.trackerId) { + const { trackerId } = options?.metadata; + const { streamExtensionToken, mediaSessionId } = this.streams.get(trackerId); const streamFormat = this.isWebRtc ? 'WebRtc' : 'Rtsp'; const result = await this.provider.authPost(`/devices/${this.nativeId}:executeCommand`, { command: `sdm.devices.commands.CameraLiveStream.Extend${streamFormat}Stream`, @@ -251,8 +251,10 @@ class NestCamera extends ScryptedDeviceBase implements Readme, Camera, VideoCame } }); + this.trackStream(trackerId, result); + const mso = this.isWebRtc ? getSdmRtcMediaStreamOptions(this.signalingMime) : getSdmRtspMediaStreamOptions(); - this.addRefreshOptions(result, mso); + this.addRefreshOptions(trackerId, mso); const ffmpegInput: FFMpegInput = { url: undefined, @@ -270,7 +272,9 @@ class NestCamera extends ScryptedDeviceBase implements Readme, Camera, VideoCame command: "sdm.devices.commands.CameraLiveStream.GenerateRtspStream", params: {} }); - return this.createFFmpegMediaObject(result); + const trackerId = randomBytes(8).toString('hex'); + this.trackStream(trackerId, result); + return this.createFFmpegMediaObject(trackerId, result); } } async getVideoStreamOptions(): Promise { @@ -538,14 +542,14 @@ class GoogleSmartDeviceAccess extends ScryptedDeviceBase implements OauthClient, break; } } - let streamResult: any; - const result = await createRTCPeerConnectionSource(createNestOfferSetup(device.signalingMime), 'default', 'MPEG-TS', device.console, async (offer) => { + const trackerId = randomBytes(8).toString('hex'); + const ffmpegInput = await createRTCPeerConnectionSource(createNestOfferSetup(device.signalingMime), 'default', 'MPEG-TS', device.console, async (offer) => { const { result, answer } = await device.sendOffer(offer); - streamResult = result; + device.trackStream(trackerId, result); return answer; }); - device.addRefreshOptions(streamResult, result.ffmpegInput.mediaStreamOptions); - return Buffer.from(JSON.stringify(result.ffmpegInput)); + device.addRefreshOptions(trackerId, ffmpegInput.mediaStreamOptions); + return Buffer.from(JSON.stringify(ffmpegInput)); } async onRequest(request: HttpRequest, response: HttpResponse): Promise { diff --git a/plugins/ring/package-lock.json b/plugins/ring/package-lock.json index 294801585..609ffd1d8 100644 --- a/plugins/ring/package-lock.json +++ b/plugins/ring/package-lock.json @@ -1,12 +1,12 @@ { "name": "@scrypted/ring", - "version": "0.0.37", + "version": "0.0.38", "lockfileVersion": 2, "requires": true, "packages": { "": { "name": "@scrypted/ring", - "version": "0.0.37", + "version": "0.0.38", "dependencies": { "@homebridge/camera-utils": "^2.0.4", "@koush/ring-client-api": "^9.24.1", diff --git a/plugins/ring/package.json b/plugins/ring/package.json index b07f6d8b2..4c3c6acc1 100644 --- a/plugins/ring/package.json +++ b/plugins/ring/package.json @@ -37,5 +37,5 @@ "devDependencies": { "@scrypted/sdk": "file:../../sdk" }, - "version": "0.0.37" + "version": "0.0.38" }