-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(plugin): use amqp10's new plugin behavior
- Loading branch information
Showing
5 changed files
with
126 additions
and
101 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,84 +1,93 @@ | ||
| 'use strict'; | ||
| var hash = require('object-hash'); | ||
| var Promise = require('bluebird'), | ||
| hash = require('object-hash'); | ||
|
|
||
| function LinkCache(client, options) { | ||
| options = options || {}; | ||
| this._client = client; | ||
| this._links = {}; | ||
| this._ttl = options.ttl || 60000; | ||
| this._purgeTimeout = null; | ||
|
|
||
| client.on('disconnected', function() {}); | ||
| } | ||
|
|
||
| LinkCache.prototype.createSender = function(address, options) { | ||
| return this._createLink(address, options, 'sender', 'createSender'); | ||
| }; | ||
|
|
||
| LinkCache.prototype.createReceiver = function(address, options) { | ||
| return this._createLink(address, options, 'receiver', 'createReceiver'); | ||
| }; | ||
|
|
||
| LinkCache.prototype.createSenderStream = function(address, options) { | ||
| return this._createLink(address, options, 'senderStream', 'createSenderStream'); | ||
| }; | ||
| var links = {}; | ||
| var ttl = 60000; | ||
| var purgeTimeout = null; | ||
|
|
||
| LinkCache.prototype.createReceiverStream = function(address, options) { | ||
| return this._createLink(address, options, 'receiverStream', 'createReceiverStream'); | ||
| }; | ||
|
|
||
| LinkCache.prototype._createLink = function(address, options, type, method) { | ||
| function createLink(address, options, type, method) { | ||
| var linkHash = hash({ type: type, address: address, options: options }); | ||
| if (this._links.hasOwnProperty(linkHash)) { | ||
| var entry = this._links[linkHash]; | ||
| if (links.hasOwnProperty(linkHash)) { | ||
| var entry = links[linkHash]; | ||
| if (!entry.hasOwnProperty('link')) | ||
| return entry; | ||
|
|
||
| this._links[linkHash].stamp = Date.now(); | ||
| return Promise.resolve(this._links[linkHash].link); | ||
| links[linkHash].stamp = Date.now(); | ||
| return Promise.resolve(links[linkHash].link); | ||
| } | ||
|
|
||
| var self = this; | ||
| var linkPromise = this._client[method](address, options) | ||
| var linkPromise = method(address, options) | ||
| .then(function(link) { | ||
| link.once('detached', function() { | ||
| if (self._links.hasOwnProperty(linkHash)) | ||
| delete self._links[linkHash]; | ||
| if (links.hasOwnProperty(linkHash)) | ||
| delete links[linkHash]; | ||
| }); | ||
|
|
||
| self._links[linkHash] = { link: link, stamp: Date.now() }; | ||
| if (!self._purgeTimeout) | ||
| self._purgeTimeout = setTimeout(self._purgeLinks.bind(self), self._ttl); | ||
| links[linkHash] = { link: link, stamp: Date.now() }; | ||
| if (!purgeTimeout) | ||
| purgeTimeout = setTimeout(purgeLinks, ttl); | ||
| return link; | ||
| }); | ||
|
|
||
|
|
||
| this._links[linkHash] = linkPromise; | ||
| links[linkHash] = linkPromise; | ||
| return linkPromise; | ||
| }; | ||
| } | ||
|
|
||
| LinkCache.prototype._purgeLinks = function() { | ||
| function purgeLinks() { | ||
| var now = Date.now(); | ||
| var _keys = Object.keys(this._links), | ||
| var _keys = Object.keys(links), | ||
| expired = [], live = 0; | ||
|
|
||
| purgeTimeout = null; | ||
| for (var i = 0, ii = _keys.length; i < ii; ++i) { | ||
| if (now - this._links[_keys[i]].stamp >= this._ttl) { | ||
| if (now - links[_keys[i]].stamp >= ttl) { | ||
| expired.push(_keys[i]); | ||
| } else { | ||
| live++; | ||
| } | ||
| } | ||
|
|
||
| for (var j = 0, jj = expired.length; j < jj; ++j) { | ||
| var cacheEntry = this._links[expired[j]]; | ||
| delete this._links[_keys[j]]; | ||
| var cacheEntry = links[expired[j]]; | ||
| delete links[expired[j]]; | ||
| cacheEntry.link.detach(); | ||
| } | ||
|
|
||
| if (live && !this._purgeTimeout) { | ||
| this._purgeTimeout = setTimeout(this._purgeLinks.bind(this), this._ttl); | ||
| if (live) { | ||
| purgeTimeout = setTimeout(purgeLinks, ttl); | ||
| } | ||
| }; | ||
| } | ||
|
|
||
| module.exports = LinkCache; | ||
| module.exports = function(options) { | ||
| // NOTE: we need to re-initialize these every time the plugin is called | ||
| options = options || {}; | ||
| links = {}; | ||
| ttl = options.ttl || 60000; | ||
| if (!!purgeTimeout) clearTimeout(purgeTimeout); | ||
| purgeTimeout = null; | ||
|
|
||
| return function(Client) { | ||
| var _createSender = Client.prototype.createSender, | ||
| _createReceiver = Client.prototype.createReceiver, | ||
| _createSenderStream = Client.prototype.createSenderStream, | ||
| _createReceiverStream = Client.prototype.createReceiverStream; | ||
|
|
||
| Client.prototype.createSender = function(address, options) { | ||
| return createLink(address, options, 'sender', _createSender.bind(this)); | ||
| }; | ||
|
|
||
| Client.prototype.createReceiver = function(address, options) { | ||
| return createLink(address, options, 'receiver', _createReceiver.bind(this)); | ||
| }; | ||
|
|
||
| Client.prototype.createSenderStream = function(address, options) { | ||
| return createLink(address, options, 'senderStream', _createSenderStream.bind(this)); | ||
| }; | ||
|
|
||
| Client.prototype.createReceiverStream = function(address, options) { | ||
| return createLink(address, options, 'receiverStream', _createReceiverStream.bind(this)); | ||
| }; | ||
| }; | ||
| }; |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,44 @@ | ||
| 'use strict'; | ||
| var amqp = require('amqp10'), | ||
| linkCache = require('..'), | ||
| AMQPClient = amqp.Client, | ||
| config = require('./config'), | ||
| expect = require('chai').expect; | ||
|
|
||
| var test = {}; | ||
| describe('purging', function() { | ||
| before(function() { amqp.use(linkCache({ ttl: 10 })); }); | ||
| beforeEach(function() { | ||
| if (!!test.client) delete test.client; | ||
| test.client = new AMQPClient(); | ||
| }); | ||
|
|
||
| afterEach(function() { | ||
| return test.client.disconnect() | ||
| .then(function() { delete test.client; }); | ||
| }); | ||
|
|
||
| it('should purge links after a given interval', function() { | ||
| var sender; | ||
| return test.client.connect(config.address) | ||
| .then(function() { return test.client.createSender('amq.topic'); }) | ||
| .then(function(s) { sender = s; }) | ||
| .delay(50) | ||
| .then(function() { | ||
| var state = sender.linkSM.getMachineState(); | ||
| expect(state).to.be.oneOf(['DETACHED', 'DETACHING']); | ||
| sender = null; | ||
| }); | ||
| }); | ||
|
|
||
| it('should purge links after a given interval (2)', function() { | ||
| var sender1, sender2; | ||
| return test.client.connect(config.address) | ||
| .then(function() { return test.client.createSender('amq.fanout'); }) | ||
| .tap(function(sender) { sender1 = sender; }) | ||
| .delay(100) | ||
| .then(function() { return test.client.createSender('amq.fanout'); }) | ||
| .tap(function(sender) { sender2 = sender; }) | ||
| .then(function() { expect(sender1).to.not.eql(sender2); }); | ||
| }); | ||
| }); |