{"_id":"@acransac/streamer","_rev":"1-439783115860a3202344b153e12cfbb0","name":"@acransac/streamer","dist-tags":{"latest":"1.0.0"},"versions":{"1.0.0":{"name":"@acransac/streamer","version":"1.0.0","description":"A model to process events with Node.js","main":"index.js","scripts":{"test":"node test_streamer.js"},"author":{"name":"Adrien Cransac"},"license":"MIT","repository":{"type":"git","url":"git+https://github.com/acransac/streamer.git"},"bugs":{"url":"https://github.com/acransac/streamer/issues"},"homepage":"https://github.com/acransac/streamer","dependencies":{"@acransac/tester":"^1.0.0"},"gitHead":"6c05b447f8409bf245cca37137286074537eb7eb","_id":"@acransac/streamer@1.0.0","_nodeVersion":"12.12.0","_npmVersion":"6.11.3","dist":{"integrity":"sha512-C1K/W23GwNGE6mjwYK9jOdUatZ3Y6Pvdz5g9L6Z+kWRGV7CYF1KA4Ubhk63m8Yks1ljIYntPLmyg/zoYc38RrA==","shasum":"7386bdd3e4606bec9c9e5fb5dcaab3e37dc5b558","tarball":"https://registry.npmjs.org/@acransac/streamer/-/streamer-1.0.0.tgz","fileCount":7,"unpackedSize":21592,"npm-signature":"-----BEGIN PGP SIGNATURE-----\r\nVersion: OpenPGP.js v3.0.4\r\nComment: https://openpgpjs.org\r\n\r\nwsFcBAEBCAAQBQJfJrvHCRA9TVsSAnZWagAAE6QP/3pWMriXyNBPMo7dBnC4\n7hvcZR92WAfJ65it33LgL/6K9DsWFs6yYVhzm+qA7BoBJ72CXaMX/NqQezRQ\n4AmHVQWh1pzgd1w3EwSAKEF9+Zl49FxqmkGTwciMfN6mH9wtttfZrjfiIe+V\nh6TftZBTH2rBmR/MNwg0ynpjQ3WziA4Rpf/Ba73Be4qqXCtucqkOVydu0Zgi\nOfhlzU2ptmuAPJar7+urOyYSDUtAj3+no1TRgcMuHfMd74K4VF4wwfiXjXYR\nERCq//G9IQWi3eZuqUVS9teWA2IGwivV8+L60gd9hNjJjCA0cn84xSmijz6b\nIXG1Iufq/STsEc/WOq0CxFBJ6oaTQFYfWxvq9f9UwP7b32UVy5YDPdQbPwIX\nFRaz6oKP7xAGq4XjkxdFtFN7fxGv+36jzPXbYVsMD7M0pJaSxmHWT1fXnHse\nCYyWCAE7jyMW0S4Au/bTIE15IX2guQ9wwDBmrPgHnwiDI9+m4f3FmMun3Uip\nHMOT6MveC5m/LkS66w+5kD+HfYAXEA/gsAT51N8tRSIIvo+hZhlnmmgSiGWr\nBdxyxDC+1yIp8ytoUw4tz7kgHEY5ZLdGKAUBCJE622Eyd69lUsfnbiDOp6x6\nJ2X04C2xygwKejYfT/h0diU6di+ypnn3C1Y2sbuIF0/egSa5sZvE9DCa768H\nrJdO\r\n=ll4M\r\n-----END PGP SIGNATURE-----\r\n","signatures":[{"keyid":"SHA256:jl3bwswu80PjjokCgh0o2w5c2U4LhQAE57gj9cz1kzA","sig":"MEUCIQDb7UeLNAS39oSSvC3KgXBW0bhStcjHya3yKb/AD/vz9gIgYJ5DVM735q7eH3n/BTlJyL+DSRojztDQ0egJAs/euXE="}]},"maintainers":[{"name":"acransac","email":"adrien.cransac@gmail.com"}],"_npmUser":{"name":"acransac","email":"adrien.cransac@gmail.com"},"directories":{},"_npmOperationalInternal":{"host":"s3://npm-registry-packages","tmp":"tmp/streamer_1.0.0_1596373958720_0.5165605950073959"},"_hasShrinkwrap":false}},"time":{"created":"2020-08-02T13:12:38.667Z","1.0.0":"2020-08-02T13:12:38.845Z","modified":"2022-04-04T11:16:08.313Z"},"maintainers":[{"name":"acransac","email":"adrien.cransac@gmail.com"}],"description":"A model to process events with Node.js","homepage":"https://github.com/acransac/streamer","repository":{"type":"git","url":"git+https://github.com/acransac/streamer.git"},"author":{"name":"Adrien Cransac"},"bugs":{"url":"https://github.com/acransac/streamer/issues"},"license":"MIT","readme":"# Introduction\n**streamer** provides an easy-to-reason-about model to process a stream of events with Node.js. There is a _source_ of events to which is attached a composition of _processes_ which see all events emitted by the source. These processes receive and output the _stream_. The latter is defined recursively as the pair of a current event with a later stream.\n\nAs a result, processes can be defined by recurring on the sequence of events, retrieving the available event from the stream and awaiting the ones coming afterwards.\n\nTo make composition easier, each process can record a variation of itself to execute on the next event carried by the stream. Also, one process can transform the value defining the event and return it to the following steps.\n\n# How To Use Streamer\n**streamer** is a small helper library. Add it to a project with:\n\n```shell\n    $ npm install @acransac/streamer\n```\n\nand import the needed functionalities:\n\n```javascript\n    const { commit, continuation, floatOn, forget, later, makeEmitter, mergeEvents, now, Source, StreamerTest, value } = require('@acransac/streamer');\n```\n\n## Make A Source\nA `Source` is built up with `Source.from` chained with `Source.withDownstream`:\n* `Source.from:: (EventEmitter, String) -> Source`\n  | Parameter            | Type         | Description             |\n  |----------------------|--------------|-------------------------|\n  | eventEmitter         | EventEmitter | A Node.js event emitter |\n  | emissionCallbackName | String       | The name of the callback of the event to listen to, as used in the statement `eventEmitter.on('someEvent', emissionCallbackName)` |\n\n* `Source.withDownstream:: Process -> Source`\n  | Parameter  | Type    | Description                                                      |\n  |------------|---------|------------------------------------------------------------------|\n  | downstream | Process | The composition of processes to execute when an event is emitted |\n\n  where `Process:: async Stream -> Stream`\n\nExample:\n\n```javascript\n    const EventEmitter = require('events');\n    const { Source } = require('@acransac/streamer');\n\n    class Emitter extends EventEmitter {\n      constructor() {\n        super();\n\n        this.onevent = () => {};\n\n        this.on('event', event => this.onevent(event));\n      }\n    };\n\n    const emitter = new Emitter();\n\n    Source.from(emitter, \"onevent\").withDownstream(async (stream) => {\n      console.log(\"event emitted and processed\");\n\n      return stream;\n    });\n\n    emitter.emit('event');\n```\n\n```shell\n    $ node example.js\n    event emitted and processed\n```\n\n**streamer** also provides the wrapper `mergeEvents` that can merge several event emitters into one. These emitters have to be constructed with `makeEmitter`:\n* `mergeEvents:: [Emitter] -> EventEmitter`\n  | Parameter | Type      | Description                             |\n  |-----------|-----------|-----------------------------------------|\n  | emitters  | [Emitter] | An array of event emitters to listen to |\n\n  The returned event emitter exposes an emission callback named `\"onevent\"` which is used as the second parameter to `Source.from`.\n\n* `makeEmitter:: (EventEmitter, String) -> Emitter`\n  | Parameter    | Type         | Description             |\n  |--------------|--------------|-------------------------|\n  | eventEmitter | EventEmitter | A Node.js event emitter |\n  | eventName    | String       | The name of the event listened to, as used in the statement `eventEmitter.on('eventName', someCallback)` |\n\nNote: it is then possible to wrap an emitter that does not expose a callback into one that does with the combination of `mergeEvents` and `makeEmitter`.\n\nExample:\n\n```javascript\n    const EventEmitter = require('events');\n    const { makeEmitter, mergeEvents, Source } = require('@acransac/streamer');\n\n    const emitter1 = new EventEmitter();\n\n    const emitter2 = new EventEmitter();\n\n    Source.from(mergeEvents([makeEmitter(emitter1, \"someEvent\"), makeEmitter(emitter2, \"anotherEvent\")]), \"onevent\")\n          .withDownstream(async (stream) => {\n      console.log(\"event emitted and processed\")\n\n      return stream;\n    });\n\n    emitter1.emit('someEvent'); // or emitter2.emit('anotherEvent');\n```\n\n```shell\n    $ node example.js\n    event emitted and processed`\n```\n\n## Make A Process\nA process is an asynchronous function that receives and outputs a stream. It can be a composition of smaller such functions. From within a process, the value attached to the available event is retrieved with `value(now(stream))`. Events that are not yet produced can be awaited with `await later(stream)`. Because the stream is defined in terms of itself, the processes lend themselves to a recursive style:\n\n* `now:: Stream -> AvailableStream`\n  | Parameter | Type   | Description |\n  |-----------|--------|-------------|\n  | stream    | Stream | The stream  |\n\n* `later:: Stream -> Promise<Stream>`\n  | Parameter | Type   | Description |\n  |-----------|--------|-------------|\n  | stream    | Stream | The stream  |\n\n* `value:: AvailableStream -> Any`\n  | Parameter | Type            | Description                                              |\n  |-----------|-----------------|----------------------------------------------------------|\n  | now       | AvailableStream | The current stream from which the event can be retrieved |\n\nExample:\n\n```javascript\n   const { later, now, Source, StreamerTest, value } = require('@acransac/streamer');\n\n   const processA = async (stream) => {\n     if (value(now(stream)) > 3) {\n       return stream;\n     }\n     else {\n       console.log(value(now(stream)));\n\n       return processA(await later(stream));\n     }\n   };\n\n   const processB = async (stream) => {\n     console.log(\"stream processed\");\n\n     return stream;\n   };\n\n   Source.from(StreamerTest.emitSequence([1, 2, 3, 4]), \"onevent\")\n         .withDownstream(async (stream) => processB(await processA(stream)));\n```\n\n```shell\n    $ node example.js\n    1\n    2\n    3\n    stream processed\n```\n\n## Make A Composition Of Processes\nComplex processes are more easily defined by chaining smaller functions implementing a specific task each. One event has to pass through every step so it is not possible to await the later stream in each of these. Instead, a function records to the stream what should be executed on the next event. The chain of future processes constitutes the _continuation_.\n\n`commit` is used to record the next iteration of a process and is called in the return statement. `continuation` returns the future processing sequence from the available stream (`continuation(now(stream))`). `forget` clears out the continuation:\n\n* `commit:: (Stream, Process) -> Stream`\n  | Parameter | Type    | Description                              |\n  |-----------|---------|------------------------------------------|\n  | stream    | Stream  | The stream                               |\n  | process   | Process | The process to execute on the next event |\n\n* `continuation:: AvailableStream -> Process`\n  | Parameter | Type            | Description          |\n  |-----------|-----------------|----------------------|\n  | now       | AvailableStream | The available stream |\n\n* `forget:: Stream -> Stream`\n  | Parameter | Type   | Description |\n  |-----------|--------|-------------|\n  | stream    | Stream | The stream  |\n\nNotes:\n\n* Using `continuation` and `forget` together in the last step of a composed process allows to define loops (see example).\n\n* A conditional loop structure in the middle of the chain of processes effectively filters out choosen events for the subsequent steps.\n\nExample:\n\n```javascript\n    const { commit, continuation, forget, later, now, Source, StreamerTest, value } = require('@acransac/streamer');\n\n    const parseLetters = parsed => async (stream) => {\n      if (typeof value(now(stream)) === \"string\" && value(now(stream)) !== \"end\") {\n        console.log(parsed + value(now(stream)));\n\n        return commit(stream, parseLetters(parsed + value(now(stream))));\n      }\n      else {\n        return commit(stream, parseLetters(parsed));\n      }\n    };\n\n    const sumNumbers = sum => async (stream) => {\n      if (typeof value(now(stream)) === \"number\") {\n        console.log(sum + value(now(stream)));\n\n        return commit(stream, sumNumbers(sum + value(now(stream))));\n      }\n      else {\n        return commit(stream, sumNumbers(sum));\n      }\n    };\n\n    const loop = async (stream) => {\n      if (value(now(stream)) === \"end\") {\n        console.log(\"stream processed\");\n\n        return stream;\n      }\n      else {\n        return loop(await continuation(now(stream))(forget(await later(stream))));\n      }\n    };\n\n    Source.from(StreamerTest.emitSequence([\"a\", 1, \"b\", 2, \"end\"]), \"onevent\")\n          .withDownstream(async (stream) => loop(await sumNumbers(0)(await parseLetters(\"\")(stream))));\n```\n\n```shell\n    $ node example.js\n    a\n    1\n    ab\n    3\n    stream processed\n```\n\n## Transform Events\nOne process can float a value downstream with `floatOn`. It is used in the return statement, possibly chained with `commit`:\n\n* `floatOn:: (Stream, Any) -> Stream`\n  | Parameter | Type   | Description                                           |\n  |-----------|--------|-------------------------------------------------------|\n  | stream    | Stream | The stream                                            |\n  | jsValue   | Any    | The value to pass on to the next steps of the process |\n\nExample:\n\n```javascript\n    const { commit, continuation, floatOn, forget, later, now, Source, StreamerTest, value } = require('@acransac/streamer');\n\n    const upperCase = async (stream) => {\n      if (value(now(stream)) !== \"end\") {\n        return commit(floatOn(stream, value(now(stream)).toUpperCase()), upperCase);\n      }\n      else {\n        return stream;\n      }\n    };\n\n    const parse = parsed => async (stream) => {\n      if (value(now(stream)) !== \"end\") {\n        console.log(parsed + value(now(stream)));\n\n        return commit(stream, parse(parsed + value(now(stream))));\n      }\n      else {\n        return stream;\n      }\n    };\n\n    const loop = async (stream) => {\n      if (value(now(stream)) === \"end\") {\n        console.log(\"stream processed\");\n\n        return stream;\n      }\n      else {\n        return loop(await continuation(now(stream))(forget(await later(stream))));\n      }\n    };\n\n    Source.from(StreamerTest.emitSequence([\"a\", \"b\", \"c\", \"end\"]), \"onevent\")\n          .withDownstream(async (stream) => loop(await parse(\"\")(await upperCase(stream))));\n```\n\n```shell\n    $ node example.js\n    A\n    AB\n    ABC\n    stream processed\n```\n\n## Test The Process\nAs observed in the examples, **streamer** provides a test event emitter `StreamerTest.emitSequence` (whose emission callback name is `\"onevent\"`):\n* `StreamerTest.emitSequence:: ([Any], Maybe<Number>) -> EventEmitter`\n  | Parameter | Type           | Description                                                 |\n  |-----------|----------------|-------------------------------------------------------------|\n  | sequence  | [Any]          | An array of values to emit in sequence                      |\n  | delay     | Maybe\\<Number> | The time interval in ms between two events. Default: 200 ms |\n","readmeFilename":"README.md"}