feat(Executor): Just-in-time connect to clients

Author: nokome

src/base/Client.ts

@@ -2,6 +2,7 @@ import { Node } from '@stencila/schema'
 import { Interface, Method, Manifest } from './Executor'
 import Request from './Request'
 import Response from './Response'
+import { Address } from './Transports'
 
 /**
  * A base client class which acts as a proxy to a remote `Executor`.
@@ -109,3 +110,7 @@ export default abstract class Client implements Interface {
     delete this.requests[response.id]
   }
 }
+
+export interface ClientType {
+  new (address: Address): Client
+}

src/base/Executor.test.ts

@@ -0,0 +1,183 @@
+import Executor, { Peer, Manifest, Method } 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'
+
+describe('Peer', () => {
+  test('capable: no capabilities', () => {
+    const peer = new Peer(
+      {
+        capabilities: {},
+        addresses: {}
+      },
+      []
+    )
+
+    expect(peer.capable(Method.compile, { foo: 'bar' })).toBe(false)
+    expect(peer.capable(Method.execute, {})).toBe(false)
+  })
+
+  test('capable: boolean capabilties', () => {
+    const peer = new Peer(
+      {
+        capabilities: {
+          decode: false,
+          compile: true,
+          execute: false
+        },
+        addresses: {}
+      },
+      []
+    )
+
+    expect(peer.capable(Method.decode, {})).toBe(false)
+    expect(peer.capable(Method.compile, {})).toBe(true)
+    expect(peer.capable(Method.execute, {})).toBe(false)
+  })
+
+  test('capable: schema object capabilties', () => {
+    const peer = new Peer(
+      {
+        capabilities: {
+          decode: {
+            required: ['content', 'format'],
+            properties: {
+              content: {
+                type: 'string'
+              },
+              format: {
+                enum: ['julia']
+              }
+            }
+          },
+          compile: {
+            type: 'object',
+            required: ['node'],
+            properties: {
+              node: {
+                type: 'object',
+                required: ['type', 'programmingLanguage'],
+                properties: {
+                  type: {
+                    enum: ['CodeChunk', 'CodeExpression']
+                  },
+                  programmingLanguage: {
+                    enum: ['python']
+                  }
+                }
+              }
+            }
+          }
+        },
+        addresses: {}
+      },
+      []
+    )
+
+    expect(peer.capable(Method.decode, {})).toBe(false)
+    expect(peer.capable(Method.decode, { content: 42 })).toBe(false)
+    expect(peer.capable(Method.decode, { content: '42' })).toBe(false)
+    expect(peer.capable(Method.decode, { content: '42', format: 'foo' })).toBe(
+      false
+    )
+    expect(
+      peer.capable(Method.decode, { content: '42', format: 'julia' })
+    ).toBe(true)
+
+    expect(peer.capable(Method.compile, {})).toBe(false)
+    expect(peer.capable(Method.compile, { node: 42 })).toBe(false)
+    expect(peer.capable(Method.compile, { node: { type: 'CodeChunk' } })).toBe(
+      false
+    )
+    expect(
+      peer.capable(Method.compile, {
+        node: { type: 'CodeChunk', programmingLanguage: 'javascript' }
+      })
+    ).toBe(false)
+    expect(
+      peer.capable(Method.compile, {
+        node: { type: 'CodeChunk', programmingLanguage: 'python' }
+      })
+    ).toBe(true)
+    expect(
+      peer.capable(Method.compile, {
+        node: { type: 'CodeExpression', programmingLanguage: 'python' }
+      })
+    ).toBe(true)
+  })
+
+  test('connect: no addresses', async () => {
+    const peer = new Peer(
+      {
+        capabilities: {},
+        addresses: {}
+      },
+      [DirectClient as ClientType]
+    )
+
+    expect(peer.connect()).toBe(false)
+  })
+
+  test('connect: no client types', async () => {
+    const peer = new Peer(
+      {
+        capabilities: {},
+        addresses: {}
+      },
+      []
+    )
+
+    expect(peer.connect()).toBe(false)
+  })
+
+  test('connect: no addresses match client types', async () => {
+    const peer = new Peer(
+      {
+        capabilities: {},
+        addresses: {
+          http: {
+            type: Transport.http
+          }
+        }
+      },
+      [DirectClient as ClientType, StdioClient as ClientType]
+    )
+
+    expect(peer.connect()).toBe(false)
+  })
+
+  test('connect: order of client types equals preference', async () => {
+    const directServer = new DirectServer()
+    const manifest: Manifest = {
+      capabilities: {},
+      addresses: {
+        direct: {
+          type: Transport.direct,
+          server: directServer
+        },
+        stdio: {
+          type: Transport.stdio,
+          command: 'echo'
+        }
+      }
+    }
+    const peer1 = new Peer(manifest, [
+      DirectClient as ClientType,
+      StdioClient as ClientType
+    ])
+    const peer2 = new Peer(manifest, [
+      StdioClient as ClientType,
+      DirectClient as ClientType
+    ])
+
+    expect(peer1.connect()).toBe(true)
+    // @ts-ignore
+    expect(peer1.client instanceof DirectClient).toBe(true)
+
+    expect(peer2.connect()).toBe(true)
+    // @ts-ignore
+    expect(peer2.client instanceof StdioClient).toBe(true)
+  })
+})

src/base/Executor.ts

@@ -1,7 +1,7 @@
 import { Node } from '@stencila/schema'
 import { JSONSchema7Definition } from 'json-schema'
 import Ajv from 'ajv'
-import Client from './Client'
+import Client, { ClientType } from './Client'
 import Server from './Server'
 import { Address, Transport } from './Transports'
 import { getLogger } from '@stencila/logga'
@@ -132,13 +132,22 @@ const ajv = new Ajv()
  *   a `WebSocketClient` to an executor running on
  *   a remote machine.
  */
