1. 程式人生 > >以中介軟體,路由,跨程序事件的姿勢使用WebSocket--Node.js篇

以中介軟體,路由,跨程序事件的姿勢使用WebSocket--Node.js篇

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

Node.js中比較流行的兩個WebSocket庫分別是socket.iows。其中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-nodeeasy-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/[email protected]/lib/easy-socket.min.js"></script>
<script>
    var client = new EasySocket({
        name: 'demo',
        autoReconnect: true,
        pingMsg: '{"type":"event","event":"ping","args":"ping"}'
    });
    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();
        }).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>
複製程式碼

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