perf: stream is now a cold observable (#36)

BREAKING CHANGE: change to public API and configuration interface.

Author: patrickmichalina

.node-version

@@ -1 +1 @@
-12.11.0
+12.11.1

.nvmrc

@@ -1 +1 @@
-12.11.0
+12.11.1

README.md

@@ -37,29 +37,30 @@ This package is designed to be run in Node. For the best developer experience us
 $ npm i camera-probe
 ```
 
-## CLI
+## CLI Usage
 For CLI usage its easier to install globally like so:
 ```sh
 $ npm i -g camera-probe
 
 // starting listening
 $ camera-probe
-```
-
-## Usage
-Starts probing the network using the default configuration.
-```ts
-import { devices$ } from 'camera-probe'
 
-devices$().subscribe(console.info)
+// This table will update as cameras come online and offline.
+┌─────────┬───────────┬─────────────┬─────────────────┬──────────────────────────────────────────┬────────────────────────────────────────────────┐
+│ (index) │   Name    │    Model    │       IP        │                   URN                    │                   Endpoint                     │
+├─────────┼───────────┼─────────────┼─────────────────┼──────────────────────────────────────────┼────────────────────────────────────────────────┤
+│    0    │ 'Amcrest' │ 'IP2M-841B' │ '192.168.1.1'   │  '38b4eeff-f5bd-46b9-92e4-30e6acffee73'  │  'http://192.168.1.1/onvif/device_service'     │
+│    1    │  'IPCAM'  │   '631GA'   │ '192.168.1.2'   │  '4f5dcb4f-eea6-4cda-b290-f2b2b7d2f14f'  │  'http://192.168.1.2:80/onvif/device_service'  │
+└─────────┴───────────┴─────────────┴─────────────────┴──────────────────────────────────────────┴────────────────────────────────────────────────┘
 ```
 
+## Programmatic Usage
 ```js
-// example probe results
-// two cameras discovered on the network with ONVIF WS-Discovery via UDP
-// This will be the last emitted value in the observable until a new camera comes online
-// or a camera is disconnected or otherwise fails to respond to a ping.
+import { onvifDevices$ } from 'camera-probe'
+
+onvifDevices$.subscribe(console.log)
 
+// results
 [ { name: 'Amcrest',
     hardware: 'IP2M-8200',
     location: 'china',
@@ -92,33 +93,3 @@ devices$().subscribe(console.info)
     profiles: [ 'Streaming' ],
     xaddrs: [ 'http://192.168.5.13:80/onvif/device_service' ] } ]
 ```
-
-If you'd like to tweak default settings feel free to override in the `.run()` method.
-
-```ts
-import { probe } from 'camera-probe'
-
-probe()
-  .run({
-    PORTS: [3702],
-    PROBE_NETWORK_TIMEOUT_MS: 20000
-  })
-  .subscribe(console.log)
-```
-
-## Default Configuration
-```ts
-export const DEFAULT_CONFIG: IProbeConfig = {
-  DOM_PARSER: new DOMParser(),
-  PORTS: {
-    UPNP: [1900],
-    WS_DISCOVERY: [3702]
-  },
-  MULTICAST_ADDRESS: '239.255.255.250',
-  PROBE_SAMPLE_TIME_MS: 2000,
-  PROBE_NETWORK_TIMEOUT_MS: 2000 * 1.5,
-  PROBE_SAMPLE_START_DELAY_TIME_MS: 0,
-  ONVIF_DEVICES: ['NetworkVideoTransmitter', 'Device', 'NetworkVideoDisplay'],
-  NOT_FOUND_STRING: 'unknown'
-}
-```

package-lock.json

@@ -52,7 +52,7 @@
     "@types/node": "^12.7.11",
     "@types/xmldom": "^0.1.29",
     "jest": "^24.9.0",
-    "rollup": "^1.23.0",
+    "rollup": "^1.23.1",
     "rollup-plugin-commonjs": "^10.1.0",
     "rollup-plugin-typescript2": "^0.24.3",
     "semantic-release": "^15.13.24",

package.json

@@ -1,15 +1,3 @@
-import { IProbeConfig } from './config.interface'
 import { DOMParser } from 'xmldom'
-
-export const DEFAULT_CONFIG: IProbeConfig = {
-  DOM_PARSER: new DOMParser(),
-  MULTICAST_ADDRESS: '239.255.255.250',
-  PROBE_SAMPLE_TIME_MS: 2000,
-  PROBE_NETWORK_TIMEOUT_MS: 12000,
-  FALLOUT_MS: 4000,
-  ONVIF_DEVICES: ['NetworkVideoTransmitter', 'Device', 'NetworkVideoDisplay'],
-  PORTS: {
-    UPNP: [1900],
-    WS_DISCOVERY: [3702]
-  }
-}
+const dom = new DOMParser()
+export const XML_PARSER_FN = (str: string) => dom.parseFromString(str, 'application/xml')

src/config/config.default.ts

@@ -1,6 +1,32 @@
+import { SocketType } from 'dgram'
+
 export type Strings = readonly string[]
 export type Numbers = readonly number[]
 
 export interface TimestampedMessage { readonly msg: string, readonly ts: number }
 export type TimestampMessages = readonly TimestampedMessage[]
 export type StringDictionary = { readonly [key: string]: string }
+
+export interface IProbeConfig {
+  readonly SOCKET_PROTOCOL: SocketType
+  readonly PORTS: Numbers
+  readonly MULTICAST_ADDRESS: string
+  readonly PROBE_REQUEST_SAMPLE_RATE_MS: number
+  readonly PROBE_RESPONSE_TIMEOUT_MS: number
+  readonly PROBE_RESPONSE_FALLOUT_MS: number
+  readonly RESULT_DEDUPE_FN: (msg: TimestampMessages) => StringDictionary
+}
+
+export const DEFAULT_PROBE_CONFIG: IProbeConfig = {
+  PORTS: [],
+  SOCKET_PROTOCOL: 'udp4',
+  MULTICAST_ADDRESS: '239.255.255.250',
+  PROBE_REQUEST_SAMPLE_RATE_MS: 3000,
+  PROBE_RESPONSE_FALLOUT_MS: 4000,
+  PROBE_RESPONSE_TIMEOUT_MS: 6000,
+  RESULT_DEDUPE_FN: (msg: TimestampMessages) => {
+    return msg.reduce((acc, curr) => {
+      return { ...acc, curr }
+    }, {})
+  }
+}

src/config/config.interface.ts

@@ -1,4 +1,4 @@
-import { flattenBuffersWithInfo } from '../core/probe'
+import { flattenBuffersWithInfo } from './probe'
 
 describe('Probe', () => {
   it('should flatten buffers to ports', done => {

src/core/interfaces.ts

@@ -1,28 +1,25 @@
-import { map, filter, scan, distinctUntilChanged, takeUntil, mapTo, tap } from 'rxjs/operators'
-import { ISocketStream, socketStream } from '../core/socket-stream'
-import { reader, IResult } from 'typescript-monads'
-import { Observable, timer } from 'rxjs'
-import { IProbeConfig } from '../config/config.interface'
-import { Strings, Numbers } from '../core/interfaces'
+import { createSocket, RemoteInfo } from 'dgram'
+import { Strings, Numbers, IProbeConfig, DEFAULT_PROBE_CONFIG } from './interfaces'
+import { Observable, Observer, fromEvent, timer } from 'rxjs'
+import { first, shareReplay, map, distinctUntilChanged, mapTo, takeWhile, takeUntil, scan } from 'rxjs/operators'
 
+type IMessage = readonly [Buffer, RemoteInfo]
 type TimestampMessages = readonly TimestampedMessage[]
 type StringDictionary = { readonly [key: string]: string }
 interface TimestampedMessage { readonly msg: string, readonly ts: number }
 interface BufferPort { readonly buffer: Buffer, readonly port: number, readonly address: string }
 
-const flattenXml = (str: string) => str.replace(/>\s*/g, '>').replace(/\s*</g, '<')
 const mapStringToBuffer = (str: string) => Buffer.from(str, 'utf8')
+const flattenXml = (str: string) => str.replace(/>\s*/g, '>').replace(/\s*</g, '<')
 const toArrayOfValues = <T extends StringDictionary>(source: Observable<T>) => source.pipe(map(a => Object.keys(a).map(b => a[b])))
 const flattenDocumentStrings = (source: Observable<Strings>) => source.pipe(map(a => a.map(flattenXml)))
-const filterOkResults = <TOk, TFail>(source: Observable<IResult<TOk, TFail>>) => source.pipe(filter(a => a.isOk()))
-const timestamp = <TFail>(source: Observable<IResult<Buffer, TFail>>) => source.pipe(map<IResult<Buffer, TFail>, TimestampedMessage>(a => ({ msg: a.unwrap().toString(), ts: Date.now() })))
+const timestamp = (source: Observable<Buffer>) => source.pipe(map<Buffer, TimestampedMessage>(a => ({ msg: a.toString(), ts: Date.now() })))
 const distinctUntilObjectChanged = <T>(source: Observable<T>) => source.pipe(distinctUntilChanged((a, b) => {
   const keys1 = Object.keys(a)
   const keys2 = Object.keys(b)
- 
-  return  keys1.length === keys2.length && keys1.reduce((acc: boolean, curr) => {
-    return acc === false ? false : keys2.includes(curr) as boolean
-  }, true)
+
+  return keys1.length === keys2.length && 
+    keys1.reduce((acc: boolean, curr) => acc === false ? false : keys2.includes(curr) as boolean, true)
 }))
 
 const accumulateFreshMessages =
@@ -42,27 +39,27 @@ export const flattenBuffersWithInfo =
         ports.reduce((acc, port) =>
           [...acc, ...buffers.map(buffer => ({ buffer, port, address }))], [] as readonly BufferPort[])
 
-export const initSocketStream = reader<IProbeConfig, ISocketStream>(c => socketStream('udp4', c.PROBE_NETWORK_TIMEOUT_MS, c.distinctFilterFn))
-
 export const probe =
-  (socket: ISocketStream) =>
-    (ports: Numbers) =>
-      (address: string) =>
-        (messagesToSend: Strings = []) =>
-          (mapFn: (msg: readonly TimestampedMessage[]) => StringDictionary) =>
-            reader<IProbeConfig, Observable<Strings>>(cfg => {
-              timer(0, cfg.PROBE_SAMPLE_TIME_MS).pipe(
-                mapTo(flattenBuffersWithInfo(ports)(address)(messagesToSend.map(mapStringToBuffer))),
-                takeUntil(socket.close$))
-                .subscribe(bfrPorts => bfrPorts.forEach(mdl => socket.socket.send(mdl.buffer, 0, mdl.buffer.length, mdl.port, mdl.address)))
+  (config?: Partial<IProbeConfig>) =>
+    (messages: Strings): Observable<Strings> =>
+      Observable.create((obs: Observer<Strings>) => {
+        const cfg = { ...DEFAULT_PROBE_CONFIG, ...(config || {}) }
+        const socket = createSocket({ type: 'udp4' })
+        const socketClosed$ = fromEvent<void>(socket, 'close').pipe(first(), shareReplay(1))
+        const socketMessages$ = fromEvent<IMessage>(socket, 'message').pipe(map(a => a[0]), shareReplay(1))
+
+        timer(0, cfg.PROBE_REQUEST_SAMPLE_RATE_MS).pipe(
+          mapTo(flattenBuffersWithInfo(cfg.PORTS)(cfg.MULTICAST_ADDRESS)(messages.map(mapStringToBuffer))),
+          takeWhile(() => !obs.closed),
+          takeUntil(socketClosed$))
+          .subscribe(bfrPorts => bfrPorts.forEach(mdl => socket.send(mdl.buffer, 0, mdl.buffer.length, mdl.port, mdl.address)))
 
-              return socket.messages$.pipe(
-                filterOkResults,
-                timestamp,
-                accumulateFreshMessages(cfg.FALLOUT_MS),
-                mapStrToDictionary(mapFn),
-                distinctUntilObjectChanged,
-                toArrayOfValues,
-                flattenDocumentStrings
-              )
-            })
+        socketMessages$.pipe(
+          timestamp,
+          accumulateFreshMessages(cfg.PROBE_RESPONSE_FALLOUT_MS),
+          mapStrToDictionary(cfg.RESULT_DEDUPE_FN),
+          distinctUntilObjectChanged,
+          toArrayOfValues,
+          flattenDocumentStrings
+        ).subscribe(msg => obs.next(msg))
+      })