michael/scrypted
1
0
Fork 0
mirror of https://github.com/koush/scrypted.git synced 2026-07-09 08:30:39 +01:00
Code Issues Actions Packages Projects Releases Wiki Activity

gda/ring/common: webrtc to rtsp gateway should operate in udp mode so ffmpeg will reorder packets.

Browse Source
This commit is contained in:
Koushik Dutta
2022-02-04 11:20:45 -08:00
parent c54f8311fa
commit 2e5c67bf63
7 changed files with 85 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
common/src/rtsp-server.ts
View File

@@ -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);
}
}

15
common/src/wrtc-ffmpeg-source.ts
View File

@@ -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,
]

4
plugins/google-device-access/package-lock.json generated
View File

@@ -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",

2
plugins/google-device-access/package.json
View File

@@ -45,5 +45,5 @@
"@types/node": "^14.17.11",
"@types/url-parse": "^1.4.3"
},
"version": "0.0.82"
"version": "0.0.83"
}

46
plugins/google-device-access/src/main.ts
View File

@@ -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<Buffer>;
streams = new Map<string, any>();
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<string> {
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<MediaObject> {
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<MediaStreamOptions[]> {
@@ -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<void> {

4
plugins/ring/package-lock.json generated
View File

@@ -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",

2
plugins/ring/package.json
View File

@@ -37,5 +37,5 @@
"devDependencies": {
"@scrypted/sdk": "file:../../sdk"
},
"version": "0.0.37"
"version": "0.0.38"
}