Branch: master
Find file History
Fetching latest commit…
Cannot retrieve the latest commit at this time.
Permalink
Type Name Latest commit message Commit time
..
Failed to load latest commit information.
src
test
.npmignore
CHANGELOG.md
LICENSE
README.md
package.json
tsconfig.json

README.md

@thi.ng/csp

npm version npm downloads Twitter Follow

This project is part of the @thi.ng/umbrella monorepo.

About

Lightweight ES6-promise based CSP implementation written in TypeScript (18KB minified, full lib).

  • Channel with/without buffering and/or transducers
    • optional channel IDs
    • choice of buffer behaviors (fixed, sliding, dropping)
    • channel selection
    • channel merging (many-to-one, serial or parallel)
    • channel piping (w/ transducers)
    • timeouts / sleeping / throttling / delaying
    • prepopulated channel ctors (iterators, ranges, promise, constants etc.)
  • Mult for channel multiplexing (one-to-many splitting)
    • individual transducers per tap
    • dynamic add/removal of taps
  • PubSub for topic subscriptions
    • each topic implemented as Mult
    • wildcard topic for processing fallthrough messages

Installation

yarn add @thi.ng/csp

Dependencies

Usage examples

File loading & word frequency analysis

import { Channel } from "@thi.ng/csp";
import * as tx from "@thi.ng/transducers";

// compose transducer to split source file into words
// and filter out short strings
const proc: tx.Transducer<string, string> = tx.comp(
    tx.mapcat((src: string) => src.toLowerCase().split(/[^\w]+/g)),
    tx.filter((w: string) => w.length > 1)
);

// define a channel which receives file paths
// and resolves them with their contents
const paths = new Channel<any>(
    tx.map((path: string) =>
        new Promise<string>(
            resolve => fs.readFile(path, (_, data) => resolve(data.toString()))
        )
    )
);

// define multiplexed output channel
// items in this channel will have this form: `[word, count]`
const results = new Mult("results");

// tap result channel and sum word counts
const counter = results
    .tap(tx.map(x => x[1]))
    .reduce(tx.add());

// 2nd output channel with streaming sort transducer
// (using a sliding window size of 500 items) and dropping
// words with < 20 occurrences
const sorted = results.tap(
    tx.comp(
        tx.streamSort(500, x => x[1]),
        tx.dropWhile(x => x[1] < 20)
    )
);

// define workflow:
// pipe source files into a new channel and
// reduce this channel using `frequencies` reducer
// finally stream the result map (word frequencies)
// into the `sorted` channel
// (`freqs` is a JS Map and is iterable)
paths.pipe(proc)
    .reduce(tx.frequencies())
    .then(freqs => results.channel().into(freqs));

// start tracing sorted outputs and
// wait for all to finish
Promise
    .all([sorted.consume(), counter])
    .then(([_, num]) => console.log("total words:", num));

// no real work has been executed thus far (only scheduled via promises)
// now kick off entire process by writing file paths into the 1st channel
paths.into(["src/channel.ts", "src/mult.ts", "src/pubsub.ts"]);

// results-tap1 : [ 'let', 20 ]
// results-tap1 : [ 'topic', 20 ]
// results-tap1 : [ 'chan', 20 ]
// results-tap1 : [ 'number', 22 ]
// results-tap1 : [ 'buf', 23 ]
// results-tap1 : [ 'length', 23 ]
// results-tap1 : [ 'tx', 25 ]
// results-tap1 : [ 'state', 25 ]
// results-tap1 : [ 'from', 27 ]
// results-tap1 : [ 'close', 28 ]
// results-tap1 : [ 'new', 33 ]
// results-tap1 : [ 'args', 34 ]
// results-tap1 : [ 'id', 36 ]
// results-tap1 : [ 'any', 36 ]
// results-tap1 : [ 'src', 38 ]
// results-tap1 : [ 'if', 40 ]
// results-tap1 : [ 'return', 47 ]
// results-tap1 : [ 'channel', 57 ]
// results-tap1 : [ 'this', 120 ]
// results-tap1 done
// total words: 1607

Channel merging

Channel.merge([
    Channel.range(0, 3),
    Channel.range(10, 15),
    Channel.range(100, 110)
]).reduce(tx.push()).then(console.log);

// [ 0, 100, 101, 102, 103, 1, 2, 104, 105, 10, 11, 12, 13, 106, 14, 107, 108, 109 ]

// emit tuples of values read from all inputs
// preserves ordering of all inputs, but
// throughput controlled by slowest input
// by default stops & closes when any of the inputs closes
Channel.mergeTuples([
    Channel.from([1, 2, 3]),
    Channel.from([10, 20, 30, 40]),
    Channel.from([100, 200, 300, 400, 500])
], null, false).consume();

// chan-3 : [ 1, 10, 100 ]
// chan-3 : [ 2, 20, 200 ]
// chan-3 : [ 3, 30, 300 ]
// chan-3 done

// same as above, however continues until all inputs are closed
Channel.mergeTuples([
    Channel.from([1, 2, 3]),
    Channel.from([10, 20, 30, 40]),
    Channel.from([100, 200, 300, 400, 500])
], null, false).consume();

// chan-3 : [ 1, 10, 100 ]
// chan-3 : [ 2, 20, 200 ]
// chan-3 : [ 3, 30, 300 ]
// chan-3 : [ undefined, 40, 400 ]
// chan-3 : [ undefined, undefined, 500 ]
// chan-3 done

PubSub

// define a channel publisher with transducer and topic function applied to each item
// the input channel receives names and transforms them into indexable objects
const pub = new PubSub(
    new Channel<any>("users", tx.map((x: string) => ({ type: x.charAt(0), val: x }))),
    (x) => x.type
);

// create subscriptions (channel + debug consumer)
// under the hood each topic is a Mult (multiplexed channel)
// sub channels are automatically named:
// `<src-id>-<topic>-tap<tapid>` (see below)
for (let i of "abc") {
    pub.sub(i).consume();
}

// start processing, then close everything down
// (pubsubs & mults are closed recursively once the input channel is closed)
pub.channel().into(["alice", "bert", "bella", "charlie", "arthur"]);

// users-a-tap0 : { type: 'a', val: 'alice' }
// users-b-tap0 : { type: 'b', val: 'bert' }
// users-b-tap0 : { type: 'b', val: 'bella' }
// users-c-tap0 : { type: 'c', val: 'charlie' }
// users-a-tap0 : { type: 'a', val: 'arthur' }
// users-b-tap0 done
// users-c-tap0 done
// users-a-tap0 done

Authors

  • Karsten Schmidt

License

© 2016-2018 Karsten Schmidt // Apache Software License 2.0