以中間件，路由，跨進程事件的姿式使用WebSocket--Node.js篇

上一篇文章介紹了在瀏覽器端以中間件，路由，跨進程事件的姿式使用原生WebSocket。這篇文章將介紹如何使用Node.js以相同的編程模式來實現WebSocket服務端。

Node.js中比較流行的兩個WebSocket庫分別是socket.io與ws。其中socket.io已經實現了跨進程事件，廣播，羣發等功能，而且服務端與瀏覽器端是配套的，在不支持WebSocket技術的瀏覽器會降級爲使用ajax輪詢。因此。這裏選擇使用相對而言較爲底層或原始的ws,在其基礎上實現文章標題所提到的編程模式。

WS

使用ws簡簡單單就能夠啓動一個WebSocket服務：

const WebSocket = require('ws');

const wss = new WebSocket.Server({ port: 8080 });

wss.on('connection', function connection(ws) {
  ws.on('message', function incoming(message) {
    console.log('received: %s', message);
  });

  ws.send('something');
});
複製代碼

上面的wss支持的事件有connection,close,error,headers,listening,ws支持的事件有message,close,error。更多詳情能夠這裏。

中間件

對ws進行封裝，上面提到的事件：WSS的connection，ws的message,close,error。分別提供註冊中間件的接口

class EasySocket {
    constructor() {
        this.connectionMiddleware = [];
        this.closeMiddleware = [];
        this.messageMiddleware = [];
        this.errorMiddleware = [];

        this.connectionFn = Promise.resolve();
        this.closeFn = Promise.resolve();
        this.messageFn = Promise.resolve();
        this.errorFn = Promise.resolve();
    }
    connectionUse(fn, runtime) {
        this.connectionMiddleware.push(fn);
        if (runtime) {
            this.connectionFn = compose(this.connectionMiddleware);
        }
        return this;
    }
    closeUse(fn, runtime) {
        this.closeMiddleware.push(fn);
        if (runtime) {
            this.closeFn = compose(this.closeMiddleware);
        }
        return this;
    }
    messageUse(fn, runtime) {
        this.messageMiddleware.push(fn);
        if (runtime) {
            this.messageFn = compose(this.messageMiddleware);
        }
        return this;
    }
    errorUse(fn, runtime) {
        this.errorMiddleware.push(fn);
        if (runtime) {
            this.errorFn = compose(this.errorMiddleware);
        }
        return this;
    }
    
}
複製代碼

經過xxxUse註冊相應的中間件。 xxxMiddleware中就是相應的中間件。xxxFn是中間件經過compose處理後的結構。使用runtime參數能夠在運行時註冊中間件。

再添加一個listen方法，處理相應的中間件而且實例化WebSocket.Server

listen(config) {
        this.socket = new WebSocket.Server(config);
        this.connectionFn = compose(this.connectionMiddleware);
        this.messageFn = compose(this.messageMiddleware);
        this.closeFn = compose(this.closeMiddleware);
        this.errorFn = compose(this.errorMiddleware);
        this.socket.on('connection', (client, req) => {
            let context = { server: this, client, req };
            this.connectionFn(context).catch(error => { console.log(error) });

            client.on('message', (message) => {
                let req;
                try {
                    req = JSON.parse(message);
                } catch (error) {
                    req = message;
                }
                let messageContext = { server: this, client, req }
                this.messageFn(messageContext).catch(error => { console.log(error) })
            });

            client.on('close', (code, message) => {
                let closeContext = { server: this, client, code, message };
                this.closeFn(closeContext).catch(error => { console.log(error) })
            });

            client.on('error', (error) => {
                let errorContext = { server: this, client, error };
                this.errorFn(errorContext).catch(error => { console.log(error) })
            });
        })
    }
複製代碼

使用koa-compose模塊處理中間件。注意xxContext傳入了哪些東西，後續定義中間件的時候均可以使用。

compose的做用可看這篇文章 傻瓜式解讀koa中間件處理模塊koa-compose

使用：

import EasySocket from 'easy-socket-node';

