more correct open/close code

kitcom/ts/ts_gen.tmpl

@@ -5,7 +5,7 @@ import {ParentIPC, ChildIPC} from 'kitten-ipc';
 {{- /*gotype: efprojects.com/kitten-ipc/kitcom/ts.tsGenData*/ -}}
 {{range $e := .Api.Endpoints}}
 export default class {{$e.Name}} {
-    private ipc: ParentIPC | ChildIPC;
+    protected ipc: ParentIPC | ChildIPC;
 
     constructor(ipc: ParentIPC | ChildIPC) {
         this.ipc = ipc;

lib/golang/lib.go

@@ -306,11 +306,11 @@ func (p *ParentIPC) Wait() (retErr error) {
 	case err := <-p.errCh:
 		retErr = fmt.Errorf("ipc internal error: %w", err)
 	case err := <-waitErrCh:
-		retErr = fmt.Errorf("cmd wait: %w", err)
+		if err != nil {
+			retErr = fmt.Errorf("cmd wait: %w", err)
+		}
 	}
 
-	killErr := p.cmd.Process.Kill()
-	retErr = mergeErr(retErr, killErr)
 	p.cleanup()
 
 	return

lib/ts/dist/lib.d.ts

@@ -1,5 +1,4 @@
 import * as net from 'node:net';
-import { QueuedEvent } from 'ts-events';
 declare enum MsgType {
     Call = 1,
     Response = 2
@@ -28,15 +27,19 @@ declare abstract class IPCCommon {
     protected conn: net.Socket | null;
     protected nextId: number;
     protected pendingCalls: Record<number, (result: CallResult) => void>;
-    protected errors: QueuedEvent<Error>;
+    protected closeRequested: boolean;
+    protected processingCalls: number;
+    protected onError?: (err: Error) => void;
+    protected onClose?: () => void;
     protected constructor(localApis: object[], socketPath: string);
     protected readConn(): void;
     protected processMsg(msg: Message): void;
-    protected handleCall(msg: CallMessage): void;
     protected sendMsg(msg: Message): void;
+    protected handleCall(msg: CallMessage): void;
     protected handleResponse(msg: ResponseMessage): void;
-    protected raiseErr(err: Error): void;
+    stop(): void;
     call(method: string, ...params: Vals): Promise<Vals>;
+    protected raiseErr(err: Error): void;
 }
 export declare class ParentIPC extends IPCCommon {
     private readonly cmdPath;

lib/ts/dist/lib.d.ts.map

@@ -1 +1 @@
-{"version":3,"file":"lib.d.ts","sourceRoot":"","sources":["../src/lib.ts"],"names":[],"mappings":"AAAA,OAAO,KAAK,GAAG,MAAM,UAAU,CAAC;AAOhC,OAAO,EAAC,WAAW,EAAC,MAAM,WAAW,CAAC;AAItC,aAAK,OAAO;IACR,IAAI,IAAI;IACR,QAAQ,IAAI;CACf;AAED,KAAK,IAAI,GAAG,GAAG,EAAE,CAAC;AAElB,UAAU,WAAW;IACjB,IAAI,EAAE,OAAO,CAAC,IAAI,CAAC;IACnB,EAAE,EAAE,MAAM,CAAC;IACX,MAAM,EAAE,MAAM,CAAC;IACf,MAAM,EAAE,IAAI,CAAC;CAChB;AAED,UAAU,eAAe;IACrB,IAAI,EAAE,OAAO,CAAC,QAAQ,CAAC;IACvB,EAAE,EAAE,MAAM,CAAC;IACX,MAAM,CAAC,EAAE,IAAI,CAAC;IACd,KAAK,CAAC,EAAE,MAAM,CAAC;CAClB;AAED,KAAK,OAAO,GAAG,WAAW,GAAG,eAAe,CAAC;AAE7C,UAAU,UAAU;IAChB,MAAM,EAAE,IAAI,CAAC;IACb,KAAK,EAAE,KAAK,GAAG,IAAI,CAAC;CACvB;AAED,uBAAe,SAAS;IACpB,SAAS,CAAC,SAAS,EAAE,MAAM,CAAC,MAAM,EAAE,GAAG,CAAC,CAAC;IACzC,SAAS,CAAC,UAAU,EAAE,MAAM,CAAC;IAC7B,SAAS,CAAC,IAAI,EAAE,GAAG,CAAC,MAAM,GAAG,IAAI,CAAQ;IACzC,SAAS,CAAC,MAAM,EAAE,MAAM,CAAK;IAC7B,SAAS,CAAC,YAAY,EAAE,MAAM,CAAC,MAAM,EAAE,CAAC,MAAM,EAAE,UAAU,KAAK,IAAI,CAAC,CAAM;IAC1E,SAAS,CAAC,MAAM,EAAE,WAAW,CAAC,KAAK,CAAC,CAAC;IAErC,SAAS,aAAa,SAAS,EAAE,MAAM,EAAE,EAAE,UAAU,EAAE,MAAM;IAU7D,SAAS,CAAC,QAAQ,IAAI,IAAI;IA0B1B,SAAS,CAAC,UAAU,CAAC,GAAG,EAAE,OAAO,GAAG,IAAI;IAWxC,SAAS,CAAC,UAAU,CAAC,GAAG,EAAE,WAAW,GAAG,IAAI;IA8C5C,SAAS,CAAC,OAAO,CAAC,GAAG,EAAE,OAAO,GAAG,IAAI;IAWrC,SAAS,CAAC,cAAc,CAAC,GAAG,EAAE,eAAe,GAAG,IAAI;IAapD,SAAS,CAAC,QAAQ,CAAC,GAAG,EAAE,KAAK,GAAG,IAAI;IAIpC,IAAI,CAAC,MAAM,EAAE,MAAM,EAAE,GAAG,MAAM,EAAE,IAAI,GAAG,OAAO,CAAC,IAAI,CAAC;CAmBvD;AAGD,qBAAa,SAAU,SAAQ,SAAS;IACpC,OAAO,CAAC,QAAQ,CAAC,OAAO,CAAS;IACjC,OAAO,CAAC,QAAQ,CAAC,OAAO,CAAW;IACnC,OAAO,CAAC,GAAG,CAA6B;IACxC,OAAO,CAAC,QAAQ,CAAC,QAAQ,CAAa;gBAE1B,OAAO,EAAE,MAAM,EAAE,OAAO,EAAE,MAAM,EAAE,EAAE,GAAG,SAAS,EAAE,MAAM,EAAE;IAahE,KAAK,IAAI,OAAO,CAAC,IAAI,CAAC;YAuBd,UAAU;IAmBlB,IAAI,IAAI,OAAO,CAAC,IAAI,CAAC;CAgB9B;AAGD,qBAAa,QAAS,SAAQ,SAAS;gBACvB,GAAG,SAAS,EAAE,MAAM,EAAE;IAI5B,KAAK,IAAI,OAAO,CAAC,IAAI,CAAC;IAUtB,IAAI,IAAI,OAAO,CAAC,IAAI,CAAC;CAK9B"}
+{"version":3,"file":"lib.d.ts","sourceRoot":"","sources":["../src/lib.ts"],"names":[],"mappings":"AAAA,OAAO,KAAK,GAAG,MAAM,UAAU,CAAC;AAUhC,aAAK,OAAO;IACR,IAAI,IAAI;IACR,QAAQ,IAAI;CACf;AAED,KAAK,IAAI,GAAG,GAAG,EAAE,CAAC;AAElB,UAAU,WAAW;IACjB,IAAI,EAAE,OAAO,CAAC,IAAI,CAAC;IACnB,EAAE,EAAE,MAAM,CAAC;IACX,MAAM,EAAE,MAAM,CAAC;IACf,MAAM,EAAE,IAAI,CAAC;CAChB;AAED,UAAU,eAAe;IACrB,IAAI,EAAE,OAAO,CAAC,QAAQ,CAAC;IACvB,EAAE,EAAE,MAAM,CAAC;IACX,MAAM,CAAC,EAAE,IAAI,CAAC;IACd,KAAK,CAAC,EAAE,MAAM,CAAC;CAClB;AAED,KAAK,OAAO,GAAG,WAAW,GAAG,eAAe,CAAC;AAE7C,UAAU,UAAU;IAChB,MAAM,EAAE,IAAI,CAAC;IACb,KAAK,EAAE,KAAK,GAAG,IAAI,CAAC;CACvB;AAED,uBAAe,SAAS;IACpB,SAAS,CAAC,SAAS,EAAE,MAAM,CAAC,MAAM,EAAE,GAAG,CAAC,CAAC;IACzC,SAAS,CAAC,UAAU,EAAE,MAAM,CAAC;IAC7B,SAAS,CAAC,IAAI,EAAE,GAAG,CAAC,MAAM,GAAG,IAAI,CAAQ;IACzC,SAAS,CAAC,MAAM,EAAE,MAAM,CAAK;IAC7B,SAAS,CAAC,YAAY,EAAE,MAAM,CAAC,MAAM,EAAE,CAAC,MAAM,EAAE,UAAU,KAAK,IAAI,CAAC,CAAM;IAC1E,SAAS,CAAC,cAAc,EAAE,OAAO,CAAS;IAC1C,SAAS,CAAC,eAAe,EAAE,MAAM,CAAK;IACtC,SAAS,CAAC,OAAO,CAAC,EAAE,CAAC,GAAG,EAAE,KAAK,KAAK,IAAI,CAAC;IACzC,SAAS,CAAC,OAAO,CAAC,EAAE,MAAM,IAAI,CAAC;IAE/B,SAAS,aAAa,SAAS,EAAE,MAAM,EAAE,EAAE,UAAU,EAAE,MAAM;IAS7D,SAAS,CAAC,QAAQ,IAAI,IAAI;IA4B1B,SAAS,CAAC,UAAU,CAAC,GAAG,EAAE,OAAO,GAAG,IAAI;IAWxC,SAAS,CAAC,OAAO,CAAC,GAAG,EAAE,OAAO,GAAG,IAAI;IAWrC,SAAS,CAAC,UAAU,CAAC,GAAG,EAAE,WAAW,GAAG,IAAI;IAuD5C,SAAS,CAAC,cAAc,CAAC,GAAG,EAAE,eAAe,GAAG,IAAI;IAapD,IAAI;IAYJ,IAAI,CAAC,MAAM,EAAE,MAAM,EAAE,GAAG,MAAM,EAAE,IAAI,GAAG,OAAO,CAAC,IAAI,CAAC;IAoBpD,SAAS,CAAC,QAAQ,CAAC,GAAG,EAAE,KAAK,GAAG,IAAI;CAGvC;AAGD,qBAAa,SAAU,SAAQ,SAAS;IACpC,OAAO,CAAC,QAAQ,CAAC,OAAO,CAAS;IACjC,OAAO,CAAC,QAAQ,CAAC,OAAO,CAAW;IACnC,OAAO,CAAC,GAAG,CAA6B;IACxC,OAAO,CAAC,QAAQ,CAAC,QAAQ,CAAa;gBAE1B,OAAO,EAAE,MAAM,EAAE,OAAO,EAAE,MAAM,EAAE,EAAE,GAAG,SAAS,EAAE,MAAM,EAAE;IAahE,KAAK,IAAI,OAAO,CAAC,IAAI,CAAC;YAuBd,UAAU;IAmBlB,IAAI,IAAI,OAAO,CAAC,IAAI,CAAC;CAgB9B;AAGD,qBAAa,QAAS,SAAQ,SAAS;gBACvB,GAAG,SAAS,EAAE,MAAM,EAAE;IAI5B,KAAK,IAAI,OAAO,CAAC,IAAI,CAAC;IAUtB,IAAI,IAAI,OAAO,CAAC,IAAI,CAAC;CAgB9B"}

lib/ts/src/lib.ts

@@ -5,7 +5,6 @@ import * as os from 'node:os';
 import * as path from 'node:path';
 import * as fs from 'node:fs';
 import * as util from 'node:util';
-import {QueuedEvent} from 'ts-events';
 
 const IPC_SOCKET_ARG = 'ipc-socket';
 
@@ -43,11 +42,13 @@ abstract class IPCCommon {
     protected conn: net.Socket | null = null;
     protected nextId: number = 0;
     protected pendingCalls: Record<number, (result: CallResult) => void> = {};
-    protected errors: QueuedEvent<Error>;
+    protected closeRequested: boolean = false;
+    protected processingCalls: number = 0;
+    protected onError?: (err: Error) => void;
+    protected onClose?: () => void;
 
     protected constructor(localApis: object[], socketPath: string) {
         this.socketPath = socketPath;
-        this.errors = new QueuedEvent();
 
         this.localApis = {};
         for (const localApi of localApis) {
@@ -63,6 +64,16 @@ abstract class IPCCommon {
             crlfDelay: Infinity,
         });
 
+        this.conn.on('error', (e) => {
+            this.raiseErr(e);
+        })
+
+        this.conn.on('close', (hadError: boolean) => {
+            if (hadError) {
+                this.raiseErr(new Error('connection closed due to error'));
+            }
+        })
+
         rl.on('line', (line) => {
             try {
                 const msg: Message = JSON.parse(line);
@@ -71,14 +82,6 @@ abstract class IPCCommon {
                 this.raiseErr(new Error(`${e}`));
             }
         });
-
-        rl.on('error', (e) => {
-            this.raiseErr(e);
-        });
-
-        rl.on('close', () => {
-            this.raiseErr(new Error('connection closed'));
-        });
     }
 
     protected processMsg(msg: Message): void {
@@ -92,6 +95,17 @@ abstract class IPCCommon {
         }
     }
 
+    protected sendMsg(msg: Message): void {
+        if (!this.conn) throw new Error('no connection');
+
+        try {
+            const data = JSON.stringify(msg) + '\n';
+            this.conn.write(data);
+        } catch (e) {
+            this.raiseErr(new Error(`send response for ${msg.id}: ${e}`));
+        }
+    }
+
     protected handleCall(msg: CallMessage): void {
         const [endpointName, methodName] = msg.method.split('.');
         if(!endpointName || !methodName) {
@@ -120,7 +134,9 @@ abstract class IPCCommon {
         }
 
         try {
-            const result = method.apply(this.localApis, msg.params);
+            this.processingCalls++;
+
+            const result = method.apply(endpoint, msg.params);
 
             if (result instanceof Promise) {
                 result
@@ -135,17 +151,12 @@ abstract class IPCCommon {
             }
         } catch (err) {
             this.sendMsg({type: MsgType.Response, id: msg.id, error: `${err}`});
+        } finally {
+            this.processingCalls--;
         }
-    }
 
-    protected sendMsg(msg: Message): void {
+        if(this.closeRequested) {
-        if (!this.conn) throw new Error('no connection');
+            if(this.onClose) this.onClose();
-
-        try {
-            const data = JSON.stringify(msg) + '\n';
-            this.conn.write(data);
-        } catch (e) {
-            this.raiseErr(new Error(`send response for ${msg.id}: ${e}`));
         }
     }
 
@@ -162,8 +173,15 @@ abstract class IPCCommon {
         callback({ result: msg.result || [], error: err });
     }
 
-    protected raiseErr(err: Error): void {
+    stop() {
-        this.errors.post(err);
+        if (this.closeRequested) {
+            throw new Error('close already requested');
+        }
+        if(!this.conn || this.conn.readyState === "closed") {
+            throw new Error('connection already closed');
+        }
+        this.closeRequested = true;
+        if(this.onClose) this.onClose();
     }
 
     call(method: string, ...params: Vals): Promise<Vals> {
@@ -185,6 +203,10 @@ abstract class IPCCommon {
             }
         });
     }
+
+    protected raiseErr(err: Error): void {
+        if(this.onError) this.onError(err);
+    }
 }
 
 
@@ -260,9 +282,9 @@ export class ParentIPC extends IPCCommon {
                     resolve();
                 }
             });
-            this.errors.attach(err => {
+            this.onError = (err) => {
                 reject(err);
-            })
+            }
         })
     }
 }
@@ -285,7 +307,15 @@ export class ChildIPC extends IPCCommon {
 
     async wait(): Promise<void> {
         return new Promise((resolve, reject) => {
-
+            this.onError = (err) => {
+                reject(err);
+            }
+            this.onClose = () => {
+                if (this.processingCalls === 0) {
+                    this.conn?.destroy();
+                    resolve();
+                }
+            }
         });
     }
 }