Node.js与Socket.io实现P2P实时消息通信的实践

    'use strict';

    // 维护socket连接的代码
    const { addSocketId, getSocketId, deleteSocketId } = require('../../../utils/socket/socketId');
    // 保存消息
    const message = require('../saveMessage');
    // socket连接许可验证
    const { socketAuth } = require('../../../middleware/socket/index')

    // socket接口，传入/bin/www.js
    function init(io) {
    
    /**
     * @description: 为每个传入执行的功能Socket，并且接收套接字和可选地将执行延迟到下一个注册的中间件的参数
     */    
    io.use((socket, next) => {
        if (socket.request.headers.cookie) return next();
        next(new Error('Authentication error'));
    });

    io.on('connection', function(socket) {

        /**
         * @description: 用户登录，则保存用户连接的相关信息,并从redis拉取未读消息，推送给该用户
         */        
        socket.on('user_login', function(socketInfo) {       
            if(!socketInfo.userId) {
                // io.sockets.to(socketInfo['socketId']).emit('disconnect', '');
                return;
            }
            // 将用户与socket插入数据库中
            addSocketId(socketInfo);  
                      
            if (process.env.NODE_ENV === 'development') {
                displayUserInfo(socketInfo);
            };

            // 推送所有消息
            message.pushMessage(socketInfo['userId']).then(pushData => {
                io.sockets.to(socketInfo['socketId']).emit('push_message', pushData);
            });
        });
    
        /**
         * @description: 发给某用户交易通知（在线实时通知，并存储至redis）
         */        
        socket.on('todo', function(todoData) {    
            // 存入redis
            message.addMessage(todoData);
            // 检测用户是否在线
            message.isOnline(todoData['receiver_id']).then(isOnline => {
                // 用户在线则通信
                if (isOnline == true) {
                    getSocketId(todoData['receiver_id']).then(socketId => {            
                        io.sockets.to(socketId).emit('todo_message', todoData);
                    }); 
                };  
            });
        });
    
        // TODO: 需要提醒前端在关闭窗口之前先断开连接（窗口刷新之前应该不需要）
        /**
         * @description: 断开连接
         */        
        socket.on('disconnect', function() {
            // 从数据库中删除连接
            deleteSocketId(socket.id);
            // 判断当前是否是开发环境
            if (process.env.NODE_ENV === 'development') {
                displayUserInfo();
            }
        });
    
    });
    
}

function displayUserInfo(user) {
    console.log(`当前登录用户信息:${user}`);
    return;
}

module.exports = {
    init
};

#!/usr/bin/env node

// 模块依赖
var app = require('../app');
var http = require('http');
const socketIndex = require('../src/routes/socket/index/socket');

// 从环境中取端口，应用到express
var port = normalizePort(process.env.PORT || '3000');
app.set('port', port);

// 创建http服务（将express注册到http中）
server = http.createServer(app);

// 监听
var io = require('socket.io')(server, {
  cors: {
      origin: '*'
  }
  // path: '/socket' // 重新定义socket连接路径
});

// 全局声明
global.io = io;

// socket的程序文件下引入io
socketIndex.init(io);