New backoff and batch implementations (#12)

* New backoff and batch implementations

* Lint picking

* Add spy

* Downgrade joi to be a peer dependency instead

* Limit max backoff tries to 10

* Share promises with dupe args to only batch uniqs

* Test that validate is safe if joi is missing

API.md

@@ -4,10 +4,11 @@
 | -------- | --------- |
 | [`all`](#all) | `[Promise a] -> Promise [a]` |
 | [`assemble`](#assemble) | `{ k: (v -> v) } -> v -> { k: v }` |
-| [`assembleP`](#assembleP) | `{ k: (v -> Promise v) } -> v -> Promise { k: v }` |
-| [`assocWith`](#assocWith) | `String -> ({ k: v } -> a) -> { k: v } -> { k: v }` |
-| [`assocWithP`](#assocWithP) | `String -> ({ k: v } -> Promise a) -> { k: v } -> Promise { k: v }` |
-| [`backoff`](#backoff) | `Number -> Number -> (a... -> Promise b) -> a... -> Promise b` |
+| [`assembleP`](#assemblep) | `{ k: (v -> Promise v) } -> v -> Promise { k: v }` |
+| [`assocWith`](#assocwith) | `String -> ({ k: v } -> a) -> { k: v } -> { k: v }` |
+| [`assocWithP`](#assocwithp) | `String -> ({ k: v } -> Promise a) -> { k: v } -> Promise { k: v }` |
+| [`backoff`](#backoff) | `{ k: v } -> (a... -> Promise b) -> a... -> Promise b` |
+| [`batch`](#batch) | `{ k: v } -> ([a] -> Promise [b]) -> a -> Promise b` |
 | [`combine`](#combine) | `({ k: v } -> { k: v }) -> { k: v } -> { k: v }` |
 | [`combineAll`](#combineall) | `[({ k: v }, ...) -> { k: v }] -> ({ k: v }, ...) -> { k: v }` |
 | [`combineAllP`](#combineallp) | `[({ k: v }, ...) -> Promise { k: v }] -> ({ k: v }, ...) -> Promise { k: v }` |
@@ -106,15 +107,62 @@ assocWithP('foo', always(Promise.resolve('bar'), {})) //=> Promise { foo: 'bar'
 `@articulate/funky/lib/backoff`
 
 ```haskell
-backoff :: Number -> Number -> (a... -> Promise b) -> a... -> Promise b
+backoff :: { k: v } -> (a... -> Promise b) -> a... -> Promise b
 ```
 
-Accepts a `base` delay in ms and max `tries`, and then wraps an async function with a [full jitter exponential backoff](https://www.awsarchitectureblog.com/2015/03/backoff.html) algorithm.  Useful for recovering from intermittent network failures.  Will retry for all caught errors until the number of `tries` is reached.
+| Option | Type | Default | Description |
+| ------ | ---- | ------- | ----------- |
+| `base` | `Number` | `250` | base delay in ms |
+| `tries` | `Number` | `10` | max number of tries |
+| `when` | `a -> Boolean` | `R.T` | only backoff if this returns true |
+
+Accepts an options object, and then wraps an async function with a [full jitter exponential backoff](https://www.awsarchitectureblog.com/2015/03/backoff.html) algorithm.  Useful for recovering from intermittent network failures.  Will retry for caught errors that pass the `when` predicate until the number of `tries` is reached.
+
+```js
+const { propEq } = require('ramda')
+
+const fetchImage = data => { /* async, and might fail sometimes */ }
+
+const opts = {
+  base: 500,
+  tries: 5,
+  when: propEq('statusCode', 429)
+}
+
+backoff(opts, fetchImage)
+//=> a new function that tries at most 5 times before rejecting if the error is `429 Too Many Requests`
+```
+
+### batch
+
+`@articulate/funky/lib/batch`
+
+```haskell
+batch :: { k: v } -> ([a] -> Promise [b]) -> a -> Promise b
+```
+
+| Option | Type | Default | Description |
+| ------ | ---- | ------- | ----------- |
+| `limit` | `Number` | `Infinity` | max length of each batch |
+| `wait` | `Number` | `32` | max wait before throttling batches |
+
+Accepts an options object, and then wraps a batched async function.  Returns a throttled, unary async function that batches the args of successive invocations and resolves each individual promise with the matching result.  Useful for cutting down IO by combining requests.
 
 ```js
-const fetchImage = opts => { /* async, and might fail sometimes */ }
+const { batch, evolveP } = require('@articulate/funky')
+const { composeP, prop } = require('ramda')
 
-backoff(250, 5, fetchImage) //=> a new function that tries at most 5 times before rejecting
+const createMedia = batch({ limit: 128 }, require('../data/createMedia'))
+
+const asset = composeP(prop('id'), createMedia)
+
+const convertMedia = evolveP({
+  audio: asset,
+  image: asset,
+  video: asset
+})
+// will create new media records for each entry in the object by batching
+// them in a single request, and then store the new ids on the result object
 ```
 
 ### combine
@@ -421,6 +469,8 @@ validate :: Schema -> a -> Promise a
 
 Validates a value against a [`Joi`](https://github.com/hapijs/joi) schema.  Curried and promisified for ease of use.
 
+**Note:** For validation to work, requires [`Joi`](https://github.com/hapijs/joi) to be installed as a dependency of the consuming application.
+
 ```js
 const schema = Joi.object({
   id: Joi.string().required()

index.js

@@ -4,6 +4,7 @@ exports.assembleP    = require('./lib/assembleP')
 exports.assocWith    = require('./lib/assocWith')
 exports.assocWithP   = require('./lib/assocWithP')
 exports.backoff      = require('./lib/backoff')
+exports.batch        = require('./lib/batch')
 exports.combine      = require('./lib/combine')
 exports.combineAll   = require('./lib/combineAll')
 exports.combineAllP  = require('./lib/combineAllP')

package.json

@@ -28,7 +28,6 @@
     "test:coverage": "nyc yarn run test"
   },
   "dependencies": {
-    "joi": "^10.6.0",
     "ramda": "^0.25.0"
   },
   "devDependencies": {
@@ -37,8 +36,13 @@
     "chai": "^4.1.1",
     "coveralls": "^2.13.1",
     "eslint": "^4.3.0",
+    "joi": "10.x",
     "mocha": "^3.5.0",
     "nyc": "^11.1.0",
-    "prop-factory": "^1.0.0"
+    "prop-factory": "^1.0.0",
+    "proxyquire": "^2.0.1"
+  },
+  "peerDependencies": {
+    "joi": "10.x"
   }
 }

src/backoff.js

@@ -1,28 +1,42 @@
-const apply = require('ramda/src/apply')
-const curry = require('ramda/src/curry')
+const apply  = require('ramda/src/apply')
+const curry  = require('ramda/src/curry')
+const ifElse = require('ramda/src/ifElse')
+const T      = require('ramda/src/T')
 
-// backoff :: Number -> Number -> (a... -> Promise b) -> a... -> Promise b
-const backoff = (base, tries, f) =>
-  (...args) => {
+const reject = Promise.reject.bind(Promise)
+
+// backoff :: { k: v } -> (a... -> Promise b) -> a... -> Promise b
+const backoff = (opts={}, f) => {
+  const {
+    base  = 250,
+    tries = 10,
+    when  = T
+  } = opts
+
+  const backedOff = (...args) => {
     let attempt = 0
 
     const retry = () =>
-      new Promise((resolve, reject) => {
+      new Promise((res, rej) => {
         setTimeout(() => {
-          Promise.resolve(args)
-            .then(apply(f))
-            .catch(tryAgain)
-            .then(resolve)
-            .catch(reject)
+          run().then(res, rej)
         }, delay(base, attempt))
       })
 
+    const run = () =>
+      Promise.resolve(args)
+        .then(apply(f))
+        .catch(ifElse(when, tryAgain, reject))
+
     const tryAgain = err =>
-      ++attempt < tries ? retry() : Promise.reject(err)
+      ++attempt < tries ? retry() : reject(err)
 
-    return retry()
+    return run()
   }
 
+  return backedOff
+}
+
 const delay = (base, attempt) =>
   attempt && randBetween(0, base * Math.pow(2, attempt))
 

src/batch.js

@@ -0,0 +1,62 @@
+const call    = require('ramda/src/call')
+const curry   = require('ramda/src/curry')
+const juxt    = require('ramda/src/juxt')
+const max     = require('ramda/src/max')
+const zipWith = require('ramda/src/zipWith')
+
+// batch :: { k: v } -> ([a] -> Promise [b]) -> a -> Promise b
+const batch = (opts={}, f) => {
+  const {
+    limit = Infinity,
+    wait  = 32
+  } = opts
+
+  let args     = []
+  let last     = 0
+  let rejects  = []
+  let resolves = []
+  let timeout  = 0
+  let uniq     = new Map()
+
+  const batched = arg => {
+    if (uniq.has(arg)) {
+      return uniq.get(arg)
+    } else {
+      const promise = new Promise((res, rej) => {
+        args.push(arg)
+        rejects.push(rej)
+        resolves.push(res)
+      })
+
+      uniq.set(arg, promise)
+
+      if (args.length >= limit) run()
+
+      else if (!timeout) {
+        const delta = new Date() - last
+        timeout = setTimeout(run, max(0, wait - delta))
+      }
+
+      return promise
+    }
+  }
+
+  const run = () => {
+    Promise.resolve(args)
+      .then(f)
+      .then(zipWith(call, resolves))
+      .catch(juxt(rejects))
+
+    clearTimeout(timeout)
+    args     = []
+    last     = +new Date()
+    rejects  = []
+    resolves = []
+    timeout  = 0
+    uniq.clear()
+  }
+
+  return batched
+}
+
+module.exports = curry(batch)

src/validate.js

@@ -1,14 +1,21 @@
 const curryN = require('ramda/src/curryN')
-const Joi    = require('joi')
 
-const promisify = require('./promisify')
+let validate
 
-const defaults = { abortEarly: false }
+try {
+  const Joi       = require('joi')
+  const promisify = require('./promisify')
+  const _validate = promisify(Joi.validate, Joi)
+  const defaults  = { abortEarly: false }
 
-const _validate = promisify(Joi.validate, Joi)
+  validate = (schema, x, opts=defaults) =>
+    _validate(x, schema, opts)
+}
 
-// validate :: Schema -> a -> Promise a
-const validate = (schema, x, opts=defaults) =>
-  _validate(x, schema, opts)
+catch (e) {
+  validate = (_, x) =>
+    Promise.resolve(x)
+}
 
+// validate :: Schema -> a -> Promise a
 module.exports = curryN(2, validate)

test/backoff.js

@@ -1,4 +1,5 @@
 const { expect } = require('chai')
+const { gte }    = require('ramda')
 const property   = require('prop-factory')
 
 const { backoff } = require('..')
@@ -11,13 +12,29 @@ const fails = times => {
       : Promise.resolve(count)
 }
 
-const sureThing   = backoff(16, 5, fails(0))
-const failsThrice = backoff(16, 5, fails(3))
-const badDay      = backoff(16, 5, fails(10))
+const opts = { base: 16, tries: 5 }
+
+const sureThing   = backoff(opts, fails(0))
+const failsThrice = backoff(opts, fails(3))
+const badDay      = backoff(opts, fails(10))
+
+const conditional = backoff({ base: 16, tries: 5, when: gte(2) }, fails(10))
 
 describe('backoff', () => {
   const res = property()
 
+  describe('with no options', () => {
+    const defaulted = backoff(undefined, fails(1))
+
+    beforeEach(() =>
+      defaulted('a').then(res)
+    )
+
+    it('has safe defaults', () =>
+      expect(res()).to.equal(1)
+    )
+  })
+
   describe('when function succeeds', () => {
     beforeEach(() =>
       sureThing('a').then(res)
@@ -47,4 +64,14 @@ describe('backoff', () => {
       expect(res()).to.equal(5)
     )
   })
+
+  describe('when backoff predicate supplied', () => {
+    beforeEach(() =>
+      conditional('a').catch(res)
+    )
+
+    it('retries only when predicate passes', () =>
+      expect(res()).to.equal(3)
+    )
+  })
 })