fix: allow remote to abort local callback

achingbrain · May 26, 2024 · bc0233a · bc0233a
1 parent b23f18a
commit bc0233a
Show file tree

Hide file tree

Showing 10 changed files with 502 additions and 171 deletions.
diff --git a/packages/it-rpc/package.json b/packages/it-rpc/package.json
@@ -66,6 +66,7 @@
   },
   "devDependencies": {
     "aegir": "^42.2.11",
+    "delay": "^6.0.0",
     "it-all": "^3.0.0",
     "it-drain": "^3.0.7",
     "protons": "^7.5.0",

diff --git a/packages/it-rpc/src/codecs/1030-function.ts b/packages/it-rpc/src/codecs/1030-function.ts
@@ -23,7 +23,6 @@ const transformer: ValueCodec<(...args: any[]) => any> = {
       return new Promise<any>((resolve, reject) => {
         const id = decode(val)
         const scope = nanoid()
-
         const callbackInvocation: Invocation = {
           scope,
           result: pDefer(),
@@ -33,20 +32,10 @@ const transformer: ValueCodec<(...args: any[]) => any> = {
           abortControllers: [],
           abortSignals: []
         }
-
         invocation.children.set(scope, callbackInvocation)
+        args = args.map(val => codec.toValue(val, null, callbackInvocation))
 
-        pushable.push(RPCMessage.encode({
-          type: MessageType.invokeCallback,
-          message: InvokeCallbackMessage.encode({
-            scope,
-            parents: callbackInvocation.parents,
-            callback: id,
-            args: args.map(val => codec.toValue(val, null, callbackInvocation))
-          })
-        }))
-
-        const signal = anySignal(invocation.abortSignals)
+        const signal = anySignal(callbackInvocation.abortSignals)
         signal.addEventListener('abort', () => {
           pushable.push(RPCMessage.encode({
             type: MessageType.abortCallbackInvocation,
@@ -57,6 +46,16 @@ const transformer: ValueCodec<(...args: any[]) => any> = {
           }))
         })
 
+        pushable.push(RPCMessage.encode({
+          type: MessageType.invokeCallback,
+          message: InvokeCallbackMessage.encode({
+            scope,
+            parents: callbackInvocation.parents,
+            callback: id,
+            args
+          })
+        }))
+
         callbackInvocation.result.promise.then(result => {
           resolve(result)
         }, err => {

diff --git a/packages/it-rpc/src/errors.ts b/packages/it-rpc/src/errors.ts
@@ -34,9 +34,9 @@ export class DuplicateScopeError extends Error {
 }
 
 export class InvalidReturnTypeError extends Error {
-  constructor (message = 'Invalid response type') {
+  constructor (message = 'Invalid return type') {
     super(message)
-    this.name = 'InvalidResponseTypeError'
+    this.name = 'InvalidReturnTypeError'
   }
 }
 
@@ -54,6 +54,13 @@ export class MissingParentScopeError extends Error {
   }
 }
 
+export class MissingInvocationError extends Error {
+  constructor (message = 'Invocation not found') {
+    super(message)
+    this.name = 'MissingInvocationError'
+  }
+}
+
 export class MissingCallbackError extends Error {
   constructor (message = 'Callback not found') {
     super(message)