Simple yet powerful async pub/sub channels for Javascript and Node.js with zero dependencies.
Tiny, fast, and reliable pubsub event emitter with promises, optional result aggregation (state), and async/sync controls.
Used in production to coordinate & route millions of events within our proprietary Datastream Realtime Event Protocol.
yarn add pubchan
or
npm install --save pubchan
Proudly built with 100% Flow Coverage and exported .flow.js files so your flow projects will benefit!
We strongly recommend you look over the types folder. This will give you an idea of how the various pieces of the package work.
This library was taken out of our broad library used for our internal DRE Protocol. While it has 100% Test Coverage there, we have not yet built standalone tests for this library.
We plan to add coverage for it when time permits. In the meantime, it does maintain it's 100% type coverage.
Pull requests are more than welcome if you happen to have the time to contribute!
/* @flow */
import createPubChan from "pubchan";
const chan = createPubChan();
// trigger asynchronously whenever foo is received
chan
.subscribe({ async: true })
.to("foo")
.do(() => console.log("foo 1!"));
// trigger once synchronously then cancel
chan
.subscribe()
.to("foo")
.once(() => console.log("foo 2!"));
chan
.emit("foo")
.send()
.then(() => console.log("Finished Emitting foo!"));
/*
// Console Output -->
foo 2!
foo 1!
Finished Emitting foo!
*/For more examples you can check out the examples directory
Note: This is currently a work in progress. The API is quite simple and can be understood by looking at our types files as this packages has 100% flowtype coverage.
Our default export, creates an instance of PubChan.
import createPubChan from "pubchan";
const chan = createPubChan();declare function createPubChan(): PubChan;Generally using the createPubChan function is the recommended method of
creating a new channel. However, you can also import the class directly if
needed (can be useful for adding as flow type).
import { PubChan } from "pubchan";
const chan = new PubChan();This should have no use other than possibly to use for flow-types. A subscriber
is useless unless created by an interface which matches the PubChan.
/* @flow */
import type { Subscriber } from "pubchan";Registry is an optional sub-module which wraps pubchan and provides a registry
for creating and getting channels by a given id.
An object which encapsulates the standard exports listed below.
const PubChanRegistry = Object.freeze({
has: hasPubChan,
get: getPubChan,
keys: pubChanKeys,
create: getPubChan,
values: pubChanValues,
entries: pubChanEntries
});
export default PubChanRegistry;Gets a PubChan with a given id which can be any type that can be a key on a
Map. If the id already exists then it returns that PubChan instead of
creating a new one.
PubChan's that are created from the registry automatically subscribe to
CLOSED events to clean themselves up when you close the channel from anywhere
in the app.
It's second argument optionally accepts a boolean which indicates if a PubChan
should only be returned if it already exists. If set to true then the function
will return undefined if the pubchan has not yet been created.
import getPubChan from "pubchan/registry";
const chan = getPubChan("mychan");declare export function getPubChan(id: mixed): PubChan
declare export function getPubChan(id: ifexists: false): PubChan
declare export function getPubChan(id: mixed, ifexists: true): void | PubChanCheck if the given PubChan(s) exists within the registry. Returns true if
every id given is found.
import { hasPubChan } from "pubchan/registry";
if (hasPubChan("mychan", "anotherchan")) {
// ...
}declare export function hasPubChan(...ids: Array<mixed>): boolean;Returns an array with all entries within the registry. Takes the form of [id, id, ...] inline with a call to a Map's .keys() call cast to an Array.
import { pubChanKeys } from "pubchan/registry";
for (const id of pubChanKeys()) {
// ...
}declare export function pubChanKeys(): Array<mixed>;Returns an array with all entries within the registry. Takes the form of [chan, chan, ...] inline with a call to a Map's .values() call cast to an Array.
import { pubChanValues } from "pubchan/registry";
for (const chan of pubChanValues()) {
// ...
}declare export function pubChanValues(): Array<PubChan>;Returns an array with all entries within the registry. Takes the form of [[key, value], ...] inline with a call to a Map's .entries() call cast to an
Array.
import { pubChanEntries } from "pubchan/registry";
for (const [id, chan] of pubChanEntries()) {
// ...
}declare export function pubChanEntries(): Array<[mixed, PubChan]>;You can import the various
flow-types
that pubchan utilizes if needed while annotating your internal functions.
import type {
PubChan$EmitIDs,
PubChan$Options,
PubChan$Pipeline,
PubChan$Listeners,
PubChan$SubscriberSet,
PubChan$EmitResponseRef,
PubChan$State,
PubChan$ResolvedPipeline,
PubChan$Ref
} from "pubchan/lib/types";There are some useful utility types
also available at
pubchan/lib/types/utils
Below is a normalized version of the PubChan class's public interface which
should provide the structure of the instance returned by our createPubChan
factory.
declare class PubChan {
// how many total active subscribers do we have?
get length(): number,
get size(): number,
// emit the given ids and execute any matching subscribers
emit: (...args: Array<Array<mixed> | mixed>) => PubChan,
// include args with the event emission
with: (...args: Array<any>) => PubChan,
// include state with the event for chained emissions
state = (...args: Array<?{ [key: string]: any }>) => PubChan,
// send the event and optionally include args to the handlers.
send: (...args: Array<any>) => Promise<null> | Promise<Array<any>>,
close: (...args: Array<any>) => Promise<null> | Promise<Array<any>>,
subscribe: (
options?: $Shape<{|
async?: boolean,
context?: Object
|}>,
) => Subscriber,
}Subscriber instances are returned by a call to .subscribe() on our PubChan
instance.
declare type PubChan$EmitID = mixed;
type Callback = (
ref: PubChan$Ref,
ids: Set<PubChan$EmitID>,
...args: Array<mixed>
) => Array<mixed> | mixed;
declare type PubChan$Callback = Array<Callback> | Callback;
// called immediately after creating the subscription (not when the
// subscription is cancelled)
declare type PubChan$CompleteCallback = (ref: PubChan$Ref) => mixed;
declare interface PubChan$Ref {
+once?: void | boolean;
+id?: PubChan$EmitID;
+state: { [key: string]: * };
+subscription: Subscriber;
+chan: PubChan;
+callback: PubChan$Callback;
+cancel: () => void;
}
declare class Subscriber {
// total number of active callbacks on the subscriber
get length(): number;
get size(): number;
// all the ids we are subscribed to
get keys(): Array<PubChan$EmitID>;
// optionally attach a custom context to all "do" and "once" callbacks
context: (context: Object) => Subscriber;
// subscribe to EmitIDS
to: (...args: Array<PubChan$EmitIDs>) => Subscriber;
// add a callback that will happen once then cancel itself
once: (
callback: PubChan$Callback,
onComplete?: PubChan$CompleteCallback
) => Subscriber;
// add a callback when this event occurs
do: (
callback: PubChan$Callback,
onComplete?: PubChan$CompleteCallback
) => Subscriber;
// cancel the entire subscriber
cancel: () => void;
}