Skip to content

Commit aed6e0c

Browse files
committed
server: wip cluster mode load balancing
1 parent 432c178 commit aed6e0c

15 files changed

+479
-401
lines changed

sdk/types/scrypted_python/scrypted_sdk/__init__.py

+7
Original file line numberDiff line numberDiff line change
@@ -62,6 +62,13 @@ def sdk_init2(scryptedStatic: ScryptedStatic):
6262
systemManager = sdk.systemManager
6363
deviceManager = sdk.deviceManager
6464
mediaManager = sdk.mediaManager
65+
async def initDescriptors():
66+
global api
67+
try:
68+
await api.setScryptedInterfaceDescriptors(TYPES_VERSION, ScryptedInterfaceDescriptors)
69+
except:
70+
pass
71+
asyncio.ensure_future(initDescriptors())
6572
if hasattr(sdk, 'clusterManager'):
6673
clusterManager = sdk.clusterManager
6774
zip = sdk.zip

sdk/types/scrypted_python/scrypted_sdk/types.py

+3
Original file line numberDiff line numberDiff line change
@@ -950,6 +950,9 @@ class TamperState(TypedDict):
950950
pass
951951

952952

953+
TYPES_VERSION = "0.3.77"
954+
955+
953956
class AirPurifier:
954957

955958
airPurifierState: AirPurifierState

sdk/types/src/build.ts

