-
Notifications
You must be signed in to change notification settings - Fork 8
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #190 from sphereio/task-repeater
Add default task repeater
- Loading branch information
Showing
11 changed files
with
237 additions
and
11 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
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,77 @@ | ||
| { Repeater } = require 'sphere-node-utils' | ||
| Promise = require 'bluebird' | ||
| TaskQueue = require './task-queue' | ||
| _ = require 'underscore' | ||
|
|
||
| # Response messages which will be handled by `RepeaterTaskQueue` | ||
| retryKeywords = [ | ||
| 'ETIMEDOUT' | ||
| 'socket hang up' | ||
| 'write EPROTO' | ||
| 'Retry later' | ||
| 'I am the LoadBalancer of' | ||
| 'Gateway Timeout' | ||
| 'Bad Gateway' | ||
| 'EAI_AGAIN' | ||
| 'ESOCKETTIMEDOUT' | ||
| 'Oops. This shouldn\'t happen' | ||
| 'InternalServerError: Undefined' | ||
| 'Temporary overloading' | ||
| 'read ECONNRESET' | ||
| 'getaddrinfo ENOTFOUND' | ||
| 'Cannot commit on stream id' | ||
| ] | ||
|
|
||
| # Public: A `RepeaterTaskQueue` adds request repeater on particular response errors | ||
| # | ||
| # `RepeaterTaskQueue` receives two objects as parameter. | ||
| # First object overriding `maxParallel` value of `TaskQueue` | ||
| # Second object contains information about the count of attempts, timeout, timeout type and retry keywords. | ||
| # There are 2 types of `timeoutType`: | ||
| # - `c`: constant delay | ||
| # - `v`: variable delay (grows with attempts count with a random component) | ||
| # It's possible to customize default list of handled error messages. To do this just pass new array to `retryKeywords` | ||
| # | ||
| # Examples | ||
| # | ||
| # task = new RepeaterTaskQueue { maxParallel: 30 }, { attempts: 50, timeout: 200, timeoutType: 'v', retryKeywords: ['test1', 'test2'] } | ||
| class RepeaterTaskQueue extends TaskQueue | ||
|
|
||
|
|
||
| constructor: (options, repeaterOptions) -> | ||
| super options | ||
| repeaterOptions = _.defaults repeaterOptions, | ||
| attempts: 50 | ||
| timeout: 200 | ||
| timeoutType: 'v' | ||
| retryKeywords: retryKeywords | ||
| @repeaterOptions = repeaterOptions | ||
|
|
||
|
|
||
| _startTask: (task) => | ||
| @_activeCount += 1 | ||
| repeater = new Repeater(@repeaterOptions) | ||
| toRepeat = repeater.execute task.fn, (err) => | ||
| if @_shouldRetry(err) | ||
| return Promise.resolve() | ||
| else | ||
| Promise.reject err | ||
| toRepeat.then (res) -> | ||
| task.resolve res | ||
| return | ||
| .catch (err) -> | ||
| task.reject err | ||
| return | ||
| .finally => | ||
| @_activeCount -= 1 | ||
| @_maybeExecute() | ||
| .done() | ||
|
|
||
|
|
||
| _shouldRetry: (error) -> | ||
| return error?.code?.toString().startsWith('5') or | ||
| error?.statusCode?.toString().startsWith('5') or | ||
| @repeaterOptions.retryKeywords.some (keyword) -> | ||
| JSON.stringify(error).toUpperCase().includes(keyword.toUpperCase()) | ||
|
|
||
| module.exports = RepeaterTaskQueue |
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
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
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,114 @@ | ||
| SphereClient = require '../lib/client' | ||
| RepeaterTaskQueue = require '../lib/repeater-task-queue' | ||
| Config = require('../config').config | ||
| util = require('util') | ||
| _ = require 'underscore' | ||
| describe 'RepeaterTaskQueue', -> | ||
|
|
||
| repeaterOptions = { | ||
| attempts: 5, | ||
| timeout: 200 | ||
| } | ||
|
|
||
| beforeEach -> | ||
| task = new RepeaterTaskQueue { maxParallel: 30 }, { attempts: 5, timeout: 200, timeoutType: 'v' } | ||
| sphereConfig = { config: Config, task } | ||
| @client = new SphereClient sphereConfig | ||
|
|
||
|
|
||
| it 'should finally resolve after four tries', (done) -> | ||
|
|
||
| callsMap = { | ||
| 0: { statusCode: 500, message: 'ETIMEDOUT' } | ||
| 1: { code: 500, message: 'ETIMEDOUT' } | ||
| 2: { code: 789, message: 'ETIMEDOUT' } | ||
| 3: { statusCode: 200, message: 'success' } | ||
| } | ||
| callCount = 0 | ||
| spyOn(@client._rest, 'GET').andCallFake (resource, callback) -> | ||
| currentCall = callsMap[callCount] | ||
| callCount++ | ||
| statusCode = currentCall.statusCode | ||
| message = currentCall.message | ||
| callback(null, { statusCode }, { message }) | ||
|
|
||
| @client.products.fetch() | ||
| .then (res) => | ||
| expect(@client._rest.GET.calls.length).toEqual 4 | ||
| expect(res.body.message).toEqual 'success' | ||
| done() | ||
| .catch (err) -> done(err) | ||
|
|
||
|
|
||
| it 'should repeat requests several times for 5xx error codes', (done) -> | ||
| spyOn(@client._rest, 'GET').andCallFake (resource, callback) -> | ||
| callback(null, { statusCode: 500 }, { code: 500 }) | ||
|
|
||
| @client.products.fetch() | ||
| .then -> | ||
| done 'Error expected' | ||
| .catch => | ||
| expect(@client._rest.GET.calls.length).toEqual repeaterOptions.attempts | ||
| done() | ||
|
|
||
|
|
||
| it 'should repeat requests several times for certain error messages', (done) -> | ||
| spyOn(@client._rest, 'GET').andCallFake (resource, callback) -> | ||
| callback(null, { statusCode: 500 }, { message: 'ETIMEDOUT' }) | ||
|
|
||
| @client.products.fetch() | ||
| .then -> | ||
| done 'Error expected' | ||
| .catch => | ||
| expect(@client._rest.GET.calls.length).toEqual repeaterOptions.attempts | ||
| done() | ||
|
|
||
|
|
||
| it 'should not repeat requests for non-5xx errors', (done) -> | ||
| spyOn(@client._rest, 'GET').andCallFake (resource, callback) -> | ||
| callback(null, { statusCode: 400 }, { message: 'something' }) | ||
|
|
||
| @client.products.fetch() | ||
| .then -> | ||
| done 'Error expected' | ||
| .catch (err) => | ||
| expect(err.message).toEqual 'something' | ||
| expect(@client._rest.GET.calls.length).toEqual 1 | ||
| done() | ||
|
|
||
| it 'should set default repeater options on empty object', (done) -> | ||
| retryKeywords = [ | ||
| 'ETIMEDOUT' | ||
| 'socket hang up' | ||
| 'write EPROTO' | ||
| 'Retry later' | ||
| 'I am the LoadBalancer of' | ||
| 'Gateway Timeout' | ||
| 'Bad Gateway' | ||
| 'EAI_AGAIN' | ||
| 'ESOCKETTIMEDOUT' | ||
| 'Oops. This shouldn\'t happen' | ||
| 'InternalServerError: Undefined' | ||
| 'Temporary overloading' | ||
| 'read ECONNRESET' | ||
| 'getaddrinfo ENOTFOUND' | ||
| 'Cannot commit on stream id' | ||
| ] | ||
| task = new RepeaterTaskQueue {}, {} | ||
| expect(_.isEqual(task.repeaterOptions.retryKeywords, retryKeywords)).toEqual true | ||
| expect(task.repeaterOptions.attempts).toEqual 50 | ||
| expect(task.repeaterOptions.timeout).toEqual 200 | ||
| expect(task.repeaterOptions.timeoutType).toEqual 'v' | ||
| done() | ||
|
|
||
| it 'should set custom repeater options', (done) -> | ||
| retryKeywords = [ | ||
| 'test1' | ||
| 'test2' | ||
| ] | ||
| task = new RepeaterTaskQueue {}, { attempts: 7, timeout: 211, timeoutType: 'c', retryKeywords: retryKeywords} | ||
| expect(_.isEqual(task.repeaterOptions.retryKeywords, retryKeywords)).toEqual true | ||
| expect(task.repeaterOptions.attempts).toEqual 7 | ||
| expect(task.repeaterOptions.timeout).toEqual 211 | ||
| expect(task.repeaterOptions.timeoutType).toEqual 'c' | ||
| done() |