在 redis 订阅事件发生后，如何加入套接字 io 房间？

Question

我正在使用 redis pub/sub，我想在 redis sub 事件之后加入一个房间：

const express = require('express');
const app = express();
const socketio = require('socket.io');
const redis = require("redis");

const expressServer = app.listen(3001, () => console.log("Express is running!"));

const io = socketio(expressServer);

const sub = redis.createClient({port: 6379, host: '127.0.0.1'});

sub.subscribe('message');


let room;

sub.on('error', function (error) {
    console.log('ERROR ' + error);
});

sub.on('connect', function(){
    console.log('Redis client connected');
});

//Redis sub scribe to message conversation channel
sub.on('message', async function (channel, data) {
    data = JSON.parse(data);
    if(channel === 'conversation'){
        room = data.id;
        console.log('room set ' + room)
    }
});

io.on('connect', (socket) => {
    socket.emit('test', 'from server')

    //Room I'd like to join
    socket.join(room);

    io.of("/").adapter.on("join-room", (room, id) => {
        console.log(`socket ${id} has joined room ${room}`);
    });
});

基本上我想要在sub.on('message') 上设置room 状态，然后在我想要socket.join(room); 之后。现在发生的事情是套接字服务器的连接速度比 redis 子事件快。在第一个子事件中，socket 服务器加入了一个未定义的房间，然后它加入了sub.on(message) 设置的最后一个房间

有没有办法可以使用节点事件发射器，以便在 redis 设置房间后加入套接字房间？或者我将如何使用 socket.io.reids.adapter 来做到这一点？

It looks like there's a way to do this with new redis socket.io adapter 函数，但没有关于如何做的文档？函数：RedisAdapter.remoteJoin(id, room)

编辑

我开始尝试使用带有套接字适配器的remoteJoin 执行此操作。问题是socket.id

const express = require('express');
const app = express();
const socketio = require('socket.io');
const redis = require("redis");
const redisAdapter = require('@socket.io/redis-adapter');

const expressServer = app.listen(3001, () => console.log("Express is running!"));

const io = socketio(expressServer);

const sub = redis.createClient({port: 6379, host: '127.0.0.1'});
const pub = sub.duplicate();

io.adapter(redisAdapter(pub, sub));

sub.subscribe('conversation','message', function(){
    // console.log('Subbed')
})

//sub.subscribe('conversation','message');

let room;
let soketId;

sub.on('error', function (error) {
    console.log('ERROR ' + error);
});

sub.on('connect', function(){
    console.log('Redis client connected');
});


sub.on('message', async function (channel, data) {
     data = JSON.parse(data);
     if(channel === 'conversation'){
        room = data.id;

        io.of('/').adapter.remoteJoin(socketId, room);

        console.log('subbed ' + room)
    }
});

io.on('connect', (socket) => {
    socketId = socket.id;

    io.of("/").adapter.on("join-room", (room, id) => {
        console.log(`socket ${id} has joined room ${room}`);
    });
});

问题是socketId = socket.id; 设置为最后一个连接的客户端。所以所有客户端都成为最后一个连接的客户端。

Answer 1

我已重构您的代码以响应来自 Redis pub/sub 的传入消息

可以使用新的 Redis 客户端或具有基于 Promise 的 API 的 ioredis 客户端来减少此代码。

每当有新客户端连接时，您都需要在 Redis 上订阅消息，以便在客户端连接时对新消息做出反应。

问题是当套接字断开时，您不再需要接收这些消息，所以我创建了一个辅助函数，它执行以下操作：

• 连接到redis
• 创建订阅
• 当有新消息到达时，它会调用回调函数
• 在调用时返回一个函数取消订阅消息并断开当前 Redis 客户端。

const express = require('express');
const app = express();
const socketio = require('socket.io');
const redis = require('redis');

const expressServer = app.listen(3001, () => console.log('Express is running!'));

const io = socketio(expressServer);

const createSubscriber = async cb => {
    const client = redis.createClient({ port: 6379, host: '127.0.0.1' });

    await new Promise((resolve, reject) => {
        client.on('connect', resolve);
        client.on('error', reason => {
            client.end();
            reject(reason);
        });
    });

    await new Promise((resolve, reject) => {
        client.subscribe('message', err => {
            if (err) {
                client.end();
                reject(err);
            } else {
                resolve();
            }
        });
    });

    client.on('message', (channel, message) => {
        if (channel === 'conversation') {
            cb(JSON.parse(message));
        }
    });

    return async () => {
        await new Promise(resolve => {
            client.unsubscribe('message', err => {
                if (err) {
                    console.error(err);
                }
                client.end();
                resolve();
            });
        });
    };
};

io.on('connect', async socket => {
    socket.emit('test', 'from server');

    try {
        const onDisconnected = await createSubscriber(data => {
            socket.join(data.id);
        });

        socket.on('disconnect', async () => {
            await onDisconnected();
        });
    } catch (error) {
        console.error(error); // redis has failed to connect
        socket.emit('error', error.message);
        socket.disconnect(); // disconnect the socket since there is nothing to do
    }
});

io.of('/').adapter.on('join-room', (room, id) => {
    console.log(`socket ${id} has joined room ${room}`);
});