added setting and deleting messaging; new interception

Author: drinking-code

example/sub-process.js

@@ -2,4 +2,6 @@ import IPOS from '../lib/main.js'
 
 const sharedObject = await IPOS.new()
 console.log('sharedObject.exampleArray', sharedObject.exampleArray)
-// console.log(sharedObject.exampleArray[0], /*...sharedObject.exampleObject.entries()*/)
+setTimeout(() =>
+        console.log(sharedObject.exampleArray[0], /*...*/sharedObject.exampleObject/*.entries()*/),
+    10)

src/init-child.ts

@@ -1,27 +1,23 @@
-import IPOSMessaging from './messaging.js'
 import IPOS from './main.js'
 
 export default function initChild(this: IPOS) {
     let resolve: Function
     const promise = new Promise(res => resolve = res)
 
-    this.messaging = new IPOSMessaging(process)
-    this.messaging.listenForType('sync', message => {
+    this.messaging?.listenForType('sync', message => {
         if (!message.fields) return
-        Object.entries(
-            IPOS.deserialize(JSON.parse(message.fields))
-        )
+        Object.entries(message.fields)
             .map(([key, value]: [string, any]) => {
                 this.createStealthy(key, value)
             })
+        this.messaging?.send('sync_ok')
         resolve()
     })
     // register with parent process
-    this.messaging.send('register')
-    this.messaging.listenForType('update', message => {
-        if (!message.do || !message.on) return
-        this.get(message.on)[message.do](...(message.with ?? []))
-    })
+    this.messaging?.send('register')
+
+    if (this.messaging)
+    this.mountListeners(this.messaging)
 
     return promise
 }

src/intercept.ts

@@ -1,20 +1,29 @@
-import InterceptedArray from './intercept/array.js'
-
 export default function intercept(value: object, interceptCallback: (object: object, method: string, ...args: any) => void): object {
-    if (Array.isArray(value)) {
-        value = InterceptedArray.new(value, interceptCallback)
-    }
-    /*Object.getOwnPropertyNames(Object.getPrototypeOf(value))
-        .filter(methodName => !(methodName.startsWith('__') && methodName.endsWith('__')))
-        .forEach(methodName => {
-            if (!value[methodName]) return
-            const method = value[methodName]
-            value[methodName] = function (...args: any) {
-                // interception
-                console.log(methodName)
-                method.call(value, ...args)
+    const arrayMutatingMethods = ['copyWithin', 'fill', 'pop', 'push', 'reverse', 'shift', 'sort', 'splice', 'unshift']
+    const objectMutatingMethods: string[] = []
+    const mapMutatingMethods = ['clear', 'delete', 'set']
+    const setMutatingMethods = ['add', 'clear', 'delete']
+    const functionMutatingMethods: string[] = []
+    const mutatingMethods = new Map()
+    mutatingMethods.set(Array, arrayMutatingMethods)
+    mutatingMethods.set({}.constructor, objectMutatingMethods)
+    mutatingMethods.set(Map, mapMutatingMethods)
+    mutatingMethods.set(Set, setMutatingMethods)
+    mutatingMethods.set(Function, functionMutatingMethods)
+
+    if (!mutatingMethods.has(value.constructor))
+        return value
+    return new Proxy(value, {
+        get(target, name: string) {
+            if (Reflect.has(target, name) && mutatingMethods.get(value.constructor).includes(name)) {
+                const method = Reflect.get(target, name)
+                return (...args: any) => {
+                    interceptCallback(value, name, ...args)
+                    method.call(value, ...args)
+                }
+            } else {
+                return Reflect.get(target, name)
             }
-        })
-    */
-    return value
+        }
+    })
 }

src/intercept/array.ts

@@ -1,7 +1,6 @@
 import {ChildProcess} from 'child_process'
 import initChild from './init-child.js'
-import IPOSMessaging from './messaging.js'
-import {deserialize, serialize} from './serialize.js'
+import IPOSMessaging, {iposMessagingMessage, iposMessagingType} from './messaging.js'
 import intercept from './intercept.js'
 
 export default class IPOS {
@@ -28,6 +27,10 @@ export default class IPOS {
         this.fieldsReverseMap = new Map()
         this.processMessagingMap = new Map()
 
+        if (process.send) {
+            this.messaging = new IPOSMessaging(process)
+        }
+
         // proxy makes all "target.fields" available as "actual" fields
         this.proxy = new Proxy(this, {
             get(target, name: string) {
@@ -36,38 +39,90 @@ export default class IPOS {
                 } else if (target.fields.has(name)) {
                     return target.fields.get(name)
                 }
-            }
+            },
+            set(target, name: string, value: any): boolean {
+                if (Reflect.has(target, name)) {
+                    throw Error(`Cannot change inherent property \`${name}\``)
+                } else if (!target.fields.has(name)) {
+                    throw Error(`Cannot set unknown field \`${name}\`. Initialise a field with \`.create()\``)
+                } else {
+                    target.create(name, value)
+                    return true
+                }
+            },
         })
         return this.proxy
     }
 
+    protected mountListeners(messaging: IPOSMessaging) {
+        messaging.listenForType('update', (message) => this.performUpdate(message))
+        messaging.listenForType('set', (message) => this.performSet(message))
+        messaging.listenForType('delete', (message) => this.performDelete(message))
+    }
+
+    protected sendToAll(type: iposMessagingType, data?: {}) {
+        this.messaging?.send(type, data)
+        this.processMessagingMap.forEach(processMessaging => {
+            processMessaging.send(type, data)
+        })
+    }
+
+    /********************* GET **********************/
     public get(key: string): any {
         return this.fields.get(key)
     }
 
