pool_cluster.js 5.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243
  1. var Pool = require('./pool.js');
  2. var PoolConfig = require('./pool_config.js');
  3. var Util = require('util');
  4. var EventEmitter = require('events').EventEmitter;
  5. module.exports = PoolCluster;
  6. /**
  7. * PoolCluster
  8. */
  9. function PoolCluster(config) {
  10. EventEmitter.call(this);
  11. config = config || {};
  12. this._canRetry = typeof config.canRetry === 'undefined' ? true : config.canRetry;
  13. this._removeNodeErrorCount = config.removeNodeErrorCount || 5;
  14. this._defaultSelector = config.defaultSelector || 'RR';
  15. this._closed = false;
  16. this._lastId = 0;
  17. this._nodes = {};
  18. this._serviceableNodeIds = [];
  19. this._namespaces = {};
  20. this._findCaches = {};
  21. }
  22. Util.inherits(PoolCluster, EventEmitter);
  23. PoolCluster.prototype.of = function(pattern, selector) {
  24. pattern = pattern || '*';
  25. selector = selector || this._defaultSelector;
  26. selector = selector.toUpperCase();
  27. if (typeof Selector[selector] === 'undefined') {
  28. selector = this._defaultSelector;
  29. }
  30. var key = pattern + selector;
  31. if (typeof this._namespaces[key] === 'undefined') {
  32. this._namespaces[key] = new PoolNamespace(this, pattern, selector);
  33. }
  34. return this._namespaces[key];
  35. };
  36. PoolCluster.prototype.add = function(id, config) {
  37. if (typeof id === 'object') {
  38. config = id;
  39. id = 'CLUSTER::' + (++this._lastId);
  40. }
  41. if (typeof this._nodes[id] === 'undefined') {
  42. this._nodes[id] = {
  43. id: id,
  44. errorCount: 0,
  45. pool: new Pool({config: new PoolConfig(config)})
  46. };
  47. this._serviceableNodeIds.push(id);
  48. this._clearFindCaches();
  49. }
  50. };
  51. PoolCluster.prototype.getConnection = function(pattern, selector, cb) {
  52. var namespace;
  53. if (typeof pattern === 'function') {
  54. cb = pattern;
  55. namespace = this.of();
  56. } else {
  57. if (typeof selector === 'function') {
  58. cb = selector;
  59. selector = this._defaultSelector;
  60. }
  61. namespace = this.of(pattern, selector);
  62. }
  63. namespace.getConnection(cb);
  64. };
  65. PoolCluster.prototype.end = function() {
  66. if (this._closed) {
  67. return;
  68. }
  69. this._closed = true;
  70. for (var id in this._nodes) {
  71. this._nodes[id].pool.end();
  72. }
  73. };
  74. PoolCluster.prototype._findNodeIds = function(pattern) {
  75. if (typeof this._findCaches[pattern] !== 'undefined') {
  76. return this._findCaches[pattern];
  77. }
  78. var foundNodeIds;
  79. if (pattern === '*') { // all
  80. foundNodeIds = this._serviceableNodeIds;
  81. } else if (this._serviceableNodeIds.indexOf(pattern) != -1) { // one
  82. foundNodeIds = [pattern];
  83. } else { // wild matching
  84. var keyword = pattern.substring(pattern.length - 1, 0);
  85. foundNodeIds = this._serviceableNodeIds.filter(function (id) {
  86. return id.indexOf(keyword) === 0;
  87. });
  88. }
  89. this._findCaches[pattern] = foundNodeIds;
  90. return foundNodeIds;
  91. };
  92. PoolCluster.prototype._getNode = function(id) {
  93. return this._nodes[id] || null;
  94. };
  95. PoolCluster.prototype._increaseErrorCount = function(node) {
  96. if (++node.errorCount >= this._removeNodeErrorCount) {
  97. var index = this._serviceableNodeIds.indexOf(node.id);
  98. if (index !== -1) {
  99. this._serviceableNodeIds.splice(index, 1);
  100. delete this._nodes[node.id];
  101. this._clearFindCaches();
  102. node.pool.end();
  103. this.emit('remove', node.id);
  104. }
  105. }
  106. };
  107. PoolCluster.prototype._decreaseErrorCount = function(node) {
  108. if (node.errorCount > 0) {
  109. --node.errorCount;
  110. }
  111. };
  112. PoolCluster.prototype._getConnection = function(node, cb) {
  113. var self = this;
  114. node.pool.getConnection(function (err, connection) {
  115. if (err) {
  116. self._increaseErrorCount(node);
  117. if (self._canRetry) {
  118. console.warn('[Error] PoolCluster : ' + err);
  119. return cb(null, 'retry');
  120. } else {
  121. return cb(err);
  122. }
  123. } else {
  124. self._decreaseErrorCount(node);
  125. }
  126. connection._clusterId = node.id;
  127. cb(null, connection);
  128. });
  129. };
  130. PoolCluster.prototype._clearFindCaches = function() {
  131. this._findCaches = {};
  132. };
  133. /**
  134. * PoolNamespace
  135. */
  136. function PoolNamespace(cluster, pattern, selector) {
  137. this._cluster = cluster;
  138. this._pattern = pattern;
  139. this._selector = new Selector[selector]();
  140. }
  141. PoolNamespace.prototype.getConnection = function(cb) {
  142. var clusterNode = this._getClusterNode();
  143. if (clusterNode === null) {
  144. return cb(new Error('Pool does Not exists.'));
  145. }
  146. this._cluster._getConnection(clusterNode, function(err, connection) {
  147. if (err) {
  148. return cb(err);
  149. }
  150. if (connection === 'retry') {
  151. return this.getConnection(cb);
  152. }
  153. cb(null, connection);
  154. }.bind(this));
  155. };
  156. PoolNamespace.prototype._getClusterNode = function() {
  157. var foundNodeIds = this._cluster._findNodeIds(this._pattern);
  158. if (foundNodeIds.length === 0) {
  159. return null;
  160. }
  161. var nodeId = (foundNodeIds.length === 1) ? foundNodeIds[0] : this._selector(foundNodeIds);
  162. return this._cluster._getNode(nodeId);
  163. };
  164. /**
  165. * Selector
  166. */
  167. var Selector = {};
  168. Selector.RR = function () {
  169. var index = 0;
  170. return function(clusterIds) {
  171. if (index >= clusterIds.length) {
  172. index = 0;
  173. }
  174. var clusterId = clusterIds[index++];
  175. return clusterId;
  176. };
  177. };
  178. Selector.RANDOM = function () {
  179. return function(clusterIds) {
  180. return clusterIds[Math.floor(Math.random() * clusterIds.length)];
  181. };
  182. };
  183. Selector.ORDER = function () {
  184. return function(clusterIds) {
  185. return clusterIds[0];
  186. };
  187. };