feat(Executor): Walk node tree and delegate; allow for in-process executors

Author: nokome

src/base/Client.ts

@@ -5,10 +5,10 @@ import Response from './Response'
 import { Address } from './Transports'
 
 /**
- * A base client class which acts as a proxy to a remote `Executor`.
+ * A client to a remote, out of process, `Executor`.
  *
- * Implements aynchronous, proxy methods for `Executor` methods `compile`, `build`, `execute`, etc.
- * Those methods send JSON-RPC requests to a `Server` that is serving the remote `Executor`.
+ * Implements asynchronous, methods for `Executor` methods `compile`, `build`, `execute`, etc.
+ * which send JSON-RPC requests to a `Server` that is serving the remote `Executor`.
  */
 export default abstract class Client implements Interface {
   /**

src/base/Executor.test.ts

@@ -1,9 +1,10 @@
-import Executor, { Peer, Manifest, Method } from './Executor'
+import Executor, { Peer, Manifest, Method, Capabilities } from './Executor'
 import DirectServer from '../direct/DirectServer'
 import DirectClient from '../direct/DirectClient'
 import { ClientType } from './Client'
 import { Transport } from './Transports'
 import StdioClient from '../stdio/StdioClient'
+import { codeChunk, Node, isA, codeExpression, Article } from '@stencila/schema'
 
 describe('Peer', () => {
   test('capable: no capabilities', () => {
@@ -19,7 +20,7 @@ describe('Peer', () => {
     expect(peer.capable(Method.execute, {})).toBe(false)
   })
 
-  test('capable: boolean capabilties', () => {
+  test('capable: boolean capabilities', () => {
     const peer = new Peer(
       {
         capabilities: {
@@ -142,7 +143,7 @@ describe('Peer', () => {
       []
     )
 
-    const canDecode = (format: string) =>
+    const canDecode = (format: string): boolean =>
       peer.capable(Method.decode, { content: 'foo', format })
     expect(canDecode('julia')).toBe(true)
     expect(canDecode('haskell')).toBe(true)
@@ -215,10 +216,148 @@ describe('Peer', () => {
 
     expect(peer1.connect()).toBe(true)
     // @ts-ignore
-    expect(peer1.client instanceof DirectClient).toBe(true)
+    expect(peer1.interface instanceof DirectClient).toBe(true)
 
     expect(peer2.connect()).toBe(true)
     // @ts-ignore
-    expect(peer2.client instanceof StdioClient).toBe(true)
+    expect(peer2.interface instanceof StdioClient).toBe(true)
+  })
+})
+
+class DeepThought extends Executor {
+  public static readonly question: string =
+    'the answer to life the universe and everything'
+
+  public async capabilities(): Promise<Capabilities> {
+    return {
+      execute: {
+        properties: {
+          node: {
+            properties: {
+              type: {
+                const: 'CodeChunk'
+              },
+              text: {
+                const: DeepThought.question
+              }
+            }
+          }
+        }
+      }
+    }
+  }
+
+  public async execute(node: Node): Promise<Node> {
+    if (isA('CodeChunk', node) && node.text === DeepThought.question) {
+      return { ...node, outputs: [42] }
+    }
+    return node
+  }
+}
+
+class Calculator extends Executor {
+  public async capabilities(): Promise<Capabilities> {
+    return {
+      execute: {
+        properties: {
+          node: {
+            required: ['type', 'text'],
+            properties: {
+              type: {
+                const: 'CodeExpression'
+              },
+              text: {
+                type: 'string'
+              }
+            }
+          }
+        }
+      }
+    }
+  }
+
+  public async execute(node: Node): Promise<Node> {
+    if (isA('CodeExpression', node)) {
+      // eslint-disable-next-line no-eval
+      return { ...node, output: eval(node.text) }
+    }
+    return node
+  }
+}
+
+describe('Executor', () => {
+  test('peers: in-process', async () => {
+    const deepThought = new DeepThought()
+    const calculator = new Calculator()
+
+    const manifests: Manifest[] = [
+      {
+        executor: deepThought,
+        capabilities: await deepThought.capabilities()
+      },
+      {
+        executor: calculator,
+        capabilities: await calculator.capabilities()
+      }
+    ]
+    const executor = new Executor(manifests)
+
+    // Delegates executable nodes to peers
+
+    expect(await executor.execute(codeChunk(DeepThought.question))).toEqual({
+      type: 'CodeChunk',
+      text: DeepThought.question,
+      outputs: [42]
+    })
+
+    expect(await executor.execute(codeExpression('6 * 7'))).toEqual({
+      type: 'CodeExpression',
+      text: '6 * 7',
+      output: 42
+    })
+
+    expect(await executor.execute(codeExpression('2 * Math.PI'))).toEqual({
+      type: 'CodeExpression',
+      text: '2 * Math.PI',
+      output: 2 * Math.PI
+    })
+
+    // Walks node tree and delegates
+
+    const article = {
+      type: 'Article',
+      content: [
+        {
+          type: 'Paragraph',
+          content: [
+            'Four times twenty one is: ',
+            {
+              type: 'CodeExpression',
+              text: '4 * 21'
+            },
+            '.'
+          ]
+        },
+        {
+          type: 'QuoteBlock',
+          content: [
+            {
+              type: 'Paragraph',
+              content: ['The answer is:']
+            },
+            {
+              type: 'CodeChunk',
+              text: DeepThought.question
+            }
+          ]
+        }
+      ]
+    }
+    const executed = (await executor.execute(article)) as Article
+    expect(executed.type).toEqual('Article')
+    // @ts-ignore
+    expect(executed.content[0].content[1].output).toEqual(84)
+    // @ts-ignore
+    expect(executed.content[1].content[1].outputs[0]).toEqual(42)
   })
 })

src/base/Executor.ts

@@ -1,7 +1,7 @@
-import { Node } from '@stencila/schema'
+import { Node, isPrimitive, nodeType } from '@stencila/schema'
 import { JSONSchema7Definition } from 'json-schema'
 import Ajv from 'ajv'
-import Client, { ClientType } from './Client'
+import { ClientType } from './Client'
 import Server from './Server'
 import { Address, Transport } from './Transports'
 import { getLogger } from '@stencila/logga'
@@ -43,16 +43,22 @@ export interface Addresses {
  * etc
  */
 export interface Manifest {
+  /**
+   * The actual in-process `Executor`, or
+   * it's id i it is ou-of-process
+   */
+  executor?: Executor | string
+
   /**
    * The capabilities of the executor
    */
-  capabilities: Capabilities
+  capabilities?: Capabilities
 
   /**
    * The addresses of servers that can be used
    * to communicate with the executor
    */
-  addresses: Addresses
+  addresses?: Addresses
 }
 
 /**
@@ -150,12 +156,13 @@ export class Peer {
   public readonly clientTypes: ClientType[]
 
   /**
-   * The client for the peer executor.
+   * The interface to the peer executor.
    *
-   * The type of client e.g. `StdioClient` vs `WebSocketClient`
+   * May be an in-process `Executor` or a `Client` to an out-of-process
+   * `Executor`, in which case it's type e.g. `StdioClient` vs `WebSocketClient`
    * will depend upon the available transports in `manifest.transports`.
    */
-  private client?: Client
+  private interface?: Interface
 
   /**
    * Ajv validation functions for each method.
@@ -184,11 +191,16 @@ export class Peer {
   public capable(method: Method, params: { [key: string]: unknown }): boolean {
     let validators = this.validators[method]
     if (validators === undefined) {
+      // Peer does not have any capabilities defined
+      if (this.manifest.capabilities === undefined) return false
+
       let capabilities = this.manifest.capabilities[method]
-      // Peer does not have any capabilities defined for this method
+      // Peer does not have any capabilities for this method defined
       if (capabilities === undefined) return false
+
       // Peer defines capability as a single JSON Schema definition
       if (!Array.isArray(capabilities)) capabilities = [capabilities]
+
       // Compile JSON Schema definitions to validation functions
       validators = this.validators[method] = capabilities.map(schema =>
         ajv.compile(schema)
@@ -201,14 +213,22 @@ export class Peer {
   }
 
   /**
-   * Connect to the remote `Executor`.
+   * Connect to the `Executor`.
    *
    * Finds the first client type that the peer
    * executor supports.
    *
    * @returns A client instance or `undefined` if not able to connect
    */
   public connect(): boolean {
+    // If the executor is in-process, just use it directly
+    // eslint-disable-next-line @typescript-eslint/no-use-before-define
+    if (this.manifest.executor instanceof Executor) {
+      this.interface = this.manifest.executor
+      return true
+    }
+    // Connect to remote executor in order of preference of
+    // transports
     for (const ClientType of this.clientTypes) {
       // Get the transport for the client type
       // There should be a better way to do this
@@ -224,10 +244,11 @@ export class Peer {
       if (transport === undefined)
         throw new Error('Wooah! This should not happen!')
 
-      // See if the peer has an address the transport
+      // See if the peer has an address for the transport
+      if (this.manifest.addresses === undefined) return false
       const address = this.manifest.addresses[transport]
       if (address !== undefined) {
-        this.client = new ClientType(address)
+        this.interface = new ClientType(address)
         return true
       }
     }
@@ -248,9 +269,9 @@ export class Peer {
     method: Method,
     params: { [key: string]: any } = {}
   ): Promise<Type> {
-    if (this.client === undefined)
+    if (this.interface === undefined)
       throw new Error("WTF, no client! You shouldn't be calling this!")
-    return this.client.call<Type>(method, params)
+    return this.interface.call<Type>(method, params)
   }
 }
 
@@ -348,6 +369,7 @@ export default class Executor implements Interface {
     }
     for (const peer of this.peers) {
       const manifest = peer.manifest
+      if (manifest.capabilities === undefined) continue
       for (const [method, additional] of Object.entries(
         manifest.capabilities
       )) {
@@ -392,8 +414,23 @@ export default class Executor implements Interface {
     return this.delegate(Method.build, { node }, async () => node)
   }
 
+  /**
+   * Execute a `Node`.
+   *
+   * Walks the node tree and attempts to delegate
+   * execution of certain types of nodes (currently `CodeChunk` and `CodeExpression`).
+   *
+   * @param node The node to execute
+   */
   public async execute(node: Node): Promise<Node> {
-    return this.delegate(Method.execute, { node }, async () => node)
+    return this.walk(node, async node => {
+      switch (nodeType(node)) {
+        case 'CodeChunk':
+        case 'CodeExpression':
+          return this.delegate(Method.execute, { node }, async () => node)
+      }
+      return node
+    })
   }
 
   public async call(
@@ -414,6 +451,26 @@ export default class Executor implements Interface {
     }
   }
 
+  private async walk(
+    node: Node,
+    transformer: (node: Node) => Promise<Node>
+  ): Promise<Node> {
+    return walk(node)
+    async function walk(node: Node): Promise<Node> {
+      const transformed = await transformer(node)
+      if (transformed === undefined || isPrimitive(transformed))
+        return transformed
+      if (Array.isArray(transformed)) return Promise.all(transformed.map(walk))
+      return Object.entries(transformed).reduce(
+        async (prev, [key, child]) => ({
+          ...(await prev),
+          ...{ [key]: await walk(child) }
+        }),
+        Promise.resolve({})
+      )
+    }
+  }
+
   private async delegate<Type>(
     method: Method,
     params: { [key: string]: any },
@@ -431,7 +488,7 @@ export default class Executor implements Interface {
     }
 
     // No peer has necessary capability so resort to fallback
-    log.debug(`Not able to  ${JSON.stringify(params)} `)
+    log.debug(`Unable to delegate node: ${JSON.stringify(params)} `)
     return fallback()
   }
 }