-class Peer {
+export class Peer {
   /**
    * The manifest of the peer executor.
    */
   private manifest: Manifest
 
-  // private Clients: new ()
+  /**
+   * A list of classes, that extend `Client`, and are available
+   * to connect to peer executors.
+   *
+   * This property is used for dependency injection, rather than importing
+   * clients for all transports into this module when they may
+   * not be used (e.g. `StdioClient` in a browser hosted `Executor`).
+   * The order of this list, defines the preference for the transport.
+   */
+  private readonly clientTypes: ClientType[]
 
   /**
    * The client for the peer executor.
@@ -156,8 +165,9 @@ class Peer {
    */
   private validators: { [key: string]: Ajv.ValidateFunction } = {}
 
-  public constructor(manifest: Manifest) {
+  public constructor(manifest: Manifest, clientTypes: ClientType[]) {
     this.manifest = manifest
+    this.clientTypes = clientTypes
   }
 
   /**
@@ -186,25 +196,56 @@ class Peer {
     return validator(params) as boolean
   }
 
-  public connect(): Client {
-    let client: Client
-    // @ts-ignore
-    return client
+  /**
+   * Connect to the remote `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 {
+    for (const ClientType of this.clientTypes) {
+      // Get the transport for the client type
+      // There should be a better way to do this
+      const transportMap: { [key: string]: Transport } = {
+        DirectClient: Transport.direct,
+        StdioClient: Transport.stdio,
+        VsockClient: Transport.vsock,
+        TcpClient: Transport.tcp,
+        HttpClient: Transport.http,
+        WebsocketClient: Transport.ws
+      }
+      const transport = transportMap[ClientType.name]
+      if (transport === undefined)
+        throw new Error('Wooah! This should not happen!')
+
+      // See if the peer has an address the transport
+      const address = this.manifest.addresses[transport]
+      if (address !== undefined) {
+        this.client = new ClientType(address)
+        return true
+      }
+    }
+    // Unable to connect to the peer
+    return false
   }
 
   /**
    * Call a method of a remote `Executor`.
    *
+   * Ensures that there is a connection to the
+   * executor and then passes the request to it.
+   *
    * @param method The name of the method
    * @param params Values of parameters (i.e. arguments)
    */
   public async call<Type>(
     method: Method,
     params: { [key: string]: any } = {}
   ): Promise<Type> {
-    if (this.client === undefined) {
-      this.client = this.connect()
-    }
+    if (this.client === undefined)
+      throw new Error("WTF, no client! You shouldn't be calling this!")
     return this.client.call<Type>(method, params)
   }
 }
@@ -230,9 +271,11 @@ export default class Executor implements Interface {
    */
   private servers: Server[] = []
 
-  public constructor(peers: Manifest[] = []) {
-    // Create peers using the manifests provided
-    this.peers = peers.map(peer => new Peer(peer))
+  public constructor(
+    manifests: Manifest[] = [],
+    clientTypes: ClientType[] = []
+  ) {
+    this.peers = manifests.map(manifest => new Peer(manifest, clientTypes))
   }
 
   /**
@@ -355,7 +398,9 @@ export default class Executor implements Interface {
     // Attempt to delegate to a peer
     for (const peer of this.peers) {
       if (peer.capable(method, params)) {
-        return peer.call<Type>(method, params)
+        if (peer.connect()) {
+          return peer.call<Type>(method, params)
+        }
       }
     }
     // No peer has necessary capability so resort to fallback

src/base/Server.ts

@@ -113,12 +113,12 @@ export default abstract class Server {
   /**
    * Start the server
    */
-  public abstract start(): void
+  public start(): void {}
 
   /**
    * Stop the server
    */
-  public abstract stop(): void
+  public stop(): void {}
 
   /**
    * Run the server with graceful shutdown on `SIGINT` or `SIGTERM`

src/base/Transports.ts

@@ -1,11 +1,17 @@
 export enum Transport {
+  direct = 'direct',
   stdio = 'stdio',
   vsock = 'vsock',
   tcp = 'tcp',
   http = 'http',
   ws = 'ws'
 }
 
+export interface DirectAddress {
+  type: Transport.direct
+  server: any
+}
+
 export interface StdioAddress {
   type: Transport.stdio
   command: string
@@ -65,6 +71,7 @@ export interface WebsocketAddress extends HttpAddress {
 }
 
 export type Address =
+  | DirectAddress
   | StdioAddress
   | VsockAddress
   | TcpAddress

src/direct/DirectClient.ts

@@ -0,0 +1,19 @@
+import Client from '../base/Client'
+import Request from '../base/Request'
+import Server from '../base/Server'
+import { DirectAddress } from '../base/Transports'
+
+export default class DirectClient extends Client {
+  private server: Server
+
+  public constructor(address: Omit<DirectAddress, 'type'>) {
+    super()
+    this.server = address.server
+  }
+
+  protected send(request: Request): void {
+    // @ts-ignore server.receive is private
+    // eslint-disable-next-line @typescript-eslint/no-floating-promises
+    this.server.receive(request).then(response => this.receive(response))
+  }
+}