{"id":1138,"date":"2018-06-18T16:47:28","date_gmt":"2018-06-18T15:47:28","guid":{"rendered":"https:\/\/solidt.eu\/site\/?p=1138"},"modified":"2025-12-16T16:48:24","modified_gmt":"2025-12-16T15:48:24","slug":"callbags","status":"publish","type":"post","link":"https:\/\/solidt.eu\/site\/callbags\/","title":{"rendered":"Callbags"},"content":{"rendered":"\n

https:\/\/loreanvictor.github.io\/callbag-common\/what-are-callbags<\/a><\/p>\n\n\n\n

https:\/\/github.com\/callbag\/callbag<\/a><\/p>\n\n\n\n

https:\/\/github.com\/staltz\/callbag-basics<\/a><\/p>\n\n\n\n

https:\/\/github.com\/staltz\/callbag-pseudo-rxjs<\/a><\/p>\n\n\n\n

\/\/from https:\/\/www.npmjs.com\/package\/callbag-catch-error\nconst catchError = fn => source => (start, sink) => {\n  if (start !== 0) return;\n  source(0, (type, data) => {\n    type === 2 && typeof data !== 'undefined' ? fn(data) : sink(type, data);\n  });\n};\n\nfunction* range(from, to) {\n  let i = from;\n  while (i <= to) {\n    yield i;\n    i++;\n  }\n}\n\nconst fromEvent = (node, name) => (start, sink) => {\n  if (start !== 0) return;\n  const handler = ev => sink(1, ev);\n  sink(0, t => {\n    if (t === 2) node.removeEventListener(name, handler);\n  });\n  node.addEventListener(name, handler);\n};\n\nconst fromIter = iter => (start, sink) => {\n    if (start !== 0) return;\n    const iterator =\n        typeof Symbol !== 'undefined' && iter[Symbol.iterator] ?\n        iter[Symbol.iterator]() :\n        iter;\n    let inloop = false;\n    let got1 = false;\n    let res;\n\n    function loop() {\n        inloop = true;\n        while (got1) {\n            got1 = false;\n            res = iterator.next();\n            if (res.done) sink(2);\n            else sink(1, res.value);\n        }\n        inloop = false;\n    }\n    sink(0, t => {\n        if (t === 1) {\n            got1 = true;\n            if (!inloop && !(res && res.done)) loop();\n        }\n    });\n};\n\nconst interval = (period) => {\n    return (start, sink) => {\n        if (start !== 0) return;\n        let i = 0;\n        const id = setInterval(() => {\n            sink(1, i++);\n        }, period);\n        sink(0, t => {\n            if (t === 2) {\n                clearInterval(id);\n            }\n        });\n    };\n};\n\nconst forEach = (operation) => {\n    return (source) => {\n        let talkback;\n        source(0, (t, d) => {\n            if (t === 0) talkback = d;\n            if (t === 1) operation(d);\n            if (t === 1 || t === 0) talkback(1);\n        });\n    };\n};\n\nconst loop = (operation) => {\n    return (source) => {\n        let talkback;\n        source(0, (t, d) => {\n            if (t === 0) talkback = d;\n            if (t === 1) operation(d);\n        });\n    };\n};\n\nconst delay = (duration) => source => {\n    return (type, sink) => {\n        if (type !== 0) return\n        source(0, (t, d) => {\n            if (t !== 1) return sink(t, d)\n            let id = setTimeout(() => {\n                clearTimeout(id)\n                sink(1, d)\n            }, duration)\n        })\n    }\n};\n\nconst until = (fn) => {\n    return (source) => {\n        return (type, sink) => {\n            if (type !== 0) return;\n            let talkback;\n            source(0, (t, d) => {\n                if (t === 0)\n                {\n                    talkback = d;\n                    sink(0, (t2) => {\n                        if (t2 === 1) talkback(1);\n                    });\n                }\n                if (t === 1) {\n                    if (fn(d)) {\n                        talkback(2);\n                    } else {\n                        sink(1, d);\n                    }\n                }\n            });\n        }\n    }\n}\n\nfunction pipe(...cbs) {\n    let res = cbs[0];\n    for (let i = 1, n = cbs.length; i < n; i++) {\n        let f = cbs[i];\n        res = f(res);\n    }\n    return res;\n};\n\n\/\/pipe(fromIter(range(10, 60)),  until((d) => d > 40), forEach(x => console.log(x)));\n\/\/pipe(interval(500), until((d) => d > 5), forEach(x => console.log(x)));\n\/\/pipe(interval(20), until((d) => d > 500), loop(x => console.log(x)));\n<\/pre>\n\n\n\n
Source: http:\/\/blog.krawaller.se\/posts\/explaining-callbags-via-typescript-definitions\/<\/a><\/p>\n\n\n\n
export type START = 0;\nexport type DATA = 1;\nexport type END = 2;\n\nexport type Callbag = (type: START | DATA | END, payload?: any) => void;\nexport type Factory = (...args: Array<any>) => Callbag;\nexport type Operator = (...args: Array<any>) => (source: Callbag) => Callbag;\n\ntype SourceTalkback = \n  & ((request: DATA) => void)\n  & ((terminate: END) => void);\n\ntype SinkTalkback = \n  & ((start: START, sourceTalkback: SourceTalkback) => void)\n  & ((deliver: DATA, data: any) => void)\n  & ((terminate: END, error?: any) => void);\n\ntype SourceInitiator = (type: START, payload: SinkTalkback) => void;\n\ntype SinkConnector = (source: SourceInitiator) => SourceInitiator | void;\n\ntype SourceFactory = (...args: Array<any>) => SourceInitiator;\n\ntype Operator = (...args: Array<any>) => SinkConnector;\n<\/pre>\n\n\n\n
\"\"<\/figure>\n\n\n\n
\"\"\/<\/figure>\n\n\n\n
Reaping the rewards<\/h3>\n\n\n\n
With these typings down we can use them as a very precise vocabulary for explaining the process of using callbags!<\/p>\n\n\n\n
Here’s the setup and handshake:<\/p>\n\n\n\n
    \n
  1. We call a\u00a0SinkConnector<\/code>\u00a0passing in a\u00a0SourceInitiator<\/code>:\u00a0sinkConnector(sourceInitiator)<\/code><\/li>\n\n\n\n
  2. The\u00a0SourceInitiator<\/code>\u00a0is called with a\u00a0SinkTalkback<\/code>:\u00a0sourceInitiator(0,sinkTalkback)<\/code><\/li>\n\n\n\n
  3. The\u00a0SinkTalkback<\/code>\u00a0is called with a\u00a0SourceTalkback<\/code>:\u00a0sinkTalkback(0,sourceTalkback)<\/code><\/li>\n<\/ol>\n\n\n\n
    And data passing:<\/p>\n\n\n\n
      \n
    1. The\u00a0SourceTalkback<\/code>\u00a0is called with a request for data (if pullable):\u00a0sourceTalkback(1)<\/code><\/li>\n\n\n\n
    2. The\u00a0SinkTalkback<\/code>\u00a0is called with data delivery:\u00a0sinkTalkback(1,data)<\/code><\/li>\n<\/ol>\n\n\n\n
      And termination:<\/p>\n\n\n\n