pengjunjie/server
1
0
Fork 1
mirror of https://github.com/ONLYOFFICE/server.git synced 2026-04-07 14:04:35 +08:00
Code Issues Packages Projects Releases Wiki Activity

error PRECONDITION_FAILED - unknown delivery tag

Browse Source
This commit is contained in:
konovalovsergey
2016-08-03 12:46:55 +03:00
parent 2a856916aa
commit d3de2411bc
1 changed files with 11 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
DocService/sources/pubsubRabbitMQ.js
View File

@ -54,14 +54,16 @@ function init(pubsub, callback) {
pubsub.exchangePublish = yield rabbitMQCore.assertExchangePromise(pubsub.channelPublish, cfgRabbitExchangePubSub,
'fanout', {durable: true});
var channelReceive = yield rabbitMQCore.createChannelPromise(conn);
var queue = yield rabbitMQCore.assertQueuePromise(channelReceive, '', {autoDelete: true, exclusive: true});
channelReceive.bindQueue(queue, cfgRabbitExchangePubSub, '');
yield rabbitMQCore.consumePromise(channelReceive, queue, function (message) {
if (message) {
pubsub.emit('message', message.content.toString());
pubsub.channelReceive = yield rabbitMQCore.createChannelPromise(conn);
var queue = yield rabbitMQCore.assertQueuePromise(pubsub.channelReceive, '', {autoDelete: true, exclusive: true});
pubsub.channelReceive.bindQueue(queue, cfgRabbitExchangePubSub, '');
yield rabbitMQCore.consumePromise(pubsub.channelReceive, queue, function (message) {
if(null != pubsub.channelReceive){
if (message) {
pubsub.emit('message', message.content.toString());
}
pubsub.channelReceive.ack(message);
}
channelReceive.ack(message);
}, {noAck: false});
//process messages received while reconnection time
repeat(pubsub);
@ -76,6 +78,7 @@ function init(pubsub, callback) {
function clear(pubsub) {
pubsub.channelPublish = null;
pubsub.exchangePublish = null;
pubsub.channelReceive = null;
}
function repeat(pubsub) {
for (var i = 0; i < pubsub.publishStore.length; ++i) {
@ -92,6 +95,7 @@ function PubsubRabbitMQ() {
this.connection = null;
this.channelPublish = null;
this.exchangePublish = null;
this.channelReceive = null;
this.publishStore = [];
}
util.inherits(PubsubRabbitMQ, events.EventEmitter);