const config = {
    port: 3001,
    perMessageDeflate: {
        zlibDeflateOptions: { // See zlib defaults.
            chunkSize: 1024,
            memLevel: 7,
            level: 3,
        },
        zlibInflateOptions: {
            chunkSize: 10 * 1024
        },
        // Other options settable:
        clientNoContextTakeover: true, // Defaults to negotiated value.
        serverNoContextTakeover: true, // Defaults to negotiated value.
        //clientMaxWindowBits: 10, // Defaults to negotiated value.
        serverMaxWindowBits: 10,       // Defaults to negotiated value.
        // Below options specified as default values.
        concurrencyLimit: 10,          // Limits zlib concurrency for perf.
        threshold: 1024,               // Size (in bytes) below which messages
        // should not be compressed.
    }
}
const easySocket = new EasySocket();
//使用中間件獲取token
easySocket
    .connectionUse((context,next)=>{
       console.log("new Connected");
       let location = url.parse(context.req.url, true);
       let token=location.query.token;
       if(!token){
           client.send("invalid token");
           client.close(1003, "invalid token");
           return;
       }
       context.client.token=token;
       next();
    });
easySocket
    .listen(config)

console.log('Now start WebSocket server on port ' + config.port + '...')
複製代碼

使用messageUse能夠註冊多個處理消息的中間件，好比

easySocket.messageUse((context, next) => {
    //羣聊處理中間件
    if (context.req.action === 'roomChatMessage') {
      //能夠在這裏持久化消息，將消息發送給其它羣聊客戶端
      console.log(context.req);
    }
    next();
  })
  .messageUse((context, next) => {
    //私聊處理中間件
    if (context.req.action === 'privateChatMessage') {
      //能夠在這裏持久化消息，將消息發送給私聊客戶端
      console.log(context.req);
    }
    next();
  })
複製代碼

每一箇中間件都要判斷context.req.action，而這個context.res就是瀏覽器端或客戶端發送的數據。怎麼消除這個頻繁的if判斷呢? 咱們實現一個簡單的消息處理路由。

路由

定義消息路由中間件

messageRouteMiddleware.js

export default (routes) => {
    return async (context, next) => {
        if (routes[context.req.action]) {
            await routes[context.req.action](context,next);
        } else {
            console.log(context.req)
            next();
        }
    }
}

複製代碼

定義路由

router.js

export default {
    roomChatMessage:function(context,next){
        //能夠在這裏持久化消息，將消息發送給其它羣聊客戶端，以及其它業務邏輯
        console.log(context.req);
        next();
    },
    privateChatMessage:function(context,next){
        //能夠在這裏持久化消息，將消息發送給私聊客戶端，以及其它業務邏輯
        console.log(context.req);
        next();
    }
}
複製代碼

使用：

easySocket.messageUse(messageRouteMiddleware(router))
複製代碼

跨進程事件

上一篇文章已經介紹了跨進程事件，這裏直接說實現。

使用Node的原生事件模塊

import compose from './compose';
const WebSocket = require('ws');
var EventEmitter = require('events').EventEmitter;
export default class EasySocket extends EventEmitter {
    constructor() {
        ...
        this.remoteEmitMiddleware = [];

        ...
        this.remoteEmitFn = Promise.resolve();
    }
    ...
    remoteEmitUse(fn, runtime) {
        this.remoteEmitMiddleware.push(fn);
        if (runtime) {
            this.remoteEmitFn = compose(this.remoteEmitMiddleware);
        }
        return this;
    }
    listen(config) {
        this.socket = new WebSocket.Server(config);
        ...
        this.remoteEmitFn = compose(this.remoteEmitMiddleware);

        ...
    }
    emit(event, args, isLocal = false) { 
        let arr = [event, args];
        if (isLocal) {
            super.emit.apply(this, arr);
            return this;
        }
        let evt = {
            event: event,
            args: args
        }
        let remoteEmitContext = { server: this, event: evt };
        this.remoteEmitFn(remoteEmitContext).catch(error => { console.log(error) })
        return this;
    }
}
複製代碼

最後

源碼地址：easy-socket-node

基於easy-socket-node與easy-socket-browser一個完整例子：

