Working operator

.babelrc.js

@@ -0,0 +1,16 @@
+const { NODE_ENV } = process.env
+const test = NODE_ENV === 'test'
+const loose = true
+
+module.exports = {
+  presets: [
+    [
+      '@babel/env',
+      {
+        loose,
+        modules: false,
+      },
+    ],
+  ],
+  plugins: [test && '@babel/transform-modules-commonjs'].filter(Boolean),
+}

.gitignore

@@ -0,0 +1,2 @@
+node_modules
+dist

.huskyrc

@@ -0,0 +1,5 @@
+{
+  "hooks": {
+    "pre-commit": "lint-staged"
+  }
+}

.lintstagedrc

@@ -0,0 +1,6 @@
+{
+  "*.{js,md}": [
+    "prettier --write",
+    "git add"
+  ]
+}

.npmrc

@@ -0,0 +1 @@
+package-lock=false

.prettierrc

@@ -0,0 +1,5 @@
+{
+  "semi": false,
+  "singleQuote": true,
+  "trailingComma": "all"
+}

.travis.yml

@@ -0,0 +1,4 @@
+language: node_js
+
+node_js:
+  - "node"

README.md

@@ -0,0 +1,28 @@
+# callbag-retry
+
+Callbag operator which resubscribes to the source given amount of times.
+
+## Example
+
+```js
+import concat from 'callbag-concat'
+import fromIter from 'callbag-from-iter'
+import pipe from 'callbag-pipe'
+import retry from 'callbag-retry'
+import subscribe from 'callbag-subscribe'
+import throwError from 'callbag-throw-error'
+
+pipe(
+  concat(fromIter([1, 2, 3]), throwError(new Error('Test error.'))),
+  retry(3),
+  subscribe({
+    next: v => {
+      // will log 1, 2, 3, 1, 2, 3, 1, 2, 3, 1, 2, 3
+      console.log(v)
+    },
+    error(err) {
+      // errors with 4th `Test error.` error
+    },
+  }),
+)
+```

__tests__/index.js

@@ -0,0 +1,45 @@
+import concat from 'callbag-concat'
+import fromIter from 'callbag-from-iter'
+import pipe from 'callbag-pipe'
+import subscribe from 'callbag-subscribe'
+import throwError from 'callbag-throw-error'
+
+import retry from '../src'
+
+test('retries correct amount of times and then errors', () => {
+  const actual = []
+
+  pipe(
+    concat(fromIter([1, 2, 3]), throwError(new Error('Test error.'))),
+    retry(3),
+    subscribe({
+      next: v => actual.push(v),
+      error(err) {
+        expect(actual).toEqual([1, 2, 3, 1, 2, 3, 1, 2, 3, 1, 2, 3])
+        expect(err.message).toBe('Test error.')
+      },
+    }),
+  )
+})
+
+test('keeps retries counter per subscription', () => {
+  const retry2 = retry(2)
+
+  const run = () => {
+    const actual = []
+    pipe(
+      concat(fromIter([1, 2]), throwError(new Error('Test error.'))),
+      retry2,
+      subscribe({
+        next: v => actual.push(v),
+        error(err) {
+          expect(actual).toEqual([1, 2, 1, 2, 1, 2])
+          expect(err.message).toBe('Test error.')
+        },
+      }),
+    )
+  }
+
+  run()
+  run()
+})

package.json

@@ -0,0 +1,49 @@
+{
+  "name": "callbag-retry",
+  "version": "0.0.0",
+  "description": "👜 Callbag operator which resubscribes to the source given amount of times.",
+  "main": "./dist/callbag-retry.cjs.js",
+  "module": "./dist/callbag-retry.esm.js",
+  "files": [
+    "dist"
+  ],
+  "scripts": {
+    "test": "jest --env=node",
+    "prebuild": "rimraf dist",
+    "build": "rollup -c",
+    "preversion": "npm test",
+    "prepare": "npm run build"
+  },
+  "repository": {
+    "type": "git",
+    "url": "git+https://github.com/Andarist/callbag-retry.git"
+  },
+  "keywords": [
+    "callbag",
+    "callbags"
+  ],
+  "author": "Mateusz Burzyński <[email protected]> (https://github.com/Andarist)",
+  "license": "MIT",
+  "bugs": {
+    "url": "https://github.com/Andarist/callbag-retry/issues"
+  },
+  "homepage": "https://github.com/Andarist/callbag-retry#readme",
+  "devDependencies": {
+    "@babel/core": "^7.1.5",
+    "@babel/plugin-transform-modules-commonjs": "^7.1.0",
+    "@babel/preset-env": "^7.1.5",
+    "babel-core": "^7.0.0-bridge.0",
+    "callbag-concat": "^1.1.0",
+    "callbag-from-iter": "^1.2.0",
+    "callbag-pipe": "^1.1.1",
+    "callbag-subscribe": "^1.5.0",
+    "callbag-throw-error": "^1.0.1",
+    "husky": "^1.1.3",
+    "jest": "^23.6.0",
+    "lint-staged": "^8.0.4",
+    "prettier": "^1.15.2",
+    "rimraf": "^2.6.2",
+    "rollup": "^0.67.0",
+    "rollup-plugin-babel": "^4.0.3"
+  }
+}

rollup.config.js

@@ -0,0 +1,15 @@
+import babel from 'rollup-plugin-babel'
+import pkg from './package.json'
+
+export default {
+  input: 'src/index.js',
+  output: [
+    { file: pkg.main, format: 'cjs', exports: 'named' },
+    { file: pkg.module, format: 'esm' },
+  ],
+  external: [
+    ...Object.keys(pkg.dependencies || {}),
+    ...Object.keys(pkg.peerDependencies || {}),
+  ],
+  plugins: [babel()],
+}

src/index.js

@@ -0,0 +1,40 @@
+export default function retry(max = -1) {
+  return source => (start, sink) => {
+    if (start !== 0) return
+    let inited = false
+    let retries = max
+    let sourceTalkback
+    let talkback = (type, data) => {
+      sourceTalkback(type, data)
+    }
+    const subscribe = () => {
+      source(0, (type, data) => {
+        if (type === 0) {
+          sourceTalkback = data
+
+          if (inited) {
+            // this pull probably shouldn't be unconditional
+            // but I'm not sure how to track if we should pull or not at the moment
+            talkback(1)
+            return
+          }
+
+          inited = true
+          sink(0, talkback)
+          return
+        }
+
+        if (type === 2 && data) {
+          if (retries !== 0) {
+            retries--
+            subscribe()
+            return
+          }
+        }
+
+        sink(type, data)
+      })
+    }
+    subscribe()
+  }
+}