Feat: support for subscription of streams/metering

.gitignore

@@ -15,4 +15,4 @@ yarn-error.log
 !.yarn/plugins
 !.yarn/releases
 !.yarn/sdks
-!.yarn/versions
+!.yarn/versions

README.md

@@ -7,15 +7,14 @@ It has been tested with _Lawo Ruby_, _Lawo R3lay_, and _Lawo MxGUI_.
 The current version is very losely based on the original library and Mr Gilles Dufour's rewrites. It is however rewritten almost completely from scratch and bears little to no resemblance to earlier libraries.
 
 ### Repository-specific Info for Developers
-* [Developer Info](DEVELOPER.md)
-* [Contribution Guidelines](CONTRIBUTING.md)
-
-### General Sofie System Info
-* [Documentation](https://nrkno.github.io/sofie-core/)
-* [Releases](https://nrkno.github.io/sofie-core/releases)
 
+- [Developer Info](DEVELOPER.md)
+- [Contribution Guidelines](CONTRIBUTING.md)
 
+### General Sofie System Info
 
+- [Documentation](https://nrkno.github.io/sofie-core/)
+- [Releases](https://nrkno.github.io/sofie-core/releases)
 
 ---
 
@@ -26,24 +25,36 @@ The current version is very losely based on the original library and Mr Gilles D
 Get Full tree:
 
 ```javascript
-const { EmberClient } = require('emberplus-connection');
-const client = new EmberClient("10.9.8.7", 9000);
-client.on("error", e => {
-   console.log(e);
-});
+const { EmberClient, StreamManager } = require('emberplus-connection')
+const client = new EmberClient('10.9.8.7', 9000)
+client.on('error', (e) => {
+	console.log(e)
+})
 await client.connect()
+
+// If you want to listen to stream updates - you can do it like this:
+client.on('streamUpdate', (internalNodePath, value) => {
+	console.log('Stream Update:', {
+		path: internalNodePath,
+		value: value,
+	})
+	// You can get the internal node path, the internal path can be different from the path you requested,
+	// depending on wheter you request a numbered node or via the description
+	// the client has a client.getInternalNodePath(node) that you can request and use as reference when subsribing to a node
+})
+
 // Get Root info
 const req = await client.getDirectory(client.tree)
 await req.response
 // Get a Specific Node
-const node = await client.getElementByPath("0.0.2")
-console.log(node);
+const node = await client.getElementByPath('0.0.2')
+console.log(node)
 // Get a node by its path identifiers
-const node2 = await client.getElementByPath("path.to.node"))
-console.log(node2);
+const node2 = await client.getElementByPath('path.to.node')
+console.log(node2)
 // Get a node by its path descriptions
-const node3 = await client.getElementByPath("descr1.descr2.descr3"))
-console.log(node3);
+const node3 = await client.getElementByPath('descr1.descr2.descr3')
+console.log(node3)
 // Expand entire tree under node 0
 await client.expand(client.tree)
 console.log(client.tree)
@@ -70,6 +81,10 @@ client
 	})
 	.then(() => client.getElementByPath('0.2'))
 	.then(async (node) => {
+		// You can get the internal node path, the internal path can be different from the requested,
+		// depending on wheter you request a numbered node or via the description
+		console.log('This is the internal node path :', client.getInternalNodePath(node))
+
 		// For non-streams a getDirectory will automatically subscribe for update
 		return (
 			await client.getDirectory(node, (update) => {
@@ -84,14 +99,22 @@ client
 			console.log(update)
 		})
 	)
+client.on('streamUpdate', (internalNodePath, value) => {
+	console.log('Stream Update:', {
+		path: internalNodePath,
+		value: value,
+	})
+})
 ```
 
 ### Setting New Value
 
 ```javascript
 client = new EmberClient(LOCALHOST, PORT)
 await client.connect()
-await (await client.getDirectory()).response
+await (
+	await client.getDirectory()
+).response
 const req = await client.setValue(await client.getElementByPath('0.0.1'), 'gdnet')
 await req.response
 console.log('result', req.response)
@@ -107,7 +130,9 @@ const { EmberClient, EmberLib } = require('node-emberplus')
 
 const client = new EmberClient(HOST, PORT)
 await client.connect()
-await (await client.getDirectory()).response
+await (
+	await client.getDirectory()
+).response
 const fn = await client.getElementByPath('path.to.function')
 const req = await client.invoke(fn, 1, 2, 3)
 console.log('result', await req.response)
@@ -126,7 +151,7 @@ const {
 	ParameterAccess,
 	MatrixImpl,
 	MatrixType,
-	MatrixAddressingMode
+	MatrixAddressingMode,
 } = require('emberplus-connection')
 
 const s = new EmberServer(9000) // start server on port 9000
@@ -187,14 +212,14 @@ const tree = {
 					undefined,
 					ParameterAccess.ReadWrite
 				)
-			)
+			),
 		}),
 
 		2: new NumberedTreeNodeImpl(2, new EmberNodeImpl('Functions', undefined, undefined, true), {
 			1: new NumberedTreeNodeImpl(
 				1,
 				new EmberFunctionImpl(undefined, undefined) //, [{ type: ParameterType.Boolean, name: 'Test' }])
-			)
+			),
 		}),
 
 		3: new NumberedTreeNodeImpl(3, new EmberNodeImpl('Matrices', undefined, undefined, true), {
@@ -211,13 +236,14 @@ const tree = {
 					5,
 					5
 				)
-			)
-		})
-	})
+			),
+		}),
+	}),
 }
 
 s.init(tree) // initiate the provider with the tree
 ```
+
 ---
 
 _The NRK logo is a registered trademark of Norsk rikskringkasting AS. The license does not grant any right to use, in any way, any trademarks, service marks or logos of Norsk rikskringkasting AS._

examples/lawo_mc2_fader_metering_example.js

@@ -0,0 +1,74 @@
+// Setting the environment variable DEBUG=emberplus-connection:*
+// will show debug information from the emberplus-connection module
+
+process.env.DEBUG = 'emberplus-connection:*'
+// Note: it's also possible to only log parts of the module by using a subset of the debug name,
+// 'emberplus-connection:S101Client' // for the S101Client class
+// 'emberplus-connection:S101Codec' // for the S101Codec class
+// 'emberplus-connection:StreamManager' // for the StreamManager class
+
+const { EmberClient } = require('../dist/index')
+
+//-------------------------------------------------------------------------
+// Client
+// log output from lawo_mc2_fader_metering_mock.js
+// ------------------------------------------------------------------------
+
+const client = new EmberClient('192.168.1.67', 9000)
+let node1InternalNodePath = ''
+
+client.on('disconnected', () => {
+	console.error('Client Lost Ember connection')
+	client.tree = []
+})
+
+// Handle successful connection
+client.on('connected', () => {
+	console.log('Client Found Ember connection')
+	client.tree = []
+
+	client
+		.getDirectory(client.tree)
+		.then((req) => {
+			console.log(' Req:', req)
+			return req.response
+		})
+		.then(() => {
+			console.log(' Getting node...')
+
+			const path_1 = 'Channels.Inputs._1.Metering.Main Level'
+			return client.getElementByPath(path_1)
+		})
+		.then((node1) => {
+			if (!node1) {
+				throw new Error(' Could not find node 1')
+			}
+			console.log('Found node:', node1)
+			node1InternalNodePath = client.getInternalNodePath(node1)
+
+			// Subscribe to changes
+			client.subscribe(node1, (node1) => {
+				const value = node1.contents
+				console.log('Node 1 subscription :', value)
+			})
+		})
+		.catch((error) => {
+			console.error(' Error:', error)
+		})
+})
+client.on('streamUpdate', (internalNodePath, value) => {
+	if (internalNodePath !== node1InternalNodePath) {
+		return
+	}
+	console.log('Stream Update:', {
+		path: internalNodePath,
+		value: value,
+	})
+})
+
+console.log('-----------------------------------------------------------------------------')
+console.log('log output from mc2_fader_metering_example.js')
+console.log('Connecting to Client...')
+client.connect().catch((error) => {
+	console.error('Client 2 Error when connecting:', error)
+})

src/Ember/Client/StreamManager.ts

@@ -0,0 +1,151 @@
+import { EventEmitter } from 'eventemitter3'
+import { Parameter, ParameterType } from '../../model/Parameter'
+import { EmberValue } from '../../types'
+import { Collection } from '../../types/types'
+import { StreamEntry } from '../../model'
+
+import Debug from 'debug'
+
+const debug = Debug('emberplus-connection:StreamManager')
+
+export type StreamManagerEvents = {
+	streamUpdate: [path: string, value: EmberValue]
+}
+
+interface StreamInfo {
+	parameter: Parameter
+	path: string
+	streamIdentifier: number
+	offset: number
+}
+
+export class StreamManager extends EventEmitter<StreamManagerEvents> {
+	private registeredStreams: Map<string, StreamInfo> = new Map()
+	// Lookup by identifier for O(1) access
+	private streamsByIdentifier: Map<number, Set<string>> = new Map()
+
+	constructor() {
+		super()
+	}
+
+	public registerParameter(parameter: Parameter, path: string): void {
+		if (!parameter.streamIdentifier) {
+			debug('Warning: Attempted to register parameter without streamIdentifier')
+			return
+		}
+		// Check if already registered
+		if (this.registeredStreams.has(path)) {
+			debug('Stream already registered:', {
+				path,
+				identifier: parameter.streamIdentifier,
+			})
+			return
+		}
+
+		const streamInfo: StreamInfo = {
+			parameter,
+			path,
+			streamIdentifier: parameter.streamIdentifier,
+			offset: parameter.streamDescriptor?.offset || 0,
+		}
+
+		// Store both mappings
+		this.registeredStreams.set(path, streamInfo)
+
+		// Add to identifier lookup
+		if (!this.streamsByIdentifier.has(parameter.streamIdentifier)) {
+			this.streamsByIdentifier.set(parameter.streamIdentifier, new Set())
+			debug('Registered new stream identifier and adding set:', parameter.streamIdentifier)
+		}
+		this.streamsByIdentifier.get(parameter.streamIdentifier)?.add(path)
+
+		debug('Registered new stream:', {
+			path,
+			identifier: parameter.streamIdentifier,
+			totalRegistered: this.registeredStreams.size,
+		})
+	}
+
+	public unregisterParameter(path: string): void {
+		const streamInfo = this.registeredStreams.get(path)
+		if (streamInfo?.streamIdentifier) {
+			// Clean up both maps
+			this.registeredStreams.delete(path)
+			const paths = this.streamsByIdentifier.get(streamInfo.streamIdentifier)
+			if (paths) {
+				paths.delete(path)
+				if (paths.size === 0) {
+					this.streamsByIdentifier.delete(streamInfo.streamIdentifier)
+				}
+			}
+
+			debug('Unregistered stream:', {
+				path: path,
+				identifier: streamInfo.parameter.identifier,
+			})
+		}
+	}
+
+	public getStreamInfoByPath(path: string): StreamInfo | undefined {
+		return this.registeredStreams.get(path)
+	}
+
+	public hasStream(identifier: string): boolean {
+		return this.registeredStreams.has(identifier)
+	}
+
+	public updateStreamValues(streamEntries: Collection<StreamEntry>): void {
+		Object.values<StreamEntry>(streamEntries).forEach((streamEntry) => {
+			// O(1) lookup by identifier
+			const paths = this.streamsByIdentifier.get(streamEntry.identifier)
+
+			if (!paths) {
+				debug('Received update for unregistered stream:', streamEntry.identifier)
+				return
+			}
+
+			// Process each matching stream
+			paths.forEach((path) => {
+				const streamInfo = this.registeredStreams.get(path)
+				if (!streamInfo || !streamEntry.value) return
+
+				if (streamEntry.value.type === ParameterType.Integer) {
+					this.updateStreamValue(path, streamEntry.value.value)
+				} else if (streamEntry.value.type === ParameterType.Octets && Buffer.isBuffer(streamEntry.value.value)) {
+					const buffer = streamEntry.value.value
+					if (buffer.length >= streamInfo.offset + 4) {
+						const view = new DataView(buffer.buffer, buffer.byteOffset, buffer.length)
+						const decodedValue = view.getFloat32(streamInfo.offset, true)
+						this.updateStreamValue(path, decodedValue)
+					}
+				}
+			})
+		})
+	}
+
+	public updateStreamValue(path: string, value: EmberValue): void {
+		if (path) {
+			const streamInfo = this.registeredStreams.get(path)
+			if (streamInfo) {
+				streamInfo.parameter.value = value
+				this.emit('streamUpdate', path, value)
+			}
+		}
+	}
+
+	public getAllRegisteredPaths(): string[] {
+		return Array.from(this.registeredStreams.keys())
+	}
+
+	// Debug helper
+	public printStreamState(): void {
+		debug('\nCurrent Stream State:')
+		debug('Registered Streams:')
+		this.registeredStreams.forEach((info, path) => {
+			debug(`  Path: ${path}`)
+			debug(`    Identifier: ${info.parameter.identifier}`)
+			debug(`    StreamId: ${info.parameter.streamIdentifier}`)
+			debug(`    Current Value: ${info.parameter.value}`)
+		})
+	}
+}