Node.js redis pub/sub缺失消息

Question

我试图从两天内调试这段代码，但我无法理解，所以我请求您。

代码非常简单:两个端点几乎完成了相同的任务。

第一个是's'，侦听频道'idS‘上的任何消息。如果10秒后它没有收到任何东西，它就结束了。如果req.body不是空的，将其发送到'c‘。

在第二个，'c'，总是发送一条消息'idS‘，并等待一个消息频道'idC’。

在“s”中，变量“id”设置为1，因此“c”可以理解“s”何时在线。

我有两个不同的测试程序(每个端点一个)。

在testS中，我调用不带req.body的端点，当我收到答复时，再次调用端点，这一次是使用req.body。

在testC中，我只需调用端点并等待答案。当我收到它时，我会在一秒钟后再次调用端点。

测试应该以这样的方式工作：

1. send，它没有什么可发送的，只能监听频道‘it’。
2. C连接，收听频道'idC‘，并在'idS’上发送“测试”。
3. S在频道'idS‘上接收消息"test“并退出。
4. 一个新的s连接，监听频道'idS‘，并在'idC’上向c发送消息。
5. C接收频道'idC‘并退出。
6. 一个新的c连接在1秒后再次连接，并从2中重复。

var express = require('express'),
    redis = require('redis'),
    util = require('util'),
    client = redis.createClient(6379, 'localhost');
var router = express.Router();

router.post('/s', function(req, res){
    var id = req.query.code;
    var cId = "c_" + id;
    var sId= "s_" + id;

    //BUILD RESPONS
    var respons = {....};

    var clientSub = redis.createClient(6379, "localhost");
    clientSub.on("message", function(channel, msg){
        if(timeoutId)
            clearTimeout(timeoutId);
        client.del(id);
        respons.val = idS;
        clientSub.unsubscribe(idS);
        clientSub.quit();
        res.send(respons).end();
    });
    clientSub.subscribe(idS);
    client.set(id, 1);

    if(req.body){
        client.publish(idC, JSON.stringify(req.body));
    }

    var timeoutId = setTimeout(function(){
        if(!res.headersSent){
            clientSub.unsubscribe(idS);
            clientSub.quit();
            client.del(id);
            console.log(" HUB TIMEOUTED " + id);

            res.json(respons).end();
        }
    },1000*10);
});

router.post('/c', function(req, res){    
    var id = //code to get id
    var idC = "c_" + id;
    var idS= "s_" + id;


    var clientSub = redis.createClient(6379, "localhost");
    clientSub.once("message", function(channel, msg){
        var respons = {};
        respons.data = msg;
        clientSub.unsubscribe(idC);
        clientSub.quit();
        res.json(respons).end();
    });
    //subscribe to a channel
    clientSub.subscribe(idC);


    //try to send message
    client.get(id, function(err, reply){
        if(err)
            console.log(err);
        else if(reply == 1){
            client.publish(idS, "test");
        } else{
            console.log("ERROR!!!");
        }

    });
});

问题是，有时'c‘没有收到'idC’上的消息(或者s不发送它？)。

错误不在测试程序中。

谢谢你的帮忙!

Answer 1

用当时的红葡萄酒解决了。问题是同步问题。