Merge pull request #128 from kibertoad/feat/stop-restart-refresh

Expose stopRefresh method

Author: swarthy

README.md

@@ -107,6 +107,27 @@ async function doSomething() {
 }
 ```
 
+#### Example with temporary refresh
+
+```javascript
+async function doSomething() {
+ const mutex = new Mutex(redisClient, 'lockingResource', {
+ lockTimeout: 120000,
+ refreshInterval: 15000
+ })
+ const lockAcquired = await mutex.tryAcquire()
+ if (!lockAcquired) {
+ return
+ }
+ try {
+ // critical cycle iteration
+ } finally {
+ // We want to let lock expire over time after operation is finished 
+ await mutex.stopRefresh()
+ }
+}
+```
+
 ### Semaphore
 
 > See [RedisLabs: Basic counting sempahore](https://redislabs.com/ebook/part-2-core-concepts/chapter-6-application-components-in-redis/6-3-counting-semaphores/6-3-1-building-a-basic-counting-semaphore/)

src/Lock.ts

@@ -62,7 +62,7 @@ export abstract class Lock {
  this._refreshInterval.unref()
  }
 
- private _stopRefresh() {
+ stopRefresh() {
  if (this._refreshInterval) {
  debug(
  `clear refresh interval ${this._kind} (key: ${this._key}, identifier: ${this._identifier})`
@@ -86,7 +86,7 @@ export abstract class Lock {
  const refreshed = await this._refresh()
  if (!refreshed) {
  this._acquired = false
- this._stopRefresh()
+ this.stopRefresh()
  const lockLostError = new LostLockError(
  `Lost ${this._kind} for key ${this._key}`
  )
@@ -127,7 +127,7 @@ export abstract class Lock {
  `release ${this._kind} (key: ${this._key}, identifier: ${this._identifier})`
  )
  if (this._refreshTimeInterval > 0) {
- this._stopRefresh()
+ this.stopRefresh()
  }
  if (this._acquired) {
  await this._release()

test/src/RedisMultiSemaphore.test.ts

@@ -87,7 +87,7 @@ describe('MultiSemaphore', () => {
  const semaphore2 = new MultiSemaphore(client, 'key', 3, 1, timeoutOptions)
  await semaphore1.acquire()
  await semaphore2.acquire()
- await delay(100)
+ await delay(400)
  expect(await client.zrange('semaphore:key', 0, -1)).to.have.members([
  semaphore1.identifier + '_0',
  semaphore1.identifier + '_1',
@@ -100,6 +100,20 @@ describe('MultiSemaphore', () => {
  await semaphore2.release()
  expect(await client.zcard('semaphore:key')).to.be.eql(0)
  })
+ it('should stop refreshing lock if stopped', async () => {
+ const semaphore1 = new MultiSemaphore(client, 'key', 3, 2, timeoutOptions)
+ const semaphore2 = new MultiSemaphore(client, 'key', 3, 1, timeoutOptions)
+ await semaphore1.acquire()
+ await semaphore2.acquire()
+ await semaphore1.stopRefresh()
+ await delay(400)
+ expect(await client.zrange('semaphore:key', 0, -1)).to.be.eql([
+ semaphore2.identifier + '_0'
+ ])
+ await semaphore2.stopRefresh()
+ await delay(400)
+ expect(await client.zcard('semaphore:key')).to.be.eql(0)
+ })
  it('should acquire maximum LIMIT semaphores', async () => {
  const s = () =>
  new MultiSemaphore(client, 'key', 3, 1, {

test/src/RedisMutex.test.ts

@@ -71,16 +71,23 @@ describe('Mutex', () => {
  it('should refresh lock every refreshInterval ms until release', async () => {
  const mutex = new Mutex(client, 'key', timeoutOptions)
  await mutex.acquire()
- await delay(100)
+ await delay(400)
  expect(await client.get('mutex:key')).to.be.eql(mutex.identifier)
  await mutex.release()
  expect(await client.get('mutex:key')).to.be.eql(null)
  })
+ it('should stop refreshing lock every refreshInterval ms if stopped', async () => {
+ const mutex = new Mutex(client, 'key', timeoutOptions)
+ await mutex.acquire()
+ mutex.stopRefresh()
+ await delay(400)
+ expect(await client.get('mutex:key')).to.be.eql(null)
+ })
  it('should re-acquire lock if lock was expired between refreshes, but was not acquired by another instance', async () => {
  const mutex = new Mutex(client, 'key', timeoutOptions)
  await mutex.acquire()
  await client.del('mutex:key') // "expired"
- await delay(100)
+ await delay(400)
  expect(await client.get('mutex:key')).to.be.eql(mutex.identifier)
  await mutex.release()
  expect(await client.get('mutex:key')).to.be.eql(null)

test/src/RedisSemaphore.test.ts

@@ -77,7 +77,7 @@ describe('Semaphore', () => {
  const semaphore2 = new Semaphore(client, 'key', 2, timeoutOptions)
  await semaphore1.acquire()
  await semaphore2.acquire()
- await delay(100)
+ await delay(400)
  expect(await client.zrange('semaphore:key', 0, -1)).to.have.members([
  semaphore1.identifier,
  semaphore2.identifier
@@ -89,6 +89,20 @@ describe('Semaphore', () => {
  await semaphore2.release()
  expect(await client.zcard('semaphore:key')).to.be.eql(0)
  })
+ it('should stop refreshing lock if stopped', async () => {
+ const semaphore1 = new Semaphore(client, 'key', 2, timeoutOptions)
+ const semaphore2 = new Semaphore(client, 'key', 2, timeoutOptions)
+ await semaphore1.acquire()
+ await semaphore2.acquire()
+ await semaphore1.stopRefresh()
+ await delay(400)
+ expect(await client.zrange('semaphore:key', 0, -1)).to.be.eql([
+ semaphore2.identifier
+ ])
+ await semaphore2.stopRefresh()
+ await delay(400)
+ expect(await client.zcard('semaphore:key')).to.be.eql(0)
+ })
  it('should acquire maximum LIMIT semaphores', async () => {
  const s = () =>
  new Semaphore(client, 'key', 3, {

test/src/RedlockMultiSemaphore.test.ts

@@ -151,7 +151,7 @@ describe('RedlockMultiSemaphore', () => {
  )
  await semaphore1.acquire()
  await semaphore2.acquire()
- await delay(100)
+ await delay(400)
  await expectZRangeAllHaveMembers('semaphore:key', [
  semaphore1.identifier + '_0',
  semaphore1.identifier + '_1',
@@ -162,6 +162,30 @@ describe('RedlockMultiSemaphore', () => {
  await semaphore2.release()
  await expectZCardAllEql('semaphore:key', 0)
  })
+ it('should stop refreshing lock if stopped', async () => {
+ const semaphore1 = new RedlockMultiSemaphore(
+ allClients,
+ 'key',
+ 3,
+ 2,
+ timeoutOptions
+ )
+ const semaphore2 = new RedlockMultiSemaphore(
+ allClients,
+ 'key',
+ 3,
+ 1,
+ timeoutOptions
+ )
+ await semaphore1.acquire()
+ await semaphore2.acquire()
+ semaphore1.stopRefresh()
+ await delay(400)
+ await expectZRangeAllEql('semaphore:key', [semaphore2.identifier + '_0'])
+ semaphore2.stopRefresh()
+ await delay(400)
+ await expectZCardAllEql('semaphore:key', 0)
+ })
  it('should acquire maximum LIMIT semaphores', async () => {
  const s = () =>
  new RedlockMultiSemaphore(allClients, 'key', 3, 1, {

test/src/RedlockMutex.test.ts

@@ -63,11 +63,18 @@ describe('RedlockMutex', () => {
  it('should refresh lock every refreshInterval ms until release', async () => {
  const mutex = new RedlockMutex(allClients, 'key', timeoutOptions)
  await mutex.acquire()
- await delay(100)
+ await delay(400)
  await expectGetAll('mutex:key', mutex.identifier)
  await mutex.release()
  await expectGetAll('mutex:key', null)
  })
+ it('should stop refreshing if stopped', async () => {
+ const mutex = new RedlockMutex(allClients, 'key', timeoutOptions)
+ await mutex.acquire()
+ mutex.stopRefresh()
+ await delay(400)
+ await expectGetAll('mutex:key', null)
+ })
  describe('lost lock case', () => {
  beforeEach(() => {
  catchUnhandledRejection()

test/src/RedlockSemaphore.test.ts

@@ -118,7 +118,7 @@ describe('RedlockSemaphore', () => {
  )
  await semaphore1.acquire()
  await semaphore2.acquire()
- await delay(100)
+ await delay(400)
  await expectZRangeAllHaveMembers('semaphore:key', [
  semaphore1.identifier,
  semaphore2.identifier
@@ -128,6 +128,28 @@ describe('RedlockSemaphore', () => {
  await semaphore2.release()
  await expectZCardAllEql('semaphore:key', 0)
  })
+ it('should stop refreshing lock if stopped', async () => {
+ const semaphore1 = new RedlockSemaphore(
+ allClients,
+ 'key',
+ 2,
+ timeoutOptions
+ )
+ const semaphore2 = new RedlockSemaphore(
+ allClients,
+ 'key',
+ 2,
+ timeoutOptions
+ )
+ await semaphore1.acquire()
+ await semaphore2.acquire()
+ semaphore1.stopRefresh()
+ await delay(400)
+ await expectZRangeAllEql('semaphore:key', [semaphore2.identifier])
+ semaphore2.stopRefresh()
+ await delay(400)
+ await expectZCardAllEql('semaphore:key', 0)
+ })
  it('should acquire maximum LIMIT semaphores', async () => {
  const s = () =>
  new RedlockSemaphore(allClients, 'key', 3, {