Data real-time feedback technology

In fact, I don't know how to start this title , This is a scene like this , In the development of background management system , Especially when real-time monitoring system , It is often necessary to show the continuous update and change of data . The common technology is polling , Or use websocket Long connection real-time communication . We know webpack There is a hot update function in debugging mode , It is achieved through the server data push function . It's called Server-Sent Events（SSE）.

For specific principles and usage, please refer to Ruan Yifeng The article

http://www.ruanyifeng.com/blog/2017/05/server-sent_events.html

This article will combine Node.js、SSE、Koa、Pm2、Rxjs Technology to achieve an elegant development technology of real-time data feedback . stay Node.js End , We can install one NPM package ：http-event-stream To do server push . The source code of this package is very simple , Readers can have a look for themselves , It's not difficult to write by yourself . up to now , Start from the server to the browser , Real time data update is very simple , But we still need the final development experience , That is, how to make the data on the server real-time “ push ” To with http-event-stream In your request ？

A simple way , That is, when you get from the client SSE On request , Start a timer , Get the data in the database or memory in the timer , And then send it to the client . Before writing this code , We need to prepare a middleware to use Rxjs The event of is converted into SSE Send out .

const { createEventStream } = require("http-event-stream")
const { Sink } = require('fastrx')
module.exports = async function(ctx, next) {
    const sink = new Sink
    const stream = createEventStream(ctx.res, {
        onClose() {
            sink.dispose()
        }
    })
    sink.next = data => stream.sendMessage({ event: 'message', data: JSON.stringify(data) })
    sink.complete = err => stream.close();
    sink.subscribe(await next())
    ctx.respond = false
}

Copy

there fastrx library , It's my own high-speed Rxjs Realization , You can go NPM Check it out on the website . I have also written articles on relevant principles in detail . With this middleware , Suppose we need to start from MongoDB Every 5 Read data once per second . our Controller The response function can be written like this （koa-router）

async function facade({collection }) {
    return pipe(interval(5000), startWith(0), switchMap(x => fromPromise(collection('apps').aggregate([{
        $lookup: {
            from: "servers",
            as: "servers",
            let: { app: "$_id" },
            pipeline: [
                { $match: { $expr: { $eq: ["$app", "$$app"] } } },
                { $group: { _id: "$env", envs: { $push: "$$ROOT" } } },
                { $project: { k: "$_id", v: "$envs", _id: 0 } }
            ]
        }
    }, { $addFields: { servers: { $arrayToObject: "$servers" } } }]).toArray())))
}

Copy

among collection Is from Mongodb Of client Ahead of time Koa Encapsulation in .

Advanced

There are many limitations in obtaining data regularly , In the real scene , We often need to broadcast data to the monitoring front desk in time when an event occurs , And some data is not stored somewhere for you to obtain . Then I need to establish a data source to Koa The pipe in the middle of the controller . Here is the introduction pm2 Managed interprocess communication . This method can also be extended to network communication .

const subject_rounds = rx.subject()
pm2.launchBus((err, bus) => bus.on('myEvent', ({ data, process }) => subject_rounds.next(data)))

Copy

combination koa-router

router.get('/getData', eventStreamM, ctx => subject_rounds)

Copy

among eventStreamM Middleware mentioned before

The browser side can use EventSource The object is listening for data . In the process of another data source , We use it process.send({type:"myEvent",data: data }) Broadcast events to pm2 Just go to the event bus .

For other system architectures , We can use different technologies to broadcast data , Such as message queue . But it can be used in the end Rxjs Medium subject As a bridge to SSE Push Events .Subject It is this kind of existence , Act as both observer and observable object , about SSE It is an observable object , For other data sources, they are observers . If more than one person opens SSE monitor , Can perfectly deal with .