Merge pull request #9 from breejs/feat/server-sent-events

Feat/server sent events

Author: shadowgate15

.lintstagedrc.js

@@ -1,8 +1,7 @@
 module.exports = {
-  "*.md,!test/snapshots/**/*.md,!test/**/snapshots/**/*.md,!locales/README.md": [
-    filenames => filenames.map(filename => `remark ${filename} -qfo`),
-    'git add'
+  '*.md,!test/snapshots/**/*.md,!test/**/snapshots/**/*.md,!locales/README.md': [
+    (filenames) => filenames.map((filename) => `remark ${filename} -qfo`)
   ],
-  'package.json': ['fixpack', 'git add'],
-  '*.js': ['xo --fix', 'git add ']
+  'package.json': ['fixpack'],
+  '*.js': ['xo --fix']
 };

README.md

@@ -6,7 +6,7 @@
 [![styled with prettier](https://img.shields.io/badge/styled_with-prettier-ff69b4.svg)](https://github.com/prettier/prettier)
 [![made with lad](https://img.shields.io/badge/made_with-lad-95CC28.svg)](https://lad.js.org)
 
-> An API for Bree.
+> An API for [Bree][].
 
 ## Table of Contents
 
@@ -39,10 +39,11 @@ The API will start automatically when the Bree constructor is called.
 
 ## Options
 
-| Option |  Type  | Description                                                                                  |
-|:------:|:------:|----------------------------------------------------------------------------------------------|
-|  port  | Number | The port the API will listen on.                                                             |
-|   jwt  | Object | Configurations for JWT. Only option is `secret` which will be the secret used to verify JWT. |
+| Option   | Type     | Description                                                                                    |
+| :------: | :------: | ---------------------------------------------------------------------------------------------- |
+| port     | Number   | The port the API will listen on. Default: `62893`                                              |
+| jwt      | Object   | Configurations for JWT. Only option is `secret` which will be the secret used to verify JWT.   |
+| sse      | Object   | Configurations for SSE. See [koa-sse][] for list of options.                                   |
 
 ## API
 
@@ -56,3 +57,11 @@ Check out the [API Docs](https://documenter.getpostman.com/view/17142435/TzzDLbN
 [MIT](LICENSE) © [Nick Baugh](http://niftylettuce.com/)
 
 ##
+
+[npm]: https://www.npmjs.com/
+
+[yarn]: https://yarnpkg.com/
+
+[Bree]: https://jobscheduler.net/#/
+
+[koa-sse]: https://github.com/yklykl530/koa-sse

api.js

@@ -1,14 +1,16 @@
 // eslint-disable-next-line import/no-unassigned-import
 require('./config/env');
 
+const process = require('process');
+
 const API = require('@ladjs/api');
 const Graceful = require('@ladjs/graceful');
 const ip = require('ip');
 
 const logger = require('./helpers/logger');
 const apiConfig = require('./config/api');
 
-const api = new API(apiConfig);
+const api = new API(apiConfig());
 
 if (!module.parent) {
   const graceful = new Graceful({
@@ -23,7 +25,7 @@ if (!module.parent) {
       if (process.send) process.send('ready');
       const { port } = api.server.address();
       logger.info(
-        `Lad API server listening on ${port} (LAN: ${ip.address()}:${port})`
+        `Bree API server listening on ${port} (LAN: ${ip.address()}:${port})`
       );
     } catch (error) {
       logger.error(error);

app/controllers/api/v1/control.js

@@ -9,28 +9,53 @@ async function checkJobName(ctx, next) {
   return next();
 }
 
-async function start(ctx) {
+async function addJobNameToQuery(ctx, next) {
   const { jobName } = ctx.params;
 
-  ctx.bree.start(jobName);
+  ctx.query = { name: jobName };
+
+  return next();
+}
+
+async function start(ctx, next) {
+  const { jobName } = ctx.params;
+
+  await ctx.bree.start(jobName);
 
   ctx.body = {};
+
+  return next();
 }
 
-async function stop(ctx) {
+async function stop(ctx, next) {
   const { jobName } = ctx.params;
 
   await ctx.bree.stop(jobName);
 
   ctx.body = {};
+
+  return next();
 }
 
-async function run(ctx) {
+async function run(ctx, next) {
   const { jobName } = ctx.params;
 
-  ctx.bree.run(jobName);
+  await ctx.bree.run(jobName);
 
   ctx.body = {};
+
+  return next();
+}
+
+async function restart(ctx, next) {
+  const { jobName } = ctx.params;
+
+  await ctx.bree.stop(jobName);
+  await ctx.bree.start(jobName);
+
+  ctx.body = {};
+
+  return next();
 }
 
-module.exports = { checkJobName, start, stop, run };
+module.exports = { checkJobName, addJobNameToQuery, start, stop, run, restart };

app/controllers/api/v1/index.js

@@ -1,9 +1,10 @@
 const config = require('./config');
 const jobs = require('./jobs');
 const control = require('./control');
+const sse = require('./sse');
 
 const test = (ctx) => {
   ctx.body = { breeExists: Boolean(ctx.bree) };
 };
 
-module.exports = { config, test, jobs, control };
+module.exports = { config, test, jobs, control, sse };

app/controllers/api/v1/jobs.js

@@ -56,15 +56,13 @@ async function add(ctx) {
   let jobs;
 
   try {
-    jobs = bree.add(body.jobs);
+    jobs = await bree.add(body.jobs);
   } catch (err) {
     return ctx.throw(Boom.badData(err));
   }
 
   if (body.start) {
-    for (const job of jobs) {
-      bree.start(job.name);
-    }
+    await Promise.all(jobs.map((j) => bree.start(j.name)));
   }
 
   ctx.body = { jobs };

app/controllers/api/v1/sse.js

@@ -0,0 +1,47 @@
+async function connect(ctx) {
+  if (ctx.sse) {
+    // likely not the best way to do this
+    // TODO: fork koa-sse and move this into the ping interval
+    // runs every 60s
+    const interval = setInterval(() => {
+      const connected = ctx.sse.send({
+        event: 'status',
+        data: isActive(ctx)
+      });
+
+      console.log('connected');
+
+      // clear the interval if the client is disconnected
+      if (!connected) {
+        clearInterval(interval);
+      }
+    }, 60_000);
+    ctx.sse.send({
+      event: 'status',
+      data: isActive(ctx)
+    });
+
+    // send bree events over sse
+    for (const event of ['worker created', 'worker deleted']) {
+      ctx.bree.on(event, (name) => {
+        ctx.sse.send({ event, data: name });
+      });
+    }
+
+    ctx.sse.on('close', () => {
+      ctx.logger.error('SSE closed');
+
+      clearInterval(interval);
+    });
+  }
+}
+
+function isActive(ctx) {
+  return Boolean(
+    ctx.bree.workers.size > 0 ||
+      ctx.bree.timeouts.size > 0 ||
+      ctx.bree.intervals.size > 0
+  );
+}
+
+module.exports = { connect };

config/api.js

@@ -1,15 +1,54 @@
 const sharedConfig = require('@ladjs/shared-config');
 const jwt = require('koa-jwt');
+const sse = require('koa-sse-stream');
 
 const logger = require('../helpers/logger');
 const routes = require('../routes');
 const config = require('../config');
 
-module.exports = {
-  ...sharedConfig('API'),
-  routes: routes.api,
-  logger,
-  hookBeforeRoutes: (app) => {
-    app.use(jwt(config.jwt));
-  }
+module.exports = (opts = {}) => {
+  const sseMiddleware = sse({ ...config.sse, ...opts.sse });
+  const jwtMiddleware = jwt({
+    ...config.jwt,
+    ...opts.jwt,
+    getToken(ctx, _) {
+      // pull token off of url if it is the sse endpoint
+      if (ctx.url.indexOf('/v1/sse') === 0) {
+        const splitUrl = ctx.url.split('/');
+
+        if (splitUrl.length === 4) {
+          return splitUrl[3];
+        }
+      }
+
+      return null;
+    }
+  });
+
+  return {
+    ...sharedConfig('API'),
+    port: config.port,
+    ...opts,
+    routes: routes.api,
+    logger,
+    hookBeforeRoutes(app) {
+      app.use((ctx, next) => {
+        // return early if jwt is set to false
+        if (!opts.jwt && typeof opts.jwt === 'boolean') {
+          return next();
+        }
+
+        return jwtMiddleware(ctx, next);
+      });
+
+      app.use((ctx, next) => {
+        // only do this on sse route
+        if (ctx.url.indexOf('/v1/sse') === 0) {
+          return sseMiddleware(ctx, next);
+        }
+
+        return next();
+      });
+    }
+  };
 };

config/env.js

@@ -1,3 +1,4 @@
+const process = require('process');
 const path = require('path');
 
 const test = process.env.NODE_ENV === 'test';

config/index.js

@@ -20,7 +20,7 @@ const config = {
   appName: env.APP_NAME,
   appColor: env.APP_COLOR,
   twitter: env.TWITTER,
-  port: 4000,
+  port: 62_893,
 
   // build directory
   buildBase: 'build',
@@ -30,6 +30,12 @@ const config = {
     secret: env.JWT_SECRET
   },
 
+  // sse options
+  sse: {
+    maxClients: 10_000,
+    pingInterval: 60_000
+  },
+
   // store IP address
   // <https://github.com/ladjs/store-ip-address>
   storeIPAddress: {