1.13.6 • Public • Published


    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


    npm install --save pubchan

    100% Flow Coverage

    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
      .subscribe({ async: true })
      .do(() => console.log("foo 1!"));
    // trigger once synchronously then cancel
      .once(() => console.log("foo 2!"));
      .then(() => console.log("Finished Emitting foo!"));
      // Console Output -->
      foo 2!
      foo 1!
      Finished Emitting foo!

    More Examples

    For more examples you can check out the examples directory

    API Reference

    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.

    Module Exports

    createPubChan (Function) (default)


    Our default export, creates an instance of PubChan.

    import createPubChan from "pubchan";
    const chan = createPubChan();
    Type Signature
    declare function createPubChan(): PubChan;

    PubChan (Class)


    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();

    Subscriber (Class)


    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 Exports

    Registry is an optional sub-module which wraps pubchan and provides a registry for creating and getting channels by a given id.

    PubChanRegistry (Frozen Object) (default)


    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;

    getPubChan (Function)


    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");
    Type Signature
    declare export function getPubChan(id: mixed): PubChan
    declare export function getPubChan(id: ifexists: false): PubChan
    declare export function getPubChan(id: mixed, ifexists: true): void | PubChan

    hasPubChan (Function)

    Check 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")) {
      // ...
    Type Signature
    declare export function hasPubChan(...ids: Array<mixed>): boolean;

    pubChanKeys (Function)


    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()) {
      // ...
    Type Signature
    declare export function pubChanKeys(): Array<mixed>;

    pubChanValues (Function)


    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()) {
      // ...
    Type Signature
    declare export function pubChanValues(): Array<PubChan>;

    pubChanEntries (Function)


    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()) {
      // ...
    Type Signature
    declare export function pubChanEntries(): Array<[mixed, PubChan]>;

    Type Exports

    You can import the various flow-types that pubchan utilizes if needed while annotating your internal functions.

    import type {
    } 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.

    Type Signature
    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;


    npm i pubchan

    DownloadsWeekly Downloads






    Unpacked Size

    63.3 kB

    Total Files


    Last publish


    • avatar