:recycle: refactoring again to push to notifications without dispatchers

backend/lib/plugins/notification.js

@@ -1,11 +1,101 @@
 const NotificationRoute = require('../routes/notification')
-const { onEvent } = require('../services/notification')
+
+/** Heavily lifted from: https://github.com/mtharrison/susie/blob/master/lib/index.js */
+
+const Stream = require('stream')
+const PassThrough = Stream.PassThrough
+const Transform = Stream.Transform
+
+const ENDER = { event: 'end', data: '' }
+
+const allStreams = {}
+
+/**
+ * Stringify a stream
+ * ?: I don't really get what this is doing
+ * @param {Stream} event
+ * @returns {string}
+ */
+const stringifyEvent = function (event) {
+    let str = ''
+    const endl = '\r\n'
+    for (const i in event) {
+        let val = event[i]
+        if (val instanceof Buffer) {
+            val = val.toString()
+        }
+        if (typeof val === 'object') {
+            val = JSON.stringify(val)
+        }
+        str += i + ': ' + val + endl
+    }
+    str += endl
+    return str
+}
+
+/**
+ * Transform extension
+ * ?: I don't really get what this is doing
+ * @param {object} options
+ * @param {object} objectMode
+ */
+class Transformer extends Transform {
+    constructor(options, objectMode) {
+        super({ objectMode })
+        options = options || {}
+        this.counter = 1
+        this.event = options.event || null
+        this.generateId = options.generateId
+            ? options.generateId
+            : () => this.counter++
+    }
+    _transform(chunk, encoding, callback) {
+        const event = {
+            id: this.generateId(chunk),
+            data: chunk,
+        }
+        if (this.event) {
+            event.event = this.event
+        }
+        this.push(stringifyEvent(event))
+        callback()
+    }
+    _flush(callback) {
+        this.push(stringifyEvent(ENDER))
+        callback()
+    }
+}
+
+/**
+ * Callback to decorate server toolkit (h)
+ * !: Currently we only support ObjectMode streams
+ * ?: I don't really get what this is doing
+ * @param {Stream} event stream input
+ * @param {Toolkit} h hapi common response toolkit
+ * @param {object} streamOptions
+ */
+const onEvent = (event, h, streamOptions) => {
+    let active
+    if (event instanceof Stream.Readable) {
+        if (event._readableState.objectMode) {
+            active = new PassThrough()
+            const through = new Transformer(streamOptions, true)
+            through.pipe(active)
+            event.pipe(through)
+        }
+        return h
+            .response(active)
+            .header('content-type', 'text/event-stream')
+            .header('content-encoding', 'identity')
+    }
+}
 
 module.exports = {
     name: 'notification-plugin',
     version: '1.0.0',
-    register: async (server, options) => {
+    register: async server => {
         await server.route(NotificationRoute)
         server.decorate('toolkit', 'event', onEvent)
+        server.expose('streams', allStreams)
     },
 }

backend/lib/routes/membership/active.js

@@ -6,6 +6,8 @@ const errorSchema = require('../../schemas/errors')
 const groupingSchema = require('../../schemas/groupings')
 const params = require('../../schemas/params')
 
+const PassThrough = require('stream').PassThrough
+
 const pluginConfig = {
     handlerType: 'grouping',
     docs: {
@@ -34,6 +36,19 @@ const responseSchemas = {
     error: errorSchema.single,
 }
 
+const dispatch = (streamName, streams, h) => {
+    const stream = streams[streamName]
+    if (!stream || !streams) return
+    const msg = {
+        name: 'MSHRM',
+        price: (500 + Math.floor(Math.random() * 100)).toString(),
+        order: null,
+        type: 'info',
+    }
+    stream.write(msg)
+    h.event(stream, h, { event: streamName })
+}
+
 module.exports = {
     method: 'GET',
     path: '/{profile_id}',
@@ -81,6 +96,10 @@ module.exports = {
                 return g
             })
 
+            const allStreams =
+                request.server.plugins['notification-plugin']['streams']
+            dispatch(`${profileId}.stonk`, allStreams, h)
+
             try {
                 return {
                     ok: true,

backend/lib/routes/notification/index.js

@@ -3,7 +3,6 @@ const apiSchema = require('../../schemas/api')
 const errorSchema = require('../../schemas/errors')
 const params = require('../../schemas/params')
 
-const Stream = require('stream')
 const PassThrough = require('stream').PassThrough
 
 const pluginConfig = {
@@ -28,8 +27,11 @@ module.exports = {
         cors: true,
         handler: async (request, h) => {
             const { profile_id } = request.params
-            const input = new PassThrough({ objectMode: true })
-            const eventType = 'stonk'
+
+            const streamName = `${profile_id}.stonk`
+            const allStreams =
+                request.server.plugins['notification-plugin']['streams']
+            allStreams[streamName] = new PassThrough({ objectMode: true })
 
             const msg = {
                 profile_id,
@@ -40,15 +42,12 @@ module.exports = {
             }
 
             // Write to the input stream
-            setInterval(() => {
-                msg.order = Math.floor(Math.random() * 2) === 1 ? 'BUY' : 'SELL'
-                input.write(msg)
-            }, 5000)
+            msg.order = Math.floor(Math.random() * 2) === 1 ? 'BUY' : 'SELL'
+            allStreams[streamName].write(msg)
 
             // h.event() Added at plugin registration
             // h is the toolkit
-            const streamOptions = { event: `${profile_id}.${eventType}` }
-            return h.event(input, h, streamOptions)
+            return h.event(allStreams[streamName], h, { event: streamName })
         },
 
         /** Validate based on validators object */

backend/lib/services/notification.js

@@ -1,128 +0,0 @@
-/** Heavily lifted from: https://github.com/mtharrison/susie/blob/master/lib/index.js */
-
-const Stream = require('stream')
-const PassThrough = Stream.PassThrough
-const Transform = Stream.Transform
-
-const ENDER = { event: 'end', data: '' }
-
-/**
- * Stringify a stream
- * ?: I don't really get what this is doing
- * @param {Stream} event
- * @returns {string}
- */
-const stringifyEvent = function (event) {
-    let str = ''
-    const endl = '\r\n'
-    for (const i in event) {
-        let val = event[i]
-        if (val instanceof Buffer) {
-            val = val.toString()
-        }
-        if (typeof val === 'object') {
-            val = JSON.stringify(val)
-        }
-        str += i + ': ' + val + endl
-    }
-    str += endl
-    return str
-}
-
-/**
- * Transform extension
- * ?: I don't really get what this is doing
- * @param {object} options
- * @param {object} objectMode
- */
-class Transformer extends Transform {
-    constructor(options, objectMode) {
-        super({ objectMode })
-        options = options || {}
-        this.counter = 1
-        this.event = options.event || null
-        this.generateId = options.generateId
-            ? options.generateId
-            : () => {
-                  return this.counter++
-              }
-    }
-    _transform(chunk, encoding, callback) {
-        const event = {
-            id: this.generateId(chunk),
-            data: chunk,
-        }
-        if (this.event) {
-            event.event = this.event
-        }
-        this.push(stringifyEvent(event))
-        callback()
-    }
-    _flush(callback) {
-        this.push(stringifyEvent(ENDER))
-        callback()
-    }
-}
-
-/**
- * Take an event stream and write content to another stream
- * ?: Save this for future extension
- * @param {Stream} event stream input
- * @param {Stream} stream to write to
- */
-const writeEvent = function (event, stream) {
-    if (event) {
-        stream.write(stringifyEvent(event))
-    } else {
-        // closing time
-        stream.write(stringifyEvent(ENDER))
-        stream.end()
-    }
-}
-
-/**
- * Callback to decorate server toolkit (h)
- * !: Currently we only support ObjectMode streams
- * ?: I don't really get what this is doing
- * @param {Stream} event stream input
- * @param {Toolkit} h hapi common response toolkit
- * @param {object} streamOptions
- */
-const onEvent = (event, h, streamOptions) => {
-    // const state = h.request.plugins.notifications = h.request.plugins.notifications || {}
-    let active
-    if (event instanceof Stream.Readable) {
-        if (event._readableState.objectMode) {
-            const through = new Transformer(streamOptions, true)
-            active = new PassThrough()
-            through.pipe(active)
-            event.pipe(through)
-        }
-        // else {
-        //     stream = new Transformer(streamOptions, false)
-        //     event.pipe(stream)
-        // }
-        console.log('streamOptions :', streamOptions)
-        return h
-            .response(active)
-            .header('content-type', 'text/event-stream')
-            .header('content-encoding', 'identity')
-    }
-    // Uncomment to do stream state stuff
-    // handle a first object arg
-    // if (!state.stream) {
-    //     active = new PassThrough()
-    //     state.stream = active
-    //     state.mode = 'object'
-    //     const response = h.response(active)
-    //         .header('content-type', 'text/event-stream')
-    //         .header('content-encoding', 'identity')
-    //     writeEvent(event, active)
-    //     return response
-    // }
-    // already have an object stream flowing, just write next event
-    // active = state.stream
-    // internals.writeEvent(event, active)
-}
-
-module.exports = { onEvent }