-    // todo: also accept and update non-object values
-    public create(key: string, value: object): void {
+    /******************** CREATE ********************/
+    public create(key: string, value: any): void {
         this.createStealthy(key, value)
-        // todo: send update message
+        this.sendToAll('set', {key, value})
     }
 
     protected createStealthy(key: string, value: object): void {
-        // console.log('create', key)
         if (typeof value === 'object')
             value = intercept(value, (object, method, ...args) =>
                 this.sendMethodCall(object, method, ...args)
             )
 
         this.fields.set(key, value)
         this.fieldsReverseMap.set(value, key)
-        // todo: send update message
     }
 
+    protected performSet(message: iposMessagingMessage) {
+        if (!message.key || !message.value) return
+        this.createStealthy(message.key, message.value)
+    }
+
+    /******************** UPDATE ********************/
+    protected performUpdate(message: iposMessagingMessage) {
+        if (!message.do || !message.on) return
+        this.get(message.on)[message.do](...(message.with ?? []))
+    }
+
+    private sendMethodCall(object: object, method: string, ...args: any) {
+        this.sendToAll('update', {
+            do: method,
+            on: this.fieldsReverseMap.get(object),
+            with: Array.from(args)
+        })
+    }
+
+    /******************** DELETE ********************/
     public delete(key: string): boolean {
+        this.sendToAll('delete', {key})
+        return this.deleteStealthy(key)
+    }
+
+    public deleteStealthy(key: string): boolean {
         return this.fields.delete(key)
-        // todo: send update message
     }
 
+    public performDelete(message: iposMessagingMessage) {
+        if (!message.key) return
+        return this.deleteStealthy(message.key)
+    }
+
+    /******************* PROCESS ********************/
     public addProcess(process: ChildProcess): Promise<void> {
         if (!process.send)
             throw new Error(`Process must have an ipc channel. Activate by passing "stdio: [<stdin>, <stdout>, <stderr>, 'ipc']" as an option.`)
@@ -79,37 +134,23 @@ export default class IPOS {
             if (registered) return
             registered = true
 
+            this.mountListeners(messaging)
             this.processMessagingMap.set(process, messaging)
             this.syncProcess(process)
-            resolve()
+                .then(() =>
+                    resolve()
+                )
         })
         return promise
     }
 
-    private syncProcess(process: ChildProcess) {
-        this.processMessagingMap.get(process)?.send('sync', {
-            fields: JSON.stringify(IPOS.serialize(this.fields))
-        })
-    }
-
-    private sendMethodCall(object: object, method: string, ...args: any) {
-        this.processMessagingMap.forEach(processMessaging => {
-            processMessaging.send('update', {
-                do: method,
-                on: this.fieldsReverseMap.get(object),
-                with: serialize(args)
-            })
+    private syncProcess(process: ChildProcess): Promise<void> {
+        let resolve: Function
+        const promise: Promise<void> = new Promise(res => resolve = res)
+        this.processMessagingMap.get(process)?.send('sync', {fields: this.fields})
+        this.processMessagingMap.get(process)?.listenOnceForType('sync_ok', () => {
+            resolve()
         })
-    }
-
-    /**
-     * Serializes types that "JSON.stringify()" doesn't properly handle
-     */
-    public static serialize(value: any): any | void {
-        return serialize(value)
-    }
-
-    public static deserialize(value: string | number | Array<any> | { $$iposType?: string, data: any }): any | void {
-        return deserialize(value)
+        return promise
     }
 }

src/main.ts

@@ -1,15 +1,21 @@
 import {ChildProcess} from 'child_process'
+import {deserialize, serialize} from './serialize.js'
 
-type iposMessagingType = 'ready' | 'register' | 'update' | 'sync'
+export type iposMessagingType = 'ready' | 'register' | 'update' | 'sync' | 'sync_ok' | 'set' | 'delete'
 type iposMessagingCallback = (message: iposMessagingMessage) => (any | void)
-type iposMessagingMessage = {
+export type iposMessagingMessage = {
     protocol: 'ipos',
     type: iposMessagingType,
+
     fields?: string,
+
     do?: string,
     on?: string,
     with?: Array<any>,
-}
+
+    key?: string,
+    value: any,
+} & { [k: string]: string }
 
 const mustHaveSendError = new Error(`Process must have a \`.send()\` method.`)
 
@@ -31,6 +37,11 @@ export default class IPOSMessaging {
                     return
                 }
 
+                for (const property in message) {
+                    if (!message.hasOwnProperty(property)) continue
+                    message[property] = deserialize(message[property])
+                }
+
                 if (this.listeners.has('any')) {
                     this.listeners.get('any')
                         ?.forEach(callback => callback(message))
@@ -56,7 +67,10 @@ export default class IPOSMessaging {
         this.process.send({
             protocol: 'ipos',
             type,
-            ...data
+            ...(Object.fromEntries(
+                Object.entries(data ?? {})
+                    .map(([key, value]) => [key, serialize(value)])
+            ))
         })
     }
 
@@ -66,6 +80,16 @@ export default class IPOSMessaging {
         this.listeners.set(type, callbacks)
     }
 
+    listenOnceForType(type: iposMessagingType | 'any', callback: iposMessagingCallback) {
+        let callbacks: Array<iposMessagingCallback> = this.listeners.get(type) ?? []
+        const onceCallback = (message: iposMessagingMessage) => {
+            delete callbacks[callbacks.indexOf(onceCallback)]
+            callback(message)
+        }
+        callbacks.push(onceCallback)
+        this.listeners.set(type, callbacks)
+    }
+
     listenForAll(callback: iposMessagingCallback) {
         this.listenForType('any', callback)
     }