+4-1
Original file line numberDiff line numberDiff line change
@@ -151,7 +151,10 @@ function selfSignature(method: any) {
151151

152152
const enums = schema.children?.filter((child) => child.kind === ReflectionKind.Enum) ?? [];
153153
const interfaces = schema.children?.filter((child: any) => Object.values(ScryptedInterface).includes(child.name)) ?? [];
154-
let python = '';
154+
let python = `
155+
TYPES_VERSION = "${typesVersion}"
156+
157+
`;
155158

156159
for (const iface of ['Logger', 'DeviceManager', 'SystemManager', 'MediaManager', 'EndpointManager', 'ClusterManager']) {
157160
const child = schema.children?.find((child: any) => child.name === iface);

server/package-lock.json

+5-4
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

server/package.json

+1-1
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@
55
"dependencies": {
66
"@scrypted/ffmpeg-static": "^6.1.0-build3",
77
"@scrypted/node-pty": "^1.0.22",
8-
"@scrypted/types": "^0.3.69",
8+
"@scrypted/types": "^0.3.77",
99
"adm-zip": "^0.5.16",
1010
"body-parser": "^1.20.3",
1111
"cookie-parser": "^1.4.7",

server/python/plugin_remote.py

+18
Original file line numberDiff line numberDiff line change
@@ -202,6 +202,21 @@ def notifyEventDetails(self, id: str, eventDetails: scrypted_python.scrypted_sdk
202202

203203
return True
204204

205+
class ClusterManager(scrypted_python.scrypted_sdk.types.ClusterManager):
206+
def __init__(self, api: Any):
207+
self.api = api
208+
self.clusterService = None
209+
210+
def getClusterMode(self) -> Any | Any:
211+
return os.getenv("SCRYPTED_CLUSTER_MODE", None)
212+
213+
def getClusterWorkerId(self) -> str:
214+
return os.getenv("SCRYPTED_CLUSTER_WORKER_ID", None)
215+
216+
async def getClusterWorkers(self) -> Mapping[str, scrypted_python.scrypted_sdk.types.ClusterWorker]:
217+
self.clusterService = self.clusterService or asyncio.ensure_future(self.api.getComponent("cluster-fork"))
218+
cs = await self.clusterService
219+
return await cs.getClusterWorkers()
205220

206221
class SystemManager(scrypted_python.scrypted_sdk.types.SystemManager):
207222
def __init__(
@@ -554,6 +569,7 @@ def __init__(
554569
self.systemState: Mapping[str, Mapping[str, SystemDeviceState]] = {}
555570
self.nativeIds: Mapping[str, DeviceStorage] = {}
556571
self.mediaManager: MediaManager
572+
self.clusterManager: ClusterManager
557573
self.consoles: Mapping[str, Future[Tuple[StreamReader, StreamWriter]]] = {}
558574
self.peer = clusterSetup.peer
559575
self.clusterSetup = clusterSetup
@@ -768,11 +784,13 @@ def read_requirements(filename: str) -> str:
768784
self.systemManager = SystemManager(self.api, self.systemState)
769785
self.deviceManager = DeviceManager(self.nativeIds, self.systemManager)
770786
self.mediaManager = MediaManager(await self.api.getMediaManager())
787+
self.clusterManager = ClusterManager(self.api)
771788

772789
try:
773790
sdk.systemManager = self.systemManager
774791
sdk.deviceManager = self.deviceManager
775792
sdk.mediaManager = self.mediaManager
793+
sdk.clusterManager = self.clusterManager
776794
sdk.remote = self
777795
sdk.api = self.api
778796
sdk.zip = zip

server/src/plugin/device.ts

+261
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,261 @@
1+
import { Device, DeviceManager, DeviceManifest, DeviceState, Logger, ScryptedNativeId, WritableDeviceState } from '@scrypted/types';
2+
import { RpcPeer } from '../rpc';
3+
import { PluginAPI, PluginLogger } from './plugin-api';
4+
import { checkProperty } from './plugin-state-check';
5+
import { SystemManagerImpl } from './system';
6+
7+
class DeviceLogger implements Logger {
8+
nativeId: ScryptedNativeId;
9+
api: PluginAPI;
10+
logger: Promise<PluginLogger>;
11+
12+
constructor(api: PluginAPI, nativeId: ScryptedNativeId, public console: any) {
13+
this.api = api;
14+
this.nativeId = nativeId;
15+
}
16+
17+
async ensureLogger(): Promise<PluginLogger> {
18+
if (!this.logger)
19+
this.logger = this.api.getLogger(this.nativeId);
20+
return await this.logger;
21+
}
22+
23+
async log(level: string, message: string) {
24+
(await this.ensureLogger()).log(level, message);
25+
}
26+
27+
a(msg: string): void {
28+
this.log('a', msg);
29+
}
30+
async clear() {
31+
(await this.ensureLogger()).clear();
32+
}
33+
async clearAlert(msg: string) {
34+
(await this.ensureLogger()).clearAlert(msg);
35+
}
36+
async clearAlerts() {
37+
(await this.ensureLogger()).clearAlerts();
38+
}
39+
d(msg: string): void {
40+
this.log('d', msg);
41+
}
42+
e(msg: string): void {
43+
this.log('e', msg);
44+
}
45+
i(msg: string): void {
46+
this.log('i', msg);
47+
}
48+
v(msg: string): void {
49+
this.log('v', msg);
50+
}
51+
w(msg: string): void {
52+
this.log('w', msg);
53+
}
54+
}
55+
56+
export class DeviceStateProxyHandler implements ProxyHandler<any> {
57+
constructor(public deviceManager: DeviceManagerImpl, public id: string,
58+
public setState: (property: string, value: any) => Promise<void>) {
59+
}
60+
61+
get?(target: any, p: PropertyKey, receiver: any) {
62+
if (p === 'id')
63+
return this.id;
64+
if (p === RpcPeer.PROPERTY_PROXY_PROPERTIES)
65+
return { id: this.id }
66+
if (p === 'setState')
67+
return this.setState;
68+
return this.deviceManager.systemManager.state[this.id][p as string]?.value;
69+
}
70+
71+
set?(target: any, p: PropertyKey, value: any, receiver: any) {
72+
checkProperty(p.toString(), value);
73+
this.deviceManager.systemManager.state[this.id][p as string] = {
74+
value,
75+
};
76+
this.setState(p.toString(), value);
77+
return true;
78+
}
79+
}
80+
81+
interface DeviceManagerDevice {
82+
id: string;
83+
storage: { [key: string]: any };
84+
}
85+
86+
export class DeviceManagerImpl implements DeviceManager {
87+
api: PluginAPI;
88+
nativeIds = new Map<string, DeviceManagerDevice>();
89+
deviceStorage = new Map<string, StorageImpl>();
90+
mixinStorage = new Map<string, Map<string, StorageImpl>>();
91+
92+
constructor(public systemManager: SystemManagerImpl,
93+
public getDeviceConsole: (nativeId?: ScryptedNativeId) => Console,
94+
public getMixinConsole: (mixinId: string, nativeId?: ScryptedNativeId) => Console) {
95+
}
96+
97+
async requestRestart() {
98+
return this.api.requestRestart();
99+
}
100+
101+
getDeviceLogger(nativeId?: ScryptedNativeId): Logger {
102+
return new DeviceLogger(this.api, nativeId, this.getDeviceConsole?.(nativeId) || console);
103+
}
104+
105+
getDeviceState(nativeId?: any): DeviceState {
106+
const handler = new DeviceStateProxyHandler(this, this.nativeIds.get(nativeId).id,
107+
(property, value) => this.api.setState(nativeId, property, value));
108+
return new Proxy(handler, handler);
109+
}
110+
111+
createDeviceState(id: string, setState: (property: string, value: any) => Promise<void>): WritableDeviceState {
112+
const handler = new DeviceStateProxyHandler(this, id, setState);
113+
return new Proxy(handler, handler);
114+
}
115+
116+
getDeviceStorage(nativeId?: any): StorageImpl {
117+
let ret = this.deviceStorage.get(nativeId);
118+
if (!ret) {
119+
ret = new StorageImpl(this, nativeId);
120+
this.deviceStorage.set(nativeId, ret);
121+
}
122+
return ret;
123+
}
124+
getMixinStorage(id: string, nativeId?: ScryptedNativeId) {
125+
let ms = this.mixinStorage.get(nativeId);
126+
if (!ms) {
127+
ms = new Map();
128+
this.mixinStorage.set(nativeId, ms);
129+
}
130+
let ret = ms.get(id);
131+
if (!ret) {
132+
ret = new StorageImpl(this, nativeId, `mixin:${id}:`);
133+
ms.set(id, ret);
134+
}
135+
return ret;
136+
}
137+
pruneMixinStorage() {
138+
for (const nativeId of this.nativeIds.keys()) {
139+
const storage = this.nativeIds.get(nativeId).storage;
140+
for (const key of Object.keys(storage)) {
141+
if (!key.startsWith('mixin:'))
142+
continue;
143+
const [, id,] = key.split(':');
144+
// there's no rush to persist this, it will happen automatically on the plugin
145+
// persisting something at some point.
146+
// the key itself is unreachable due to the device no longer existing.
147+
if (id && !this.systemManager.state[id])
148+
delete storage[key];
149+
}
150+
}
151+
}
152+
async onMixinEvent(id: string, nativeId: ScryptedNativeId, eventInterface: string, eventData: any) {
153+
return this.api.onMixinEvent(id, nativeId, eventInterface, eventData);
154+
}
155+
getNativeIds(): string[] {
156+
return Array.from(this.nativeIds.keys());
157+
}
158+
async onDeviceDiscovered(device: Device) {
159+
return this.api.onDeviceDiscovered(device);
160+
}
161+
async onDeviceRemoved(nativeId: string) {
162+
return this.api.onDeviceRemoved(nativeId);
163+
}
164+
async onDeviceEvent(nativeId: any, eventInterface: any, eventData?: any) {
165+
return this.api.onDeviceEvent(nativeId, eventInterface, eventData);
166+
}
167+
async onDevicesChanged(devices: DeviceManifest) {
168+
return this.api.onDevicesChanged(devices);
169+
}
170+
}
171+
172+
173+
function toStorageString(value: any) {
174+
if (value === null)
175+
return 'null';
176+
if (value === undefined)
177+
return 'undefined';
178+
179+
return value.toString();
180+
}
181+
182+
export class StorageImpl implements Storage {
183+
api: PluginAPI;
184+
[name: string]: any;
185+
186+
private static allowedMethods = [
187+
'length',
188+
'clear',
189+
'getItem',
190+
'setItem',
191+
'key',
192+
'removeItem',
193+
];
194+
private static indexedHandler: ProxyHandler<StorageImpl> = {
195+
get(target, property) {
196+
const keyString = property.toString();
197+
if (StorageImpl.allowedMethods.includes(keyString)) {
198+
const f = target[keyString];
199+
if (keyString === 'length')
200+
return f;
201+
return f.bind(target);
202+
}
203+
return target.getItem(toStorageString(property));
204+
},
205+
set(target, property, value): boolean {
206+
target.setItem(toStorageString(property), value);
207+
return true;
208+
}
209+
};
210+
211+
constructor(public deviceManager: DeviceManagerImpl, public nativeId: ScryptedNativeId, public prefix?: string) {
212+
this.deviceManager = deviceManager;
213+
this.api = deviceManager.api;
214+
this.nativeId = nativeId;
215+
if (!this.prefix)
216+
this.prefix = '';
217+
218+
return new Proxy(this, StorageImpl.indexedHandler);
219+
}
220+
221+
get storage(): { [key: string]: any } {
222+
return this.deviceManager.nativeIds.get(this.nativeId).storage;
223+
}
224+
225+
get length(): number {
226+
return Object.keys(this.storage).filter(key => key.startsWith(this.prefix)).length;
227+
}
228+
229+
clear(): void {
230+
if (!this.prefix) {
231+
this.deviceManager.nativeIds.get(this.nativeId).storage = {};
232+
}
233+
else {
234+
const storage = this.storage;
235+
Object.keys(this.storage).filter(key => key.startsWith(this.prefix)).forEach(key => delete storage[key]);
236+
}
237+
this.api.setStorage(this.nativeId, this.storage);
238+
}
239+
240+
getItem(key: string): string {
241+
return this.storage[this.prefix + key];
242+
}
243+
key(index: number): string {
244+
if (!this.prefix) {
245+
return Object.keys(this.storage)[index];
246+
}
247+
return Object.keys(this.storage).filter(key => key.startsWith(this.prefix))[index].substring(this.prefix.length);
248+
}
249+
removeItem(key: string): void {
250+
delete this.storage[this.prefix + key];
251+
this.api.setStorage(this.nativeId, this.storage);
252+
}
253+
setItem(key: string, value: string): void {
254+
key = toStorageString(key);
255+
value = toStorageString(value);
256+
if (this.storage[this.prefix + key] === value)
257+
return;
258+
this.storage[this.prefix + key] = value;
259+
this.api.setStorage(this.nativeId, this.storage);
260+
}
261+
}

0 commit comments

Comments
 (0)