feat(HTTP, Websocket): Add initial HTTP and Websocket clients and servers

Author: nokome

package-lock.json

@@ -30,6 +30,10 @@
     "@stencila/logga": "^1.3.0",
     "@stencila/schema": "^0.29.0",
     "ajv": "^6.10.2",
+    "cross-fetch": "^3.0.4",
+    "fastify": "^2.10.0",
+    "fastify-websocket": "^0.6.0",
+    "isomorphic-ws": "^4.0.1",
     "length-prefixed-stream": "^2.0.0",
     "minimist": "^1.2.0"
   },

package.json

@@ -180,7 +180,9 @@ describe('Peer', () => {
         capabilities: {},
         addresses: {
           http: {
-            type: Transport.http
+            type: Transport.http,
+            host: '127.0.0.1',
+            port: 8000
           }
         }
       },

src/base/Executor.test.ts

@@ -309,16 +309,18 @@ export default class Executor implements Interface {
    * @param servers An array of `Server` instances that pass
    *                requests on to this executor
    */
-  public start(servers: Server[] = []): void {
+  public async start(servers: Server[] = []): Promise<void[]> {
     this.servers = servers
-    log.info('Starting servers')
-    this.servers.forEach(server => server.start())
+    log.info(
+      `Starting servers: ${this.servers.map(server => server.address.type)}`
+    )
+    return Promise.all(this.servers.map(server => server.start()))
   }
 
   /**
    * Stop servers for the executor.
    */
-  public stop(): void {
+  public async stop(): Promise<void> {
     log.info('Stopping servers')
     this.servers.forEach(server => server.stop())
   }
@@ -388,7 +390,7 @@ export default class Executor implements Interface {
    */
   protected async addresses(): Promise<Addresses> {
     return this.servers
-      .map(server => server.address())
+      .map(server => server.address)
       .reduce((prev, curr) => ({ ...prev, ...{ [curr.type]: curr } }), {})
   }
 

src/base/Executor.ts

@@ -14,13 +14,13 @@ export default abstract class Server {
    */
   private executor: Executor
 
+  public abstract readonly address: Address
+
   public constructor(executor?: Executor) {
     if (executor === undefined) executor = new Executor()
     this.executor = executor
   }
 
-  public abstract address(): Address
-
   /**
    * Handle a request
    *
@@ -54,8 +54,12 @@ export default abstract class Server {
     }
 
     try {
+      if (request === null) {
+        throw new Error(-32600, `Invalid request`)
+      }
+
       if (typeof request === 'string') {
-        // Parse JSON into an request
+        // Parse JSON into a request
         try {
           request = JSON.parse(request) as Request
         } catch (err) {
@@ -113,22 +117,24 @@ export default abstract class Server {
   /**
    * Start the server
    */
-  public start(): void {}
+  public async start(): Promise<void> {}
 
   /**
    * Stop the server
    */
-  public stop(): void {}
+  public async stop(): Promise<void> {}
 
   /**
    * Run the server with graceful shutdown on `SIGINT` or `SIGTERM`
    */
-  public run(): void {
+  public async run(): Promise<void> {
     if (process !== undefined) {
-      const stop = (): void => this.stop()
+      const stop = async (): Promise<void> => {
+        await this.stop()
+      }
       process.on('SIGINT', stop)
       process.on('SIGTERM', stop)
     }
-    this.start()
+    await this.start()
   }
 }

src/base/Server.ts

@@ -1,4 +1,4 @@
-import { tcpAddress } from './Transports'
+import { TcpAddress } from './Transports'
 
 describe('tcpAddress', () => {
   const defaults = {
@@ -7,21 +7,24 @@ describe('tcpAddress', () => {
   }
 
   test('parses a string with host and port', () => {
-    expect(tcpAddress('example.com:2010', defaults)).toEqual({
+    expect(new TcpAddress('example.com:2010', defaults)).toEqual({
+      type: 'tcp',
       host: 'example.com',
       port: 2010
     })
   })
 
   test('parses a number as the port', () => {
-    expect(tcpAddress(2020, defaults)).toEqual({
+    expect(new TcpAddress(2020, defaults)).toEqual({
+      type: 'tcp',
       host: '127.0.0.1',
       port: 2020
     })
   })
 
   test('parses a string with just port', () => {
-    expect(tcpAddress('2030', defaults)).toEqual({
+    expect(new TcpAddress('2030', defaults)).toEqual({
+      type: 'tcp',
       host: '127.0.0.1',
       port: 2030
     })

src/base/Transports.test.ts

@@ -25,51 +25,86 @@ export interface VsockAddress {
   cid?: number
 }
 
-export interface TcpAddress {
-  type: Transport.tcp | Transport.http | Transport.ws
-  host?: string
-  port?: number
-}
+export type TcpAddressInitializer =
+  | undefined
+  | number
+  | string
+  | { host: string; port: number }
+  | TcpAddress
+export class TcpAddress {
+  public readonly type: Transport.tcp | Transport.http | Transport.ws =
+    Transport.tcp
 
-export function tcpAddress(
-  address: undefined | string | number | Omit<TcpAddress, 'type'>,
-  defaults: {
-    host: string
-    port: number
-  }
-): Required<Omit<TcpAddress, 'type'>> {
-  if (address === undefined) {
-    return defaults
-  } else if (typeof address === 'string') {
-    const parts = address.split(':')
-    if (parts.length === 1) {
-      return {
-        host: defaults.host,
-        port: parseInt(parts[0])
-      }
-    } else if (parts.length >= 2) {
-      return {
-        host: parts[0],
-        port: parseInt(parts[1])
-      }
-    } else {
-      return defaults
+  public readonly host: string
+
+  public readonly port: number
+
+  public constructor(
+    address?: TcpAddressInitializer,
+    defaults: {
+      host: string
+      port: number
+    } = {
+      host: '127.0.0.1',
+      port: 2000
     }
-  } else if (typeof address === 'number') {
-    return { host: defaults.host, port: address }
-  } else {
-    const { host = defaults.host, port = defaults.port } = address
-    return { host, port }
+  ) {
+    const { host, port } = (function() {
+      if (address === undefined) {
+        return defaults
+      } else if (typeof address === 'string') {
+        const parts = address.split(':')
+        if (parts.length === 1) {
+          return {
+            host: defaults.host,
+            port: parseInt(parts[0])
+          }
+        } else if (parts.length >= 2) {
+          return {
+            host: parts[0],
+            port: parseInt(parts[1])
+          }
+        } else {
+          return defaults
+        }
+      } else if (typeof address === 'number') {
+        return { host: defaults.host, port: address }
+      } else {
+        const { host = defaults.host, port = defaults.port } = address
+        return { host, port }
+      }
+    })()
+    if (address instanceof TcpAddress) this.type = address.type
+    this.host = host
+    this.port = port
+  }
+
+  public toString(): string {
+    return `${this.type}://${this.host}:${this.port}`
   }
 }
 
-export interface HttpAddress extends TcpAddress {
-  type: Transport.http | Transport.ws
-  jwt?: string
+export class HttpAddress extends TcpAddress {
+  public readonly type: Transport.http | Transport.ws = Transport.http
+
+  public readonly jwt?: string
+
+  public constructor(
+    address?: TcpAddressInitializer,
+    defaults: {
+      host: string
+      port: number
+    } = {
+      host: '127.0.0.1',
+      port: 8000
+    }
+  ) {
+    super(address, defaults)
+  }
 }
 
-export interface WebsocketAddress extends HttpAddress {
-  type: Transport.ws
+export class WebSocketAddress extends HttpAddress {
+  public readonly type: Transport.ws = Transport.ws
 }
 
 export type Address =
@@ -78,4 +113,4 @@ export type Address =
   | VsockAddress
   | TcpAddress
   | HttpAddress
-  | WebsocketAddress
+  | WebSocketAddress

src/base/Transports.ts

@@ -1,16 +1,18 @@
-import { defaultHandler, getLogger, LogLevel, replaceHandlers } from '@stencila/logga';
-import fs from 'fs';
-import minimist from 'minimist';
-import { promisify } from 'util';
-import { ClientType } from './base/Client';
-import Executor from './base/Executor';
-import Server from './base/Server';
-import discoverStdio from './stdio/discover';
-import StdioClient from './stdio/StdioClient';
-import TcpServer from './tcp/TcpServer';
-
-const readFile = promisify(fs.readFile)
-const writeFile = promisify(fs.writeFile)
+import {
+  defaultHandler,
+  getLogger,
+  LogLevel,
+  replaceHandlers
+} from '@stencila/logga'
+import minimist from 'minimist'
+import { ClientType } from './base/Client'
+import Executor from './base/Executor'
+import Server from './base/Server'
+import discoverStdio from './stdio/discover'
+import StdioClient from './stdio/StdioClient'
+import HttpServer from './http/HttpServer'
+import TcpServer from './tcp/TcpServer'
+import WebSocketServer from './ws/WebSocketServer'
 
 const { _: args, ...options } = minimist(process.argv.slice(2))
 
@@ -56,51 +58,47 @@ const init = async () => {
 /**
  * Serve the executor
  */
-const serve = (executor: Executor) => {
+const serve = async (executor: Executor) => {
   // Add server classes based on supplied options
   const servers: Server[] = []
   if (options.tcp !== undefined) {
     servers.push(new TcpServer(executor, options.tcp))
   }
+  if (options.http !== undefined) {
+    servers.push(new HttpServer(executor, options.http))
+  }
+  if (options.ws !== undefined) {
+    servers.push(new WebSocketServer(executor, options.ws))
+  }
   if (servers.length === 0) {
     log.warn(
       'No servers specified in options (e.g. --tcp --stdio). Executor will not be accessible.'
     )
   }
-  executor.start(servers)
+  await executor.start(servers)
 }
 
 /**
  * Convert a document
  */
 const convert = async (executor: Executor): Promise<void> => {
   const input = args[1]
-  const output = args[2] || '-'
-
-  const content = await readFile(input, 'utf8')
+  const output = args[2] !== undefined ? args[2] : '-'
 
-  const decoded = await executor.decode(content)
-  const encoded = await executor.encode(decoded)
-
-  if (output === '-') console.log(encoded)
-  else await writeFile(output, encoded)
+  const decoded = await executor.decode(input)
+  await executor.encode(decoded, output)
 }
 
 /**
  * Execute a document
  */
 const execute = async (executor: Executor): Promise<void> => {
   const input = args[1]
-  const output = args[2] || input
-
-  const content = await readFile(input, 'utf8')
+  const output = args[2] !== undefined ? args[2] : input
 
-  const decoded = await executor.decode(content)
+  const decoded = await executor.decode(input)
   const executed = await executor.execute(decoded)
-  const encoded = await executor.encode(executed)
-
-  if (output === '-') console.log(encoded)
-  else await writeFile(output, encoded)
+  await executor.encode(executed, output)
 }
 
 // Run the main function and log any exceptions

src/cli.ts

@@ -6,7 +6,7 @@ import Executor from '../base/Executor'
 describe('DirectClient and DirectServer', () => {
   const executor = new Executor()
   const server = new DirectServer(executor)
-  const client = new DirectClient(server.address())
+  const client = new DirectClient(server.address)
 
   test('calling manifest', async () => {
     expect(await client.manifest()).toEqual(await executor.manifest())
@@ -27,6 +27,4 @@ describe('DirectClient and DirectServer', () => {
     // @ts-ignore
     expect(Object.keys(client.requests).length).toEqual(0)
   })
-
-  server.stop()
 })