feat: sip streaming api (#88)

This includes some performance improvements in camera live streams.  The time to initiate a stream has been reduced and multiple streams should be able to function at the same time.

BREAKING CHANGE: `SipSession` api has changed and now exposes `Observable`s for RTP packets on `audioStream` and `videoStream`

Author: dgreif

api/api.ts

@@ -61,13 +61,22 @@ export class RingApi {
         cameraStatusPollingSeconds,
         cameraDingsPollingSeconds
       } = this.options,
-      camerasRequestUpdate$ = merge(
+      onCamerasRequestUpdate = merge(
         ...cameras.map(camera => camera.onRequestUpdate)
-      ).pipe(throttleTime(500)),
+      ),
+      onCamerasRequestActiveDings = merge(
+        ...cameras.map(camera => camera.onRequestActiveDings)
+      ),
       onUpdateReceived = new Subject(),
+      onActiveDingsReceived = new Subject(),
       onPollForStatusUpdate = cameraStatusPollingSeconds
         ? onUpdateReceived.pipe(debounceTime(cameraStatusPollingSeconds * 1000))
         : EMPTY,
+      onPollForActiveDings = cameraDingsPollingSeconds
+        ? onActiveDingsReceived.pipe(
+            debounceTime(cameraDingsPollingSeconds * 1000)
+          )
+        : EMPTY,
       camerasById = cameras.reduce(
         (byId, camera) => {
           byId[camera.id] = camera
@@ -80,7 +89,7 @@ export class RingApi {
       return
     }
 
-    merge(camerasRequestUpdate$, onPollForStatusUpdate)
+    merge(onCamerasRequestUpdate, onPollForStatusUpdate)
       .pipe(
         throttleTime(500),
         switchMap(async () => {
@@ -107,32 +116,26 @@ export class RingApi {
       onUpdateReceived.next() // kick off polling
     }
 
-    if (cameraDingsPollingSeconds) {
-      const onPollForActiveDings = new Subject()
-
-      onPollForActiveDings
-        .pipe(
-          debounceTime(cameraDingsPollingSeconds * 1000),
-          switchMap(() => {
-            return this.fetchActiveDings().catch(() => null)
-          })
-        )
-        .subscribe(activeDings => {
-          onPollForActiveDings.next()
+    merge(onCamerasRequestActiveDings, onPollForActiveDings).subscribe(
+      async () => {
+        const activeDings = await this.fetchActiveDings().catch(() => null)
+        onActiveDingsReceived.next()
 
-          if (!activeDings || !activeDings.length) {
-            return
-          }
+        if (!activeDings || !activeDings.length) {
+          return
+        }
 
-          activeDings.forEach(activeDing => {
-            const camera = camerasById[activeDing.doorbot_id]
-            if (camera) {
-              camera.processActiveDing(activeDing)
-            }
-          })
+        activeDings.forEach(activeDing => {
+          const camera = camerasById[activeDing.doorbot_id]
+          if (camera) {
+            camera.processActiveDing(activeDing)
+          }
         })
+      }
+    )
 
-      onPollForActiveDings.next() // kick off polling
+    if (cameraDingsPollingSeconds) {
+      onActiveDingsReceived.next() // kick off polling
     }
   }
 

api/ring-camera.ts

@@ -18,8 +18,12 @@ import {
   share,
   take
 } from 'rxjs/operators'
+import { createSocket } from 'dgram'
+import { bindToRandomPort, getPublicIp } from './rtp-utils'
 import { delay, logError } from './util'
-import { RtpOptions, SipSession } from './sip-session'
+import { SipSession, SrtpOptions } from './sip-session'
+
+const getPort = require('get-port')
 
 const snapshotRefreshDelay = 500,
   maxSnapshotRefreshSeconds = 30,
@@ -49,6 +53,7 @@ export class RingCamera {
 
   onData = new BehaviorSubject<CameraData>(this.initialData)
   onRequestUpdate = new Subject()
+  onRequestActiveDings = new Subject()
 
   onNewDing = new Subject<ActiveDing>()
   onActiveDings = new BehaviorSubject<ActiveDing[]>([])
@@ -160,6 +165,7 @@ export class RingCamera {
       )
       .toPromise()
     await this.startVideoOnDemand()
+    this.onRequestActiveDings.next()
     return vodPromise
   }
 
@@ -271,18 +277,46 @@ export class RingCamera {
         .find(x => !this.sipUsedDingIds.includes(x.id_str)),
       targetDing = existingDing || (await this.getSipConnectionDetails())
 
+    this.sipUsedDingIds.push(targetDing.id_str)
+
     return {
       to: targetDing.sip_to,
       from: targetDing.sip_from,
       dingId: targetDing.id_str
     }
   }
 
-  async createSipSession(rtpOptions: RtpOptions) {
-    const sipOptions = await this.getSipOptions()
-
-    this.sipUsedDingIds.push(sipOptions.dingId)
+  async createSipSession(
+    srtpOption: { audio?: SrtpOptions; video?: SrtpOptions } = {}
+  ) {
+    const videoSocket = createSocket('udp4'),
+      audioSocket = createSocket('udp4'),
+      [sipOptions, publicIpPromise, videoPort, audioPort] = await Promise.all([
+        this.getSipOptions(),
+        getPublicIp(),
+        bindToRandomPort(videoSocket),
+        bindToRandomPort(audioSocket)
+      ]),
+      rtpOptions = {
+        address: await publicIpPromise,
+        audio: {
+          port: audioPort,
+          ...srtpOption.audio
+        },
+        video: {
+          port: videoPort,
+          ...srtpOption.video
+        }
+      }
 
-    return new SipSession(sipOptions, rtpOptions)
+    return new SipSession(
+      {
+        ...sipOptions,
+        tlsPort: await getPort() // get a random port, this can still cause race conditions.
+      },
+      rtpOptions,
+      videoSocket,
+      audioSocket
+    )
   }
 }

api/rtp-utils.ts

@@ -0,0 +1,58 @@
+import { Socket } from 'dgram'
+import { AddressInfo } from 'net'
+import { v4 as fetchPublicIp } from 'public-ip'
+const stun = require('stun')
+
+export function getPublicIpViaStun() {
+  return new Promise<string>((resolve, reject) => {
+    reject(new Error('test'))
+    stun.request('stun.l.google.com:19302', (err: Error, response: any) => {
+      if (err) {
+        return reject(err)
+      }
+
+      resolve(response.getXorAddress().address)
+    })
+  })
+}
+
+export function getPublicIp() {
+  return fetchPublicIp()
+    .catch(() => fetchPublicIp({ https: true }))
+    .catch(() => getPublicIpViaStun())
+    .catch(() => {
+      throw new Error('Failed to retrieve public ip address')
+    })
+}
+
+function isRtpMessage(message: Buffer) {
+  const payloadType = message.readUInt8(1) & 0x7f
+  return payloadType > 90 || payloadType === 0
+}
+
+export function getSsrc(message: Buffer) {
+  try {
+    const isRtp = isRtpMessage(message)
+    return message.readUInt32BE(isRtp ? 8 : 4)
+  } catch (_) {
+    return null
+  }
+}
+
+export function bindToRandomPort(socket: Socket) {
+  return new Promise<number>(resolve => {
+    // 0 means select a random open port
+    socket.bind(0, () => {
+      const { port } = socket.address() as AddressInfo
+      resolve(port)
+    })
+  })
+}
+
+export function sendUdpHolePunch(
+  socket: Socket,
+  port: number,
+  address: string
+) {
+  socket.send('', port, address)
+}