:recycle: further refactor notifications into server methods

backend/lib/plugins/notification.js

@@ -1,4 +1,4 @@
-const { allStreams } = require('../utils')
+const _allStreams = {}
 
 const NotificationRoute = require('../routes/notification')
 
@@ -16,7 +16,7 @@ const ENDER = { event: 'end', data: '' }
  * @param {Stream} event
  * @returns {string}
  */
-const stringifyEvent = function (event) {
+const _stringifyEvent = function (event) {
     let str = ''
     const endl = '\r\n'
     for (const i in event) {
@@ -57,11 +57,11 @@ class Transformer extends Transform {
         if (this.event) {
             event.event = this.event
         }
-        this.push(stringifyEvent(event))
+        this.push(_stringifyEvent(event))
         callback()
     }
     _flush(callback) {
-        this.push(stringifyEvent(ENDER))
+        this.push(_stringifyEvent(ENDER))
         callback()
     }
 }
@@ -74,7 +74,7 @@ class Transformer extends Transform {
  * @param {Toolkit} h hapi common response toolkit
  * @param {object} streamOptions
  */
-const onEvent = (event, h, streamOptions) => {
+const _event = (event, h, streamOptions) => {
     let active
     if (event instanceof Stream.Readable) {
         if (event._readableState.objectMode) {
@@ -90,12 +90,30 @@ const onEvent = (event, h, streamOptions) => {
     }
 }
 
+/**
+ * Takes an open HTTP stream and writes
+ * a msg to it, then fires notification plugin's
+ * _event callback
+ * @param {object} msg you want to send
+ * @param {string} name <profileId>.<eventType>
+ * @param {boolean} shouldInitialize
+ * @param {Toolkit} h hapi common response toolkit
+ */
+const onNotify = (name, msg, h, shouldInitialize = false) => {
+    if (shouldInitialize) {
+        _allStreams[name] = new PassThrough({ objectMode: true })
+    }
+    if (!_allStreams[name]) return
+    _allStreams[name].write(msg)
+
+    return _event(_allStreams[name], h, { event: name })
+}
+
 module.exports = {
     name: 'notification-plugin',
     version: '1.0.0',
     register: async server => {
         await server.route(NotificationRoute)
-        server.decorate('toolkit', 'event', onEvent)
-        server.expose('streams', allStreams)
+        server.method('notify', onNotify)
     },
 }

backend/lib/routes/membership/active.js

@@ -6,8 +6,6 @@ const errorSchema = require('../../schemas/errors')
 const groupingSchema = require('../../schemas/groupings')
 const params = require('../../schemas/params')
 
-const { dispatchNotification } = require('../../utils')
-
 const pluginConfig = {
     handlerType: 'grouping',
     docs: {
@@ -83,14 +81,14 @@ module.exports = {
                 return g
             })
 
-            dispatchNotification(
+            request.server.methods.notify(
+                `${profileId}.stonk`,
                 {
                     name: 'MSHRM',
                     price: (500 + Math.floor(Math.random() * 100)).toString(),
                     order: null,
                     type: 'info',
                 },
-                `${profileId}.stonk`,
                 h,
             )
 

backend/lib/routes/notification/index.js

@@ -3,8 +3,6 @@ const apiSchema = require('../../schemas/api')
 const errorSchema = require('../../schemas/errors')
 const params = require('../../schemas/params')
 
-const { intializeStream } = require('../../utils')
-
 const pluginConfig = {
     handlerType: 'notifictaion',
     docs: {
@@ -32,7 +30,18 @@ module.exports = {
              * Write the initial stream
              * !: this must remain open for notifications to work
              */
-            return intializeStream(`${profile_id}.stonk`, h)
+            return request.server.methods.notify(
+                `${profile_id}.stonk`,
+                {
+                    profile_id,
+                    name: 'BDGRS',
+                    price: (500 + Math.floor(Math.random() * 100)).toString(),
+                    order: Math.floor(Math.random() * 2) === 1 ? 'BUY' : 'SELL',
+                    type: 'info',
+                },
+                h,
+                true,
+            )
         },
 
         /** Validate based on validators object */

backend/lib/utils.js

@@ -1,43 +0,0 @@
-const PassThrough = require('stream').PassThrough
-
-const allStreams = {}
-
-const activeStream = name => allStreams[name]
-
-/**
- * Takes an open HTTP stream and writes
- * a msg to it, then fires notification plugin's
- * onEvent callback
- * @param {object} msg you want to send
- * @param {string} name <profileId>.<eventType>
- * @param {object} req server request object
- * @param {Toolkit} h hapi common response toolkit
- */
-const dispatchNotification = (msg, name, h) => {
-    const stream = activeStream(name)
-    if (!stream) return
-    stream.write(msg)
-    return h.event(stream, h, { event: name })
-}
-
-const intializeStream = (name, h) => {
-    allStreams[name] = new PassThrough({ objectMode: true })
-
-    return dispatchNotification(
-        {
-            profile_id: name,
-            name: 'BDGRS',
-            price: (500 + Math.floor(Math.random() * 100)).toString(),
-            order: Math.floor(Math.random() * 2) === 1 ? 'BUY' : 'SELL',
-            type: 'info',
-        },
-        name,
-        h,
-    )
-}
-
-module.exports = {
-    dispatchNotification,
-    allStreams,
-    intializeStream,
-}