diff --git a/rpc/.vscode/launch.json b/rpc/.vscode/launch.json index 41d62525d..64885145d 100644 --- a/rpc/.vscode/launch.json +++ b/rpc/.vscode/launch.json @@ -9,7 +9,7 @@ "type": "python", "request": "launch", "program": "${workspaceFolder}/rpc.py", - "console": "integratedTerminal" + "console": "integratedTerminal", }, { "type": "pwa-node", diff --git a/rpc/rpc.py b/rpc/rpc.py index 5b300d37f..3fb0a1a9c 100644 --- a/rpc/rpc.py +++ b/rpc/rpc.py @@ -1,3 +1,4 @@ +import debugpy from asyncio.events import AbstractEventLoop from asyncio.futures import Future from typing import Callable @@ -337,7 +338,7 @@ async def readLoop(loop, peer, reader): pass -async def main(loop: AbstractEventLoop): +async def async_main(loop: AbstractEventLoop): reader, writer = await asyncio.open_connection( '127.0.0.1', 3033) @@ -368,6 +369,12 @@ async def main(loop: AbstractEventLoop): # # pokemon = json.loads(contents) # # print(pokemon['name']) -loop = asyncio.get_event_loop() -loop.run_until_complete(main(loop)) -loop.close() + +def main(): + loop = asyncio.get_event_loop() + loop.run_until_complete(async_main(loop)) + loop.close() + + +if __name__ == "__main__": + main() diff --git a/sdk/bin/scrypted-webpack.js b/sdk/bin/scrypted-webpack.js index de6e111ba..61e8e3434 100755 --- a/sdk/bin/scrypted-webpack.js +++ b/sdk/bin/scrypted-webpack.js @@ -17,8 +17,26 @@ const rimraf = require('rimraf'); const webpack = require('webpack'); const esbuild = require('esbuild'); +if (fs.existsSync(path.resolve(cwd, 'src/main.py'))) { -if (false) { + let out; + if (process.env.NODE_ENV == 'production') + out = path.resolve(cwd, 'dist'); + else + out = path.resolve(cwd, 'out'); + + const resolved = path.resolve(cwd, 'src/main.py'); + + const zip = new AdmZip(); + + zip.addLocalFile(resolved); + + const zipfs = path.join(cwd, 'fs'); + if (fs.existsSync(zipfs)) + zip.addLocalFolder(zipfs, 'fs'); + zip.writeZip(path.join(out, 'plugin.zip')); +} +else if (false) { let out; if (process.env.NODE_ENV == 'production') diff --git a/server/python/__pycache__/rpc.cpython-39.pyc b/server/python/__pycache__/rpc.cpython-39.pyc new file mode 100644 index 000000000..c66cb4c7e Binary files /dev/null and b/server/python/__pycache__/rpc.cpython-39.pyc differ diff --git a/server/python/plugin-remote.py b/server/python/plugin-remote.py new file mode 100644 index 000000000..8deb57a2a --- /dev/null +++ b/server/python/plugin-remote.py @@ -0,0 +1,111 @@ +from collections.abc import Mapping, Sequence +from rpc import RpcPeer, readLoop +import asyncio +from asyncio.events import AbstractEventLoop +import json +import aiofiles +import os +from typing import TypedDict + + +class SystemDeviceState(TypedDict): + lastEventTime: int + stateTime: int + value: any + + +class DeviceStorage: + id: str + nativeId: str + storage: Mapping[str, str] = {} + + +class PluginRemote: + systemState: Mapping[str, Mapping[str, SystemDeviceState]] = {} + nativeIds: Mapping[str, DeviceStorage] = {} + pluginId: str + + def __init__(self, api, pluginId): + self.api = api + self.pluginId = pluginId + + async def loadZip(self, packageJson, zipData, options=None): + pass + + async def setSystemState(self, state): + self.systemState = state + + async def setNativeId(self, nativeId, id, storage): + if nativeId: + ds = DeviceStorage() + ds.id = id + ds.storage = storage + self.nativeIds[nativeId] = ds + else: + self.nativeIds.pop(nativeId, None) + + async def updateDeviceState(self, id, state): + if not state: + self.systemState.pop(id, None) + else: + self.systemState[id] = state + + async def notify(self, id, eventTime, eventInterface, property, value, changed = False): + if property: + state = None + if self.systemState: + state = self.systemState.get(id, None) + if not state: + print('state not found for %s' % id) + return + state[property] = value + # systemManager.events.notify(id, eventTime, eventInterface, property, value.value, changed); + else: + # systemManager.events.notify(id, eventTime, eventInterface, property, value, changed); + pass + + async def ioEvent(self, id, event, message = None): + pass + + async def createDeviceState(self, id, setState): + pass + + async def getServicePort(self, name): + pass + +async def async_main(loop: AbstractEventLoop): + reader = await aiofiles.open(3, mode='r') + # writer = open(4, 'r+') + + def send(message, reject = None): + jsonString = json.dumps(message) + try: + os.write(4, bytes(jsonString + '\n', 'utf8')) + except Exception as e: + if reject: + reject(e) + + peer = RpcPeer(send) + peer.params['print'] = print + peer.params['getRemote'] = lambda api, pluginId: PluginRemote(api, pluginId) + + async def consoleTest(): + console = await peer.getParam('console') + # await console.log('test', 'poops', 'peddeps') + + await asyncio.gather(readLoop(loop, peer, reader), consoleTest()) + print('done') + + # print("line %s" % line) + + + + +def main(): + loop = asyncio.get_event_loop() + loop.run_until_complete(async_main(loop)) + loop.close() + + +if __name__ == "__main__": + main() diff --git a/server/python/rpc.py b/server/python/rpc.py new file mode 100644 index 000000000..d15f39fb8 --- /dev/null +++ b/server/python/rpc.py @@ -0,0 +1,382 @@ +from asyncio.events import AbstractEventLoop +from asyncio.futures import Future +from typing import Callable +import asyncio +import json +import traceback +import inspect +import json +from collections.abc import Mapping, Sequence +import weakref + +jsonSerializable = set() +jsonSerializable.add(float) +jsonSerializable.add(int) +jsonSerializable.add(str) +jsonSerializable.add(dict) +jsonSerializable.add(bool) +jsonSerializable.add(list) + + +async def maybe_await(value): + if (inspect.iscoroutinefunction(value) or inspect.iscoroutine(value)): + return await value + return value + + +class RpcResultException(Exception): + name = None + stack = None + + def __init__(self, caught, message): + self.caught = caught + self.message = message + + +class RpcSerializer: + def serialize(self, value): + pass + + def deserialize(self, value): + pass + + +class RpcProxyMethod: + def __init__(self, proxy, name): + self.__proxy = proxy + self.__proxy_method_name = name + + def __call__(self, *args, **kwargs): + return self.__proxy.__apply__(self.__proxy_method_name, args) + + +class RpcProxy: + def __init__(self, peer, proxyId: str, proxyConstructorName: str, proxyProps: any, proxyOneWayMethods: list[str]): + self.__proxy_id = proxyId + self.__proxy_constructor = proxyConstructorName + self.__proxy_peer = peer + self.__proxy_props = proxyProps + self.__proxy_oneway_methods = proxyOneWayMethods + + def __getattr__(self, name): + if self.__proxy_props and hasattr(self.__proxy_props, name): + return self.__proxy_props[name] + return RpcProxyMethod(self, name) + + def __call__(self, *args, **kwargs): + print('call') + pass + + def __apply__(self, method: str, args: list): + return self.__proxy_peer.__apply__(self.__proxy_id, self.__proxy_oneway_methods, method, args) + + +class RpcPeer: + idCounter = 1 + peerName = 'Unnamed Peer' + params: Mapping[str, any] = {} + localProxied: Mapping[any, str] = {} + localProxyMap: Mapping[str, any] = {} + constructorSerializerMap = {} + proxyCounter = 1 + pendingResults: Mapping[str, Future] = {} + remoteWeakProxies: Mapping[str, any] = {} + nameDeserializerMap: Mapping[str, RpcSerializer] = {} + + def __init__(self, send: Callable[[object, Callable[[Exception], None]], None]) -> None: + self.send = send + + def __apply__(self, proxyId: str, oneWayMethods: list[str], method: str, argArray: list): + args = [] + for arg in argArray: + args.append(self.serialize(arg, False)) + + rpcApply = { + 'type': 'apply', + 'id': None, + 'proxyId': proxyId, + 'argArray': args, + 'method': method, + } + + if not oneWayMethods or method not in oneWayMethods: + rpcApply['oneway'] = True + self.send(rpcApply) + future = Future() + future.set_result(None) + return future + + async def send(id: str, reject: Callable[[Exception], None]): + rpcApply['id'] = id + self.send(rpcApply, reject) + return self.createPendingResult(send) + + def kill(self): + self.killed = True + + def createErrorResult(self, result: any, name: str, message: str, tb: str): + result['stack'] = tb if tb else 'no stack' + result['result'] = name if name else 'no name' + result['message'] = message if message else 'no message' + + def serialize(self, value, requireProxy): + if (not value or (not requireProxy and type(value) in jsonSerializable)): + return value + __remote_constructor_name = 'Function' if callable(value) else value.__proxy_constructor if hasattr( + value, '__proxy_constructor') else type(value).__name__ + proxyId = self.localProxied.get(value, None) + if proxyId: + ret = { + '__remote_proxy_id': proxyId, + '__remote_constructor_name': __remote_constructor_name, + '__remote_proxy_props': getattr(value, '__proxy_props', None), + '__remote_proxy_oneway_methods': getattr(value, '__proxy_oneway_methods', None), + } + return ret + + __proxy_id = getattr(value, '__proxy_id', None) + __proxy_peer = getattr(value, '__proxy_peer', None) + if __proxy_id and __proxy_peer == self: + ret = { + '__local_proxy_id': __proxy_id, + } + return ret + + serializerMapName = self.constructorSerializerMap.get( + type(value).__name__) + if serializerMapName: + __remote_constructor_name = serializerMapName + serializer = self.nameDeserializerMap.get(serializerMapName, None) + serialized = serializer.serialize(value) + if not serialized or (not requireProxy and type(serialized).__name in jsonSerializable): + ret = { + '__remote_proxy_id': None, + '__remote_constructor_name': __remote_constructor_name, + '__remote_proxy_props': getattr(value, '__proxy_props', None), + '__remote_proxy_oneway_methods': getattr(value, '__proxy_oneway_methods', None), + '__serialized_value': value, + } + return ret + + proxyId = str(self.proxyCounter) + self.proxyCounter = self.proxyCounter + 1 + self.localProxied[value] = proxyId + self.localProxyMap[proxyId] = value + + ret = { + '__remote_proxy_id': proxyId, + '__remote_constructor_name': __remote_constructor_name, + '__remote_proxy_props': getattr(value, '__proxy_props', None), + '__remote_proxy_oneway_methods': getattr(value, '__proxy_oneway_methods', None), + } + + return ret + + def finalize(self, id: str): + pass + + def newProxy(self, proxyId: str, proxyConstructorName: str, proxyProps: any, proxyOneWayMethods: list[str]): + proxy = RpcProxy(self, proxyId, proxyConstructorName, + proxyProps, proxyOneWayMethods) + wr = weakref.ref(proxy) + self.remoteWeakProxies[proxyId] = wr + weakref.finalize(proxy, lambda: self.finalize(proxyId)) + return proxy + + def deserialize(self, value): + if not value: + return value + + if type(value) != dict: + return value + + __remote_proxy_id = value.get('__remote_proxy_id', None) + __local_proxy_id = value.get('__local_proxy_id', None) + __remote_constructor_name = value.get( + '__remote_constructor_name', None) + __serialized_value = value.get('__serialized_value', None) + __remote_proxy_props = value.get('__remote_proxy_props', None) + __remote_proxy_oneway_methods = value.get( + '__remote_proxy_oneway_methods', None) + + if __remote_proxy_id: + weakref = self.remoteWeakProxies.get('__remote_proxy_id', None) + proxy = weakref() if weakref else None + if not proxy: + proxy = self.newProxy(__remote_proxy_id, __remote_constructor_name, + __remote_proxy_props, __remote_proxy_oneway_methods) + return proxy + + if __local_proxy_id: + ret = self.localProxyMap.get(__local_proxy_id, None) + if not ret: + raise RpcResultException( + None, 'invalid local proxy id %s' % __local_proxy_id) + return ret + + deserializer = self.nameDeserializerMap.get( + __remote_constructor_name, None) + if deserializer: + return deserializer.deserialize(__serialized_value) + + return value + + async def handleMessage(self, message: any): + try: + type = message['type'] + if type == 'param': + result = { + 'type': 'result', + 'id': message['id'], + } + + try: + value = self.params.get(message['param'], None) + value = await maybe_await(value) + result['result'] = self.serialize( + value, message.get('requireProxy', None)) + except Exception as e: + tb = traceback.format_exc() + self.createErrorResult( + result, type(e).__name, str(e), tb) + + self.send(result) + + elif type == 'apply': + result = { + 'type': 'result', + 'id': message['id'], + } + method = message.get('method', None) + + try: + target = self.localProxyMap.get( + message['proxyId'], None) + if not target: + raise Exception('proxy id %s not found' % + message['proxyId']) + + args = [] + for arg in (message['argArray'] or []): + args.append(self.deserialize(arg)) + + value = None + if method: + if not hasattr(target, method): + raise Exception( + 'target %s does not have method %s' % (type(target), method)) + invoke = getattr(target, method) + value = await maybe_await(invoke(*args)) + else: + value = await maybe_await(target(*args)) + + result['result'] = self.serialize(value, False) + except Exception as e: + print('failure', method, e) + tb = traceback.format_exc() + self.createErrorResult( + result, type(e).__name, str(e), tb) + + if not message.get('oneway', False): + self.send(result) + + elif type == 'result': + future = self.pendingResults.get(message['id'], None) + if not future: + raise RpcResultException( + None, 'unknown result %s' % message['id']) + del message['id'] + if hasattr(message, 'message') or hasattr(message, 'stack'): + e = RpcResultException( + None, message.get('message', None)) + e.stack = message.get('stack', None) + e.name = message.get('name', None) + future.set_exception(e) + return + future.set_result(self.deserialize( + message.get('result', None))) + elif type == 'finalize': + local = self.localProxyMap.pop( + message['__local_proxy_id'], None) + self.localProxied.pop(local, None) + else: + raise RpcResultException( + None, 'unknown rpc message type %s' % type) + except Exception as e: + print("unhandled rpc error", self.peerName, e) + pass + + async def createPendingResult(self, cb: Callable[[str, Callable[[Exception], None]], None]): + # if (Object.isFrozen(this.pendingResults)) + # return Promise.reject(new RPCResultError('RpcPeer has been killed')); + + id = str(self.idCounter) + self.idCounter = self.idCounter + 1 + future = Future() + self.pendingResults[id] = future + await cb(id, lambda e: future.set_exception(RpcResultException(e, None))) + return await future + + async def getParam(self, param): + async def send(id: str, reject: Callable[[Exception], None]): + paramMessage = { + 'id': id, + 'type': 'param', + 'param': param, + } + self.send(paramMessage, reject) + return await self.createPendingResult(send) + +# c = RpcPeer() + + +async def readLoop(loop, peer, reader): + async for line in reader: + try: + message = json.loads(line) + asyncio.run_coroutine_threadsafe(peer.handleMessage(message), loop) + except Exception as e: + print('read loop error', e) + pass + + +async def async_main(loop: AbstractEventLoop): + reader, writer = await asyncio.open_connection( + '127.0.0.1', 3033) + + async def send(message, reject): + jsonString = json.dumps(message) + writer.write(bytes(jsonString + '\n', 'utf8')) + try: + await writer.drain() + except Exception as e: + if reject: + reject(e) + + peer = RpcPeer(send) + peer.params['print'] = print + + async def consoleTest(): + console = await peer.getParam('console') + await console.log('test', 'poops', 'peddeps') + + await asyncio.gather(readLoop(loop, peer, reader), consoleTest()) + print('done') + + # print("line %s" % line) + + # async with aiofiles.open(0, mode='r') as f: + # async for line in f: + # print("line %s" % line) + # # pokemon = json.loads(contents) + # # print(pokemon['name']) + + +def main(): + loop = asyncio.get_event_loop() + loop.run_until_complete(async_main(loop)) + loop.close() + + +if __name__ == "__main__": + main() diff --git a/server/src/plugin/plugin-api.ts b/server/src/plugin/plugin-api.ts index 4a0bceebd..e5638bdcf 100644 --- a/server/src/plugin/plugin-api.ts +++ b/server/src/plugin/plugin-api.ts @@ -140,7 +140,7 @@ export interface PluginRemote { loadZip(packageJson: any, zipData: Buffer, options?: PluginRemoteLoadZipOptions): Promise; setSystemState(state: {[id: string]: {[property: string]: SystemDeviceState}}): Promise; setNativeId(nativeId: ScryptedNativeId, id: string, storage: {[key: string]: any}): Promise; - updateDescriptor(id: string, state: {[property: string]: SystemDeviceState}): Promise; + updateDeviceState(id: string, state: {[property: string]: SystemDeviceState}): Promise; notify(id: string, eventTime: number, eventInterface: string, property: string|undefined, value: SystemDeviceState|any, changed?: boolean): Promise; ioEvent(id: string, event: string, message?: any): Promise; diff --git a/server/src/plugin/plugin-host.ts b/server/src/plugin/plugin-host.ts index 25701baff..7d1556e6b 100644 --- a/server/src/plugin/plugin-host.ts +++ b/server/src/plugin/plugin-host.ts @@ -24,6 +24,8 @@ import { install as installSourceMapSupport } from 'source-map-support'; import net from 'net' import child_process from 'child_process'; import { PluginDebug } from './plugin-debug'; +import readline from 'readline'; +import { Readable, Writable } from 'stream'; export class PluginHost { worker: child_process.ChildProcess; @@ -88,44 +90,16 @@ export class PluginHost { this.packageJson = plugin.packageJson; const logger = scrypted.getDeviceLogger(scrypted.findPluginDevice(plugin._id)); - if (true) { - const cwd = path.join(process.cwd(), 'volume', 'plugins', this.pluginId); - try { - mkdirp.sync(cwd); - } - catch (e) { - } - - this.startPluginClusterHost(logger, { - SCRYPTED_PLUGIN_VOLUME: cwd, - }); + const cwd = path.join(process.cwd(), 'volume', 'plugins', this.pluginId); + try { + mkdirp.sync(cwd); } - else { - const remote = new RpcPeer((message, reject) => { - try { - this.peer.handleMessage(message); - } - catch (e) { - if (reject && reject) - reject(e); - } - }); - - this.peer = new RpcPeer((message, reject) => { - try { - remote.handleMessage(message); - } - catch (e) { - if (reject) - reject(e); - } - }); - - attachPluginRemote(remote, { - createMediaManager: async (systemManager) => new MediaManagerImpl(systemManager, console), - }); + catch (e) { } + this.startPluginClusterHost(logger, { + SCRYPTED_PLUGIN_VOLUME: cwd, + }, plugin.packageJson.scrypted.runtime); this.io.on('connection', async (socket) => { try { @@ -219,23 +193,82 @@ export class PluginHost { }); } - startPluginClusterHost(logger: Logger, env?: any) { - const execArgv: string[] = process.execArgv.slice(); - if (this.pluginDebug) { - execArgv.push(`--inspect=0.0.0.0:${this.pluginDebug.inspectPort}`); + startPluginClusterHost(logger: Logger, env?: any, runtime?: string) { + let connected = true; + + if (runtime === 'python') { + const args: string[] = []; + if (this.pluginDebug) { + args.push( + '-m', + 'debugpy', + '--listen', + `0.0.0.0:${this.pluginDebug.inspectPort}`, + '--wait-for-client', + path.join(__dirname, '../../python', 'plugin-remote.py'), + ) + } + + this.worker = child_process.spawn('python', args, { + // stdin, stdout, stderr, peer in, peer out + stdio: ['pipe', 'pipe', 'pipe', 'pipe', 'pipe'], + }); + + const peerin = this.worker.stdio[3] as Writable; + const peerout = this.worker.stdio[4] as Readable; + peerout.on('data', data => { + console.log(data.toString()); + }) + + this.peer = new RpcPeer((message, reject) => { + if (connected) { + peerin.write(JSON.stringify(message) + '\n', e => e && reject?.(e)); + } + else if (reject) { + reject(new Error('peer disconnected')); + } + }); + + const readInterface = readline.createInterface({ + input: peerout, + terminal: false, + }); + readInterface.on('line', line => { + this.peer.handleMessage(JSON.parse(line)); + }); + } + else { + const execArgv: string[] = process.execArgv.slice(); + if (this.pluginDebug) { + execArgv.push(`--inspect=0.0.0.0:${this.pluginDebug.inspectPort}`); + } + + this.worker = child_process.fork(require.main.filename, ['child', JSON.stringify(env)], { + stdio: 'pipe', + serialization: 'advanced', + execArgv, + }); + + this.peer = new RpcPeer((message, reject) => { + if (connected) { + this.worker.send(message, undefined, e => { + if (e && reject) + reject(e); + }); + } + else if (reject) { + reject(new Error('peer disconnected')); + } + }); + + this.worker.on('message', message => this.peer.handleMessage(message as any)); } - this.worker = child_process.fork(require.main.filename, ['child', JSON.stringify(env)], { - stdio: 'pipe', - serialization: 'advanced', - execArgv, - }); this.worker.stdout.on('data', data => { process.stdout.write(data); }); this.worker.stderr.on('data', data => process.stderr.write(data)); - let connected = true; this.worker.on('disconnect', () => { connected = false; logger.log('e', `${this.pluginName} disconnected`); @@ -248,19 +281,7 @@ export class PluginHost { connected = false; logger.log('e', `${this.pluginName} error ${e}`); }); - this.worker.on('message', message => this.peer.handleMessage(message as any)); - this.peer = new RpcPeer((message, reject) => { - if (connected) { - this.worker.send(message, undefined, e => { - if (e && reject) - reject(e); - }); - } - else if (reject) { - reject(new Error('peer disconnected')); - } - }); this.peer.peerName = this.pluginId; this.peer.onOob = (oob: any) => { @@ -519,7 +540,7 @@ export function startPluginClusterWorker() { // deleted? return; } - const {pluginId, nativeId: mixinNativeId} = await plugins.getDeviceInfo(mixinId); + const { pluginId, nativeId: mixinNativeId } = await plugins.getDeviceInfo(mixinId); const port = await plugins.getRemoteServicePort(pluginId, 'console-writer'); const socket = net.connect(port); socket.write(mixinNativeId + '\n'); @@ -626,10 +647,10 @@ class LazyRemote implements PluginRemote { await this.remoteReadyPromise; return this.remote.setNativeId(nativeId, id, storage); } - async updateDescriptor(id: string, state: { [property: string]: SystemDeviceState; }): Promise { + async updateDeviceState(id: string, state: { [property: string]: SystemDeviceState; }): Promise { if (!this.remote) await this.remoteReadyPromise; - return this.remote.updateDescriptor(id, state); + return this.remote.updateDeviceState(id, state); } async notify(id: string, eventTime: number, eventInterface: string, property: string, propertyState: SystemDeviceState, changed?: boolean): Promise { if (!this.remote) diff --git a/server/src/plugin/plugin-remote.ts b/server/src/plugin/plugin-remote.ts index 975d1a1e8..cedd10d68 100644 --- a/server/src/plugin/plugin-remote.ts +++ b/server/src/plugin/plugin-remote.ts @@ -4,7 +4,7 @@ import path from 'path'; import { ScryptedNativeId, DeviceManager, Logger, Device, DeviceManifest, DeviceState, EndpointManager, SystemDeviceState, ScryptedStatic, SystemManager, MediaManager, ScryptedMimeTypes, ScryptedInterface, ScryptedInterfaceProperty, HttpRequest } from '@scrypted/sdk/types' import { PluginAPI, PluginLogger, PluginRemote, PluginRemoteLoadZipOptions } from './plugin-api'; import { SystemManagerImpl } from './system'; -import { RpcPeer } from '../rpc'; +import { RpcPeer, RPCResultError } from '../rpc'; import { BufferSerializer } from './buffer-serializer'; import { EventEmitter } from 'events'; import { createWebSocketClass } from './plugin-remote-websocket'; @@ -271,9 +271,14 @@ interface WebSocketCallbacks { export async function setupPluginRemote(peer: RpcPeer, api: PluginAPI, pluginId: string): Promise { - peer.addSerializer(Buffer, 'Buffer', new BufferSerializer()); - const getRemote = await peer.getParam('getRemote'); - return getRemote(api, pluginId); + try { + peer.addSerializer(Buffer, 'Buffer', new BufferSerializer()); + const getRemote = await peer.getParam('getRemote'); + return await getRemote(api, pluginId); + } + catch (e) { + throw new RPCResultError('error while retrieving PluginRemote', e); + } } export interface PluginRemoteAttachOptions { @@ -325,7 +330,7 @@ export function attachPluginRemote(peer: RpcPeer, options?: PluginRemoteAttachOp __proxy_required: true, __proxy_oneway_methods: [ 'notify', - 'updateDescriptor', + 'updateDeviceState', 'setSystemState', 'ioEvent', 'setNativeId', @@ -367,7 +372,7 @@ export function attachPluginRemote(peer: RpcPeer, options?: PluginRemoteAttachOp } }, - async updateDescriptor(id: string, state: { [property: string]: SystemDeviceState }) { + async updateDeviceState(id: string, state: { [property: string]: SystemDeviceState }) { if (!state) { delete systemManager.state[id]; systemManager.events.notify(id, Date.now(), ScryptedInterface.ScryptedDevice, ScryptedInterfaceProperty.id, id, true); diff --git a/server/src/scrypted-main.ts b/server/src/scrypted-main.ts index f452eb0c8..9e74505ec 100644 --- a/server/src/scrypted-main.ts +++ b/server/src/scrypted-main.ts @@ -25,6 +25,7 @@ import httpAuth from 'http-auth'; import semver from 'semver'; import { Info } from './services/info'; import { getAddresses } from './addresses'; +import { sleep } from './sleep'; if (!semver.gte(process.version, '16.0.0')) { throw new Error('"node" version out of date. Please update node to v16 or higher.') @@ -64,23 +65,41 @@ else { let workerInspectPort: number = undefined; - const debugServer = net.createServer(socket => { + async function doconnect(): Promise { + return new Promise((resolve, reject) => { + const target = net.connect(workerInspectPort); + target.once('error', reject) + target.once('connect', () => resolve(target)) + }) + } + + const debugServer = net.createServer(async (socket) => { if (!workerInspectPort) { socket.destroy(); return; } - const target = net.connect(workerInspectPort); - socket.pipe(target).pipe(socket); - socket.on('error', () => { - socket.destroy(); - target.destroy(); - }); - target.on('error', e => { - console.error('debugger target error', e); - socket.destroy(); - target.destroy(); - }); + for (let i = 0; i < 10; i++) { + try { + const target = await doconnect(); + socket.pipe(target).pipe(socket); + socket.on('error', () => { + socket.destroy(); + target.destroy(); + }); + target.on('error', e => { + console.error('debugger target error', e); + socket.destroy(); + target.destroy(); + }); + return; + } + catch (e) { + await sleep(500); + } + } + console.warn('debugger connect timed out'); + socket.destroy(); }) listenServerPort('SCRYPTED_DEBUG_PORT', SCRYPTED_DEBUG_PORT, debugServer); diff --git a/server/src/state.ts b/server/src/state.ts index 3245034f8..0486ebe22 100644 --- a/server/src/state.ts +++ b/server/src/state.ts @@ -73,13 +73,13 @@ export class ScryptedStateManager extends EventRegistry { updateDescriptor(device: PluginDevice) { for (const plugin of Object.values(this.scrypted.plugins)) { - plugin.remote?.updateDescriptor(device._id, device.state); + plugin.remote?.updateDeviceState(device._id, device.state); } } removeDevice(id: string) { for (const plugin of Object.values(this.scrypted.plugins)) { - plugin.remote?.updateDescriptor(id, undefined); + plugin.remote?.updateDeviceState(id, undefined); } this.notify(undefined, undefined, ScryptedInterface.ScryptedDevice, ScryptedInterfaceProperty.id, id, true);