{"action":"created","issue":{"url":"https://api.github.com/repos/LeanKit-Labs/wascally/issues/39","labels_url":"https://api.github.com/repos/LeanKit-Labs/wascally/issues/39/labels{/name}","comments_url":"https://api.github.com/repos/LeanKit-Labs/wascally/issues/39/comments","events_url":"https://api.github.com/repos/LeanKit-Labs/wascally/issues/39/events","html_url":"https://github.com/LeanKit-Labs/wascally/issues/39","id":57772366,"number":39,"title":"Unacked messages piling up in response queue","user":{"login":"neverfox","id":3344033,"avatar_url":"https://avatars.githubusercontent.com/u/3344033?v=3","gravatar_id":"","url":"https://api.github.com/users/neverfox","html_url":"https://github.com/neverfox","followers_url":"https://api.github.com/users/neverfox/followers","following_url":"https://api.github.com/users/neverfox/following{/other_user}","gists_url":"https://api.github.com/users/neverfox/gists{/gist_id}","starred_url":"https://api.github.com/users/neverfox/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/neverfox/subscriptions","organizations_url":"https://api.github.com/users/neverfox/orgs","repos_url":"https://api.github.com/users/neverfox/repos","events_url":"https://api.github.com/users/neverfox/events{/privacy}","received_events_url":"https://api.github.com/users/neverfox/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":2,"created_at":"2015-02-16T07:18:51Z","updated_at":"2015-02-18T01:39:45Z","closed_at":null,"body":"I have a producer/dispatcher/consumer topology that involves one case of `request` instead of `publish` but since adding that, the related response queue is filling up with unacked messages, despite the fact that I'm calling `reply` on the consumer side and `res.ack` in the promise `then` of the request. What could be going on here? I even tried explicitly `ack`ing the message in the consumer, with no luck.\r\n\r\n```javascript\r\n// topology.js\r\n'use strict';\r\n\r\nmodule.exports = function (rabbit, subscribeTo) {\r\n\treturn rabbit.configure({\r\n\t\t// arguments used to establish a connection to a broker\r\n\t\tconnection: {\r\n\t\t\tuser:   'guest',\r\n\t\t\tpass:   'guest',\r\n\t\t\tserver: ['127.0.0.1'],\r\n\t\t\tport:   5672,\r\n\t\t\tvhost:  '%2f'\r\n\t\t},\r\n\r\n\t\t// define the exchanges\r\n\t\texchanges: [\r\n\t\t\t{\r\n\t\t\t\tname: 'tasks-x',\r\n\t\t\t\ttype: 'fanout',\r\n\t\t\t\tautoDelete: true\r\n\t\t\t},\r\n\t\t\t{\r\n\t\t\t\tname: 'dispatch-x',\r\n\t\t\t\ttype: 'direct',\r\n\t\t\t\tautoDelete: true\r\n\t\t\t},\r\n\t\t\t{\r\n\t\t\t\tname: 'requests-x',\r\n\t\t\t\ttype: 'fanout',\r\n\t\t\t\tautoDelete: true\r\n\t\t\t}\r\n\t\t],\r\n\r\n\t\t// setup the queues, only subscribing to the one this service\r\n\t\t// will consume messages from\r\n\t\tqueues: [\r\n\t\t\t{\r\n\t\t\t\tname: 'tasks-q',\r\n\t\t\t\tautoDelete: true,\r\n\t\t\t\tlimit: 1,\r\n\t\t\t\tsubscribe: subscribeTo === 'tasks'\r\n\t\t\t},\r\n\t\t\t{\r\n\t\t\t\tname: 'requests-q',\r\n\t\t\t\tautoDelete: true,\r\n\t\t\t\tlimit: 1,\r\n\t\t\t\tsubscribe: subscribeTo === 'requests'\r\n\t\t\t},\r\n\t\t\t{\r\n\t\t\t\tname: 'p1-q',\r\n\t\t\t\tautoDelete: true,\r\n\t\t\t\tlimit: 1,\r\n\t\t\t\tsubscribe: subscribeTo === 'p1'\r\n\t\t\t},\r\n\t\t\t{\r\n\t\t\t\tname: 'p2-q',\r\n\t\t\t\tautoDelete: true,\r\n\t\t\t\tlimit: 1,\r\n\t\t\t\tsubscribe: subscribeTo === 'p2'\r\n\t\t\t}\r\n\t\t],\r\n\r\n\t\t// binds exchanges and queues to one another\r\n\t\tbindings: [\r\n\t\t\t{\r\n\t\t\t\texchange: 'tasks-x',\r\n\t\t\t\ttarget:   'tasks-q',\r\n\t\t\t\tkeys:     []\r\n\t\t\t},\r\n\t\t\t{\r\n\t\t\t\texchange: 'requests-x',\r\n\t\t\t\ttarget:   'requests-q',\r\n\t\t\t\tkeys:     []\r\n\t\t\t},\r\n\t\t\t{\r\n\t\t\t\texchange: 'dispatch-x',\r\n\t\t\t\ttarget:   'p1-q',\r\n\t\t\t\tkeys:     ['p1']\r\n\t\t\t},\r\n\t\t\t{\r\n\t\t\t\texchange: 'dispatch-x',\r\n\t\t\t\ttarget:   'p2-q',\r\n\t\t\t\tkeys:     ['p2']\r\n\t\t\t}\r\n\t\t]\r\n\t});\r\n};\r\n```\r\n\r\n```javascript\r\n// producer.js\r\n'use strict';\r\n\r\nvar rabbit = require('wascally');\r\n\r\nvar i = 0;\r\n\r\nrequire('./topology')(rabbit).then(function () {\r\n\tsetInterval(function () {\r\n\t\tconsole.log('Publishing task');\r\n\t\trabbit.publish('tasks-x', 'task', {\r\n\t\t\ttask: ++i\r\n\t\t});\r\n\t}, 3000);\r\n});\r\n```\r\n\r\n```javascript\r\n// dispatcher.js\r\n'use strict';\r\n\r\nvar Kefir  = require('kefir');\r\nvar rabbit = require('wascally');\r\n\r\nvar taskEmitter = Kefir.emitter();\r\nvar reqEmitter  = Kefir.emitter();\r\nvar pairs = Kefir.zip([taskEmitter, reqEmitter]);\r\n\r\npairs.onValue(function (pair) {\r\n\tvar task = pair[0];\r\n\tvar req  = pair[1];\r\n\tconsole.log(task.body, req.body);\r\n\trabbit.request('dispatch-x', {\r\n\t\ttype: 'process',\r\n\t\tbody: task.body,\r\n\t\troutingKey: 'p1'\r\n\t})\r\n\t\t.then(function (res) {\r\n\t\t\tconsole.log(res.body);\r\n\t\t\tres.ack();\r\n\t\t\ttask.ack();\r\n\t\t\treq.ack();\r\n\t\t});\r\n});\r\n\r\nrabbit.handle('task', function (task) {\r\n\tconsole.log('Dispatch received', JSON.stringify(task.body));\r\n\ttaskEmitter.emit(task);\r\n});\r\n\r\nrabbit.handle('request', function (req) {\r\n\tconsole.log('Dispatch received', JSON.stringify(req.body));\r\n\treqEmitter.emit(req);\r\n});\r\n\r\nrequire('./topology')(rabbit).then(function () {\r\n\trabbit.startSubscription('requests-q');\r\n\trabbit.startSubscription('tasks-q');\r\n});\r\n```\r\n\r\n```javascript\r\n/// consumer.js\r\n'use strict';\r\n\r\nvar rabbit = require('wascally');\r\n\r\nvar i = 0;\r\n\r\nfunction requestJob () {\r\n\trabbit.publish('requests-x', 'request', {\r\n\t\treq: ++i\r\n\t});\r\n}\r\n\r\nrabbit.handle('process', function (msg) {\r\n\tconsole.log('Received:', JSON.stringify(msg.body));\r\n\tmsg.reply({ success: true });\r\n\trequestJob();\r\n});\r\n\r\nrequire('./topology')(rabbit, 'p1').then(function () {\r\n\trequestJob();\r\n});\r\n```"},"comment":{"url":"https://api.github.com/repos/LeanKit-Labs/wascally/issues/comments/74794928","html_url":"https://github.com/LeanKit-Labs/wascally/issues/39#issuecomment-74794928","issue_url":"https://api.github.com/repos/LeanKit-Labs/wascally/issues/39","id":74794928,"user":{"login":"arobson","id":115508,"avatar_url":"https://avatars.githubusercontent.com/u/115508?v=3","gravatar_id":"","url":"https://api.github.com/users/arobson","html_url":"https://github.com/arobson","followers_url":"https://api.github.com/users/arobson/followers","following_url":"https://api.github.com/users/arobson/following{/other_user}","gists_url":"https://api.github.com/users/arobson/gists{/gist_id}","starred_url":"https://api.github.com/users/arobson/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/arobson/subscriptions","organizations_url":"https://api.github.com/users/arobson/orgs","repos_url":"https://api.github.com/users/arobson/repos","events_url":"https://api.github.com/users/arobson/events{/privacy}","received_events_url":"https://api.github.com/users/arobson/received_events","type":"User","site_admin":false},"created_at":"2015-02-18T01:39:45Z","updated_at":"2015-02-18T01:39:45Z","body":"@neverfox - Thanks for trying that out. I'm stumped. I've spent a good bit of time slowly importing the new approach to the working commit trying to find what's caused the defect and nothing yet."}}