发布/订阅

Question

我试图在Node中实现发布/订阅模式，但没有使用Redis。功能应该是一样的;您可以发布到频道，订阅频道并在订阅后收听数据;这里是Redis的功能：

pub.publish('channel1', myData1, function (err, data) {  
}); 


sub.subscribe('channel1', function (err, data) { 
}); 


sub.on("message", function (channel, data) { 
    //Now you are easily listening to data 
}); 

我曾尝试以下，这可能不是在所有的好方法（我只是展示我的企图！）;我有自己的频道对象，用于保存在向特定频道发布新数据时需要触发的频道和功能;所以你看到我正在为频道订阅功能;但是，我不知道是否有任何方法可以像redis提供的数据一样收听数据！所以，请你帮我，如果因为Redis的做我能做到同样的方式，但eventEmitter或...

var events = require('events'); 
var emitter = new events.EventEmitter; 
var channels = {}; 
//channels ={"channelA": ["func1", "func2"], "channelB": []} 

//publish data to a channel and emits all of the functions which are registered for listening to the channel 
function publish(channel, data, callback) { 
    if (channels[channel] && channels[channel].length > 0) { 
     for (var i = 0; i < channels[channel].length; i++) { 
      emitter.emit(channels[channel][i], data); 
     } 
    } 
    callback(); 

} 

//subscribe a function to a channel 
function subscribe(channel, FuncName, next) { 
    if (!channels[channel]) { 
     channels[channel] = []; 
    } 
    channels[channel].push(FuncName); 
    next(); 
} 

//Subscribe a callback function from a channel 
function unsubscribe(channel, FuncName) { 
    if (channels[channel]) { 
     var indx = channels[channel].indexOf(FuncName); 
     if (indx != -1) { 
      channels[channel].splice(indx, 1); 
     } 
    } 
} 

Answer 1

你可以使用一个EventEmitter每个通道。另外，您可能希望保存对回调的引用，而不是函数名称字符串。

var events = require('events'); 
var channels = {}; 

//publish data to a channel and emits all of the functions which are registered for listening to the channel 
function publish(channel, data, callback) { 
    if (channels[channel]) { 
    channels[channel].emit('message', channel, data); 
    } 
    callback(); 
} 

//subscribe a function to a channel 
function subscribe(channel, fn, next) { 
    if (!channels[channel]) { 
    channels[channel] = new events.EventEmitter(); 
    } 
    channels[channel].addListener('message', fn); 
    next(); 
} 

//unsubscribe a callback function from a channel 
function unsubscribe(channel, fn) { 
    if (channels[channel]) { 
    channels[channel].removeListener('message', fn); 
    } 
}