{"id":1281,"date":"2018-09-18T22:50:03","date_gmt":"2018-09-18T21:50:03","guid":{"rendered":"https:\/\/solidt.eu\/site\/?p=1281"},"modified":"2019-03-15T11:21:31","modified_gmt":"2019-03-15T10:21:31","slug":"callbag-examples","status":"publish","type":"post","link":"https:\/\/solidt.eu\/site\/callbag-examples\/","title":{"rendered":"Callbag examples"},"content":{"rendered":"\n

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

const fromIter = require('callbag-from-iter');\nconst iterate = require('callbag-iterate');\nconst flatten = require('callbag-flatten');\nconst pipe = require('callbag-pipe');\nconst map = require('callbag-map');\n\nconst source = pipe(\n  fromIter('hi'),\n  map(char => pipe(\n    fromIter([10, 20, 30]),\n    map(num => char + num)\n  )),\n  flatten,\n  iterate(x => console.log(x))\n);\n\n\/\/ h10\n\/\/ h20\n\/\/ h30\n\/\/ i10\n\/\/ i20\n\/\/ i30<\/pre>\n\n\n\n
let source = pipe(\n    fromIter([\".\/BaggingInstructionTemplate.csv\"]), \/\/ start with an array of arguments\n    map((fileName : string) => fromCsvFile(fileName, { delimiter: ',' })), \/\/ map each arguments to a new source\n    flatten, \/\/ flatten the sources, so the results from the file will become the data\n    filter((d: any) => parseInt(d[1]) > 40), \/\/ filter the data on column value\n    take(2), \/\/ take just the first 2 rows that passed the filter\n    map((r: any) => [r[0], r[1]]), \/\/ map the data as we like \n    \/\/forEach((x: any) => console.log(x)) \/\/ we could log here, but the result of a pipe is a source also!\n)\n\nlet src2  = merge(source, source); \/\/ we could merge data from 2 different sources, or just execute the full pipe above twice\n\nforEach(console.log)(src2); \/\/ call forEach with a funcion to execute, that returns a sink, that can be called with a source as argument\n<\/pre>\n\n\n\n
import Promise from 'bluebird';\nvar glob = require('glob');\nconst globAsync : Function = Promise.promisify(glob);\n\npipe(\n    fromPromise(globAsync('*.csv')),\n    map((x: any) => {\n        return fromIter(x);\n    }),\n    flatten,\n    map((x: any) => {\n        console.log(x);\n        return x;\n    }),\n    map((fileName : string) => fromCsvFile(fileName, { delimiter: ',' })), \/\/ map each arguments to a new source\n    flatten, \/\/ flatten the sources, so the results from the file will become the data\n    filter((d: any) => parseInt(d[1]) > 40), \/\/ filter the data on column value\n    take(2), \/\/ take just the first 2 rows that passed the filter\n    map((r: any) => [r[0], r[1]]), \/\/ map the data as we like\n    forEach((x: any) => console.log(x))\n)<\/pre>\n\n\n\n
import { MessageType } from \"..\/enums\";\n\nvar fs = require('fs');\nvar parse = require('csv-parse');\n\ninterface CsvOptions {\n    delimiter: string\n}\n\nexport let fromCsvFile = (fileName: string, options: CsvOptions = { delimiter: ':' }) => (start: number, sink: any) => {\n    if (start !== MessageType.START) return;\n\n    let closed = false;\n    let stream = fs.createReadStream(fileName)\n    stream\n        .pipe(parse(options))\n        .on('data', (row: any) => {\n            if (!closed) {\n                sink(MessageType.DATA, row);\n            }\n        })\n        .on('error', (err: any) => {\n            if (!closed) {\n                closed = true;\n                sink(MessageType.END, err);\n            }\n        })\n        .on('end', function () {\n            closed = true;\n            sink(MessageType.END);\n        });\n    sink(MessageType.START, (t: number) => {\n        if (t === MessageType.END) {\n            closed = true;\n            stream.close();\n        }\n    });\n};\n<\/pre>\n","protected":false},"excerpt":{"rendered":"
Source: https:\/\/github.com\/staltz\/callbag-flatten const fromIter = require(‘callbag-from-iter’); const iterate = require(‘callbag-iterate’); const flatten = require(‘callbag-flatten’); const pipe = require(‘callbag-pipe’); const map = require(‘callbag-map’); const source = pipe( fromIter(‘hi’), map(char => pipe( fromIter([10, 20, 30]), map(num => char + num) )), flatten, iterate(x => console.log(x)) ); \/\/ h10 \/\/ h20 \/\/ h30 \/\/ i10 \/\/ i20 \/\/ i30 […]<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"closed","sticky":false,"template":"","format":"standard","meta":{"inline_featured_image":false,"footnotes":""},"categories":[5],"tags":[],"class_list":["post-1281","post","type-post","status-publish","format-standard","hentry","category-javascript"],"_links":{"self":[{"href":"https:\/\/solidt.eu\/site\/wp-json\/wp\/v2\/posts\/1281","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/solidt.eu\/site\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/solidt.eu\/site\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/solidt.eu\/site\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/solidt.eu\/site\/wp-json\/wp\/v2\/comments?post=1281"}],"version-history":[{"count":3,"href":"https:\/\/solidt.eu\/site\/wp-json\/wp\/v2\/posts\/1281\/revisions"}],"predecessor-version":[{"id":1997,"href":"https:\/\/solidt.eu\/site\/wp-json\/wp\/v2\/posts\/1281\/revisions\/1997"}],"wp:attachment":[{"href":"https:\/\/solidt.eu\/site\/wp-json\/wp\/v2\/media?parent=1281"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/solidt.eu\/site\/wp-json\/wp\/v2\/categories?post=1281"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/solidt.eu\/site\/wp-json\/wp\/v2\/tags?post=1281"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}