index.html

<!DOCTYPE html>
<html lang="en">

<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <meta http-equiv="X-UA-Compatible" content="ie=edge">
    <title>Document</title>
</head>

<body>
</body>
<script src="https://unpkg.com/easy-socket-browser@1.1.1/lib/easy-socket.min.js"></script>
<script> <!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <meta name="viewport" content="width=device-width, initial-scale=1.0"> <meta http-equiv="X-UA-Compatible" content="ie=edge"> <title>Document</title> </head> <body> </body> <script src="https://unpkg.com/easy-socket-browser@1.1.1/lib/easy-socket.min.js"></script>
<script> var client = new EasySocket({ name: 'demo', autoReconnect: true, pingMsg: '{"type":"event","event":"ping","args":"ping"}'//模擬emit 消息體 }); client.openUse((context, next) => { console.log("open"); next(); }) .closeUse((context, next) => { console.log("close"); next(); }).errorUse((context, next) => { console.log("error", context.event); next(); }).messageUse((context, next) => { if (context.res.type === 'event') { context.client.emit(context.res.event, context.res.args, true); } next(); }) .reconnectUse((context, next) => { console.log('正在進行重連') next(); }) .remoteEmitUse((context, next) => { let client = context.client; let event = context.event; if (client.socket.readyState !== 1) { console.log("鏈接已斷開"); } else { client.socket.send(JSON.stringify({ type: 'event', event: event.event, args: event.args })); next(); } }); client.connect('ws://localhost:3001'); var msg = 1; setInterval(() => { client.emit('chatMessage', msg++) }, 3000); client.on("serverMessage", (data) => { console.log("serverMessage:" + data) }); </script>

</html>

</script>

</html>
複製代碼

server.js

var EasySocket = require('easy-socket-node').default;
var config = {
    port: 3001,
    perMessageDeflate: {
        zlibDeflateOptions: { // See zlib defaults.
            chunkSize: 1024,
            memLevel: 7,
            level: 3,
        },
        zlibInflateOptions: {
            chunkSize: 10 * 1024
        },
        // Other options settable:
        clientNoContextTakeover: true, // Defaults to negotiated value.
        serverNoContextTakeover: true, // Defaults to negotiated value.
        //clientMaxWindowBits: 10, // Defaults to negotiated value.
        serverMaxWindowBits: 10,       // Defaults to negotiated value.
        // Below options specified as default values.
        concurrencyLimit: 10,          // Limits zlib concurrency for perf.
        threshold: 1024,               // Size (in bytes) below which messages
        // should not be compressed.
    }
}

var remoteEmitMiddleware = (context, next) => {
    var server = context.server;
    var event = context.event;
    for (let client of server.clients.values()) {
        client.readyState == 1 && client.send(makeEventMessage(event));
    }
}
function makeEventMessage(event) {
    return JSON.stringify({
        type: 'event',
        event: event.event,
        args: event.args
    })
}
var messageRouteMiddleware = (routes) => {
    return (context, next) => {
        if (context.req.type === 'event') {
            if (routes[context.req.event]) {
                routes[context.req.event](context, next);
            } else {
                context.server.emit(context.req.event, context.req.args);//將會直接觸發remoteEmitMiddleware 中間件的調用
                next();
            }
        } else {
            next();
        }
    }
}
var router = {
    chatMessage: (context, next) => {
        var req = context.req;
        context.server.emit('serverMessage', req.args);
    }
}
var server = new EasySocket();
server
    .connectionUse((context, next) => {
        context.server.clients.set(1, context.client)
        console.log('new connection')
    })
    .closeUse((context, next) => {
        console.log('close')
    })
    .messageUse(messageRouteMiddleware(router))
    .remoteEmitUse(remoteEmitMiddleware)
    .listen(config)

console.log('Now start WebSocket server on port ' + config.port + '...')
複製代碼

運行過程，能夠中止後端服務，而後再啓動，測下心跳重連

實現的聊天室例子：online chat demo

聊天室前端源碼：lazy-mock-im

聊天室服務端源碼：lazy-mock