{"action":"created","issue":{"url":"https://api.github.com/repos/squaremo/amqp.node/issues/144","labels_url":"https://api.github.com/repos/squaremo/amqp.node/issues/144/labels{/name}","comments_url":"https://api.github.com/repos/squaremo/amqp.node/issues/144/comments","events_url":"https://api.github.com/repos/squaremo/amqp.node/issues/144/events","html_url":"https://github.com/squaremo/amqp.node/issues/144","id":60771482,"number":144,"title":"Publishing without closing channel","user":{"login":"dbousamra","id":648137,"avatar_url":"https://avatars.githubusercontent.com/u/648137?v=3","gravatar_id":"","url":"https://api.github.com/users/dbousamra","html_url":"https://github.com/dbousamra","followers_url":"https://api.github.com/users/dbousamra/followers","following_url":"https://api.github.com/users/dbousamra/following{/other_user}","gists_url":"https://api.github.com/users/dbousamra/gists{/gist_id}","starred_url":"https://api.github.com/users/dbousamra/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/dbousamra/subscriptions","organizations_url":"https://api.github.com/users/dbousamra/orgs","repos_url":"https://api.github.com/users/dbousamra/repos","events_url":"https://api.github.com/users/dbousamra/events{/privacy}","received_events_url":"https://api.github.com/users/dbousamra/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":3,"created_at":"2015-03-12T05:00:11Z","updated_at":"2015-03-14T22:09:03Z","closed_at":null,"body":"Hi all,\r\n\r\nWe are trying to publish without closing the channel (i.e. have a long running channel as suggested). When we publish, if we don't close the channel, messages aren't flushed from the buffer and sent.\r\n\r\nHow do we \"flush\" regularly, so messages are sent, while still keeping a channel open?\r\n\r\nDom"},"comment":{"url":"https://api.github.com/repos/squaremo/amqp.node/issues/comments/80741858","html_url":"https://github.com/squaremo/amqp.node/issues/144#issuecomment-80741858","issue_url":"https://api.github.com/repos/squaremo/amqp.node/issues/144","id":80741858,"user":{"login":"dbousamra","id":648137,"avatar_url":"https://avatars.githubusercontent.com/u/648137?v=3","gravatar_id":"","url":"https://api.github.com/users/dbousamra","html_url":"https://github.com/dbousamra","followers_url":"https://api.github.com/users/dbousamra/followers","following_url":"https://api.github.com/users/dbousamra/following{/other_user}","gists_url":"https://api.github.com/users/dbousamra/gists{/gist_id}","starred_url":"https://api.github.com/users/dbousamra/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/dbousamra/subscriptions","organizations_url":"https://api.github.com/users/dbousamra/orgs","repos_url":"https://api.github.com/users/dbousamra/repos","events_url":"https://api.github.com/users/dbousamra/events{/privacy}","received_events_url":"https://api.github.com/users/dbousamra/received_events","type":"User","site_admin":false},"created_at":"2015-03-14T22:09:03Z","updated_at":"2015-03-14T22:09:03Z","body":"The problem we are seeing is extremely low throughput when publishing. The RabbitMQ instance we are using is located in US-East (and we are publishing from Sydney).\r\n\r\nOur code for publishing looked like this:\r\n\r\n```javascript\r\nvar queueConnection = connect(config.queues.snapshots.host)\r\nvar queueName = config.queues.snapshots.queue_name\r\n\r\nfunction connect(queue) {\r\n  config.logger.info(\"Connecting to RabbitMQ at %s\", queue)\r\n  var connection = amqp.connect(queue);\r\n  return connection.then(function(conn) {\r\n    config.logger.info(\"Connected to RabbitMQ at %s\", queue)\r\n    conn.on('error', function(error) {\r\n      config.logger.error(JSON.stringify({ message: error.message, stack: error.stack }))\r\n      return reconnect(queue);\r\n    });\r\n    return conn\r\n  }).catch(function(err) {\r\n    config.logger.error(JSON.stringify({ message: err.message, stack: err.stack }))\r\n    return reconnect(queue);\r\n  })\r\n};\r\n\r\nfunction reconnect(queue) {\r\n  return Promise.delay(config.reconnect_rate).then(function() {\r\n    return connect(queue)\r\n  })\r\n}\r\n\r\nvar sendToSnapshotsQueue = function(message) {\r\n  return queueConnection.then(function(conn) {\r\n    return conn.createChannel()\r\n  }).then(function(ch) {\r\n    var queue = ch.assertQueue(queueName, { durable: false });\r\n    return queue.then(function() {\r\n      ch.sendToQueue(queueName, new Buffer(JSON.stringify(message)), { deliveryMode: true });\r\n      return ch.close();\r\n    });\r\n  })\r\n};\r\n```\r\n\r\nNotice how our sendToSnapshotsQueue function works. We reuse the same connection, but we create a channel, assertQueue, and sendToQueue, before closing the channel... each time we send a message. We could remove the assertQueue, and do that in our connection code rather than each time, sure.\r\n\r\nBut it seems as if opening a channel, sending and closing a channel, is taking a good 750ms. If we reuse the same channel, and send multiple messages, before closing, we can bump our throughput into the 1000's of messages per second, rather than 10's.\r\n\r\nThe issue that Dan pointed out, is that you don't close the channel, it doesn't seem to actually send them to the queue. Looking through the code, it seems when a buffer is filled, it actually performs the socket send, and queues the messages.\r\n\r\nI couldn't seem to fill it with enough messages (I queued 10000 messages in a loop, to see if it would send without closing the channel, but it didn't).\r\n\r\nSo I then came up with the following code to try and batch message queuing under a single channel:\r\n\r\n```javascript\r\nvar async = require('async')\r\n\r\nvar publishQueue = async.cargo((messages, cb) => {\r\n  return queueConnection.then(function(conn) {\r\n    return conn.createChannel()\r\n  }).then(function(ch) {\r\n    var queue = ch.assertQueue(queueName, { durable: false });\r\n    return queue.then(function() {\r\n      _.each(messages, function(message) {\r\n        ch.sendToQueue(queueName, new Buffer(JSON.stringify(message)), { deliveryMode: true });\r\n      })\r\n      return ch.close();\r\n    });\r\n  })\r\n}, 1000)\r\n\r\nvar sendToSnapshotsQueue = function(message) {\r\n  publishQueue.push(message);\r\n  return Promise.resolve(message);  \r\n};\r\n```\r\n\r\nThis is not the final code we used, but it illustrates the idea. We use async's [cargo construct] (https://github.com/caolan/async#cargo). When we call sendToSnapshotsQueue, we publish the in memory cargo queue, which tries to send as often as possible, in batches of 1000 (by reusing the same channel). \r\n\r\nThis increased our throughput massively, but is it the right solution? Any ideas?"}}