Event Bus

This library is a simple publish/subscribe bus implementation, which will allow you to send and receive events across your application, organised in different channels with the option to get all the event payloads strictly typed.

Install

yarn add @mcastiello/event-bus

npm install @mcastiello/event-bus

Reference

The event bus is organised in channel. Each channel gets created the first time it is requested, every subsequent get will return the previously initialised instance, unless they get destroyed/closed.

1import { EventBus } from "@mcastiello/event-bus";
2
3const bus = new EventBus();
4
5// Creates and return a channel called "network".
6bus.getChannel("network");
7
8// Close and destroy the previously created channel.
9bus.closeChannel("network");
10
11// Close all the channels created so far.
12bus.clear();

publish/run/subscribe events

You can publish events on a channel. When you do that, all the subscriptions will be executed asynchronously, and they will receive the event payload as a parameter

1import { EventBus } from "@mcastiello/event-bus";
2
3const bus = new EventBus();
4
5// Adds a subscription to the event "update" and it will log the payload when that is published
6bus.getChannel("network").subscribe("update", (data) => console.log(data));
7
8// Publish a message that will be received by the subscriptions.
9bus.getChannel("network").publish("update", "Hello World!");

You can remove your subscription at any time. The subscribe method will return a callback that, once executed, it will remove the event listener.

1const clear = bus.getChannel("network").subscribe("update", (data) => console.log(data));
2
3// Remove the subscription
4clear();

You can also remove all the subscription to a specific event

1bus.getChannel("network").clearSubscriptions("update");
2// Omitting the name of the event will clear all the subscription for every event.

You can also decide to automatically unsubscribe from the event after the subscription is executed once.

1bus.getChannel("network").subscribe("update", (data) => console.log(data), {
2  once: true,
3});

Event cache

Whenever an event is published, its payload is cached, so that when a new subscription is created, it can immediately receive the latest data.

This behaviour can be disabled in a variety of ways. First of all, you can disable it for the entire Event Bus:

1import { EventBus } from "@mcastiello/event-bus";
2
3const bus = new EventBus({
4  cacheEvents: false
5});

Or it can be disabled for a specific channel.

1import { EventBus } from "@mcastiello/event-bus";
2
3const bus = new EventBus();
4
5// Make the channel synchronous.
6const channel = bus.getChannel("network", {
7  cacheEvents: false
8});

it can also be disabled for specific events using the Event configuration (more on that later): This behaviour can be disabled in a variety of ways. First of all, you can disable it for the entire Event Bus:

1import { EventBus } from "@mcastiello/event-bus";
2
3const bus = new EventBus({
4  config: {
5    network: {
6      update: {
7        cache: false
8      }
9    }
10  }
11});
12
13bus.getChannel("network").publish("update", "Hello World!"); // This won't be cached

Run subscriptions synchronously

By default, all subscriptions will be executed asynchronously, but there are different ways to force them to run synchronously.

The following example will cause all subscriptions in all channels to run synchronously:

1import { EventBus } from "@mcastiello/event-bus";
2
3const bus = new EventBus({
4  publishAsynchronously: false
5});

You can also decide to make a specific channel synchronously (even just temporarily):

1import { EventBus } from "@mcastiello/event-bus";
2
3const bus = new EventBus();
4
5// Make the channel synchronous.
6const channel = bus.getChannel("network", {
7  publishAsynchronously: false
8});
9
10// This event will be published synchronously
11channel.publish("update", "Hello World!");

Or you can specify that a specific subscription must always be executed synchronously, even if the event is published asynchronously.

1import { EventBus } from "@mcastiello/event-bus";
2
3const bus = new EventBus();
4
5// Adds a synchronous subscription
6bus.getChannel("network").subscribe("update", (data) => console.log(data), {
7  sync: true,
8});
9
10bus.getChannel("network").publish("update", "Hello World!");

Finally, the event itself can be published synchronously by using the method run. All subscriptions will be executed in sync:

1import { EventBus } from "@mcastiello/event-bus";
2
3const bus = new EventBus();
4
5// Adds an asynchronous subscription
6bus.getChannel("network").subscribe("update", (data) => console.log(data));
7
8// Force all subscriptions to run synchronously
9bus.getChannel("network").run("update", "Hello World!");

once

There is also a special method that returns a promise that will get resolved once the event is published for the first time.

1import { EventBus } from "@mcastiello/event-bus";
2
3const bus = new EventBus();
4
5setTimeout(() => {
6  // Publish the event after 100ms
7  bus.getChannel("network").publish("update", "Hello World!");
8}, 100);
9
10// Await for the event to be dispatched, and returns it to the caller
11const data = await bus.getChannel("network").once("update");
12
13console.log(data);

There is no need to pass an event handler, as the event payload is returned as part of the resolved promise.

request/response events

Using the response method allows you to flip the logic, instead of having a subscription that waits to receive data, you can create a handler that can serve data in response to a request.

It works similarly to a promise; the handler will receive the data from the request, and it can use that to do its calculation and resolve or reject the request.

1import { EventBus } from "@mcastiello/event-bus";
2
3const bus = new EventBus();
4
5// Create a response to an event request. It will convert a string with a numeric list, into a number array.
6const clear = bus.getChannel("network").response("request-numeric-list", (data: string, resolve, reject) => {
7  const response = data.split(",").map((value) => Number(value));
8
9  if (response.includes(NaN)) {
10    reject("Unable to parse numbers");
11  } else {
12    resolve(response);
13  }
14});
15
16const data = await bus.getChannel("network").request("request-numeric-list", "1,2,3");
17
18console.log(data); // [1, 2, 3]

The response returns a clear function that you can use to remove the responder. You can also use:

1bus.getChannel("network").clearResponders("request-numeric-list");
2// Omitting the name of the event will clear all the responses for every event.

intercept events

The intercept method allows you to intercept events or responses in a channel. Each interceptor will receive the event payload, and it has to return the exact same type of payload, but it can alter its content or generate a new one.

1import { EventBus } from "@mcastiello/event-bus";
2
3const bus = new EventBus();
4
5// Adds a subscription to the event "update" and it will log the payload when that is published
6bus.getChannel("network").subscribe("update", (data) => console.log(data)); // false
7
8// Intercept a message with a boolean payload and negate it
9const clear = bus.getChannel("network").intercept("update", (value: boolean) => !value);
10
11// Publish a boolean payload
12bus.getChannel("network").publish("update", true);

The intercept returns a clear function that you can use to remove the interceptor. You can also use:

1bus.getChannel("network").clearInterceptors("update");
2// Omitting the name of the event will clear all the interceptors for every event.

Many different interceptors can be added for the same event, and they will be all executed in sequence, each one updating the data generated from the previous one.

It is although possible to specify the priority of an interceptor in order to control which one is executed first:

1import { EventBus } from "@mcastiello/event-bus";
2
3const bus = new EventBus();
4
5// The higher is the priority, the sooner the interceptor will be executed
6bus.getChannel("network").intercept("update", (value: boolean) => !value);
7// This should normally be executed second, but because of the higher priority (100), it will be executed first
8bus.getChannel("network").intercept("update", (value: boolean) => !value, 100);

Interceptors will also receive a couple of functions to control the execution flow dynamically.

1import { EventBus } from "@mcastiello/event-bus";
2
3const bus = new EventBus();
4
5bus.getChannel("network").intercept("update", (value: boolean, { stopInterceptors, preventPublishing }) => {
6  if (!value) {
7    preventPublishing();
8  }
9});

The 2 functions are:

• stopInterceptors: All the interceptors that should have been executed after the current one will be ignored, and the published event will feature whatever the current interceptor returns.
• preventPublishing: This function has the same effect of the previous one, but it will also prevent the event from being published into the channel. If the intercepted event was the response to a request, the request promise will be resolved with undefined.

Types

The EventBus works well without enforcing any type, but it also allows to add types for all channels, events, and payloads.

You can define what channels are going to be part of the bus, and what events can be dispatched on each channel, and what payload each event is expecting.

Those types will be enforced, and it will also help resolving what type of responses you are going to receive from each request.

Without configuration

The easiest way of adding types on the bus is by declaring a type that extends GenericEventBusDefinition. Each event can define what its payload is going to be, and if the event can respond to a request with another event, or with an error. The following is an example of how to define a simple configuration.

Let's start by defining the name of the channels and of the events.

1enum Channels {
2  Network = "network",
3  Server = "server",
4}
5
6enum NetworkEvents {
7  Update = "update",
8  Message = "message",
9}
10
11enum ServerEvents {
12  Request = "req",
13  Response = "resp",
14  Error = "err",
15}

Then let's put them together to define the bus type:

1type BusDefinition = {
2  [Channels.Network]: {
3    [NetworkEvents.Update]: { payload: boolean };
4    [NetworkEvents.Message]: { payload: string };
5  };
6  [Channels.Server]: {
7    // The Request event will send a string, and it will respond with a Response event, which resolve in an array of numbers. 
8    [ServerEvents.Request]: {
9      payload: string;
10      responseEvent: RequestEvents.Response; // This is optional
11      errorEvent: RequestEvents.Error; // This is optional
12    };
13    [ServerEvents.Response]: { payload: number[] };
14    [ServerEvents.Error]: { payload: string };
15  };
16};

Now that we have our definition, we can add it to the bus using the generic type, to have everything typed.

1import { EventBus } from "@mcastiello/event-bus";
2
3const bus = new EventBus<BusDefinition>();
4
5// If you don't pass a string, TypeScript will complain
6const data = await bus.getChannel(Channels.Server).request(ServerEvents.Request, "1,2,3");
7// The returned value will automatically infer the type `number[]`.

With configuration

The bus will infer the type definition from the config parameter, which can also be used to add some extra functionalities.

You can, in fact, specify if an event can be cached, and if so, you can also specify an initial value that is added to the cache immediately.

If you also specify a responseEvent and/or an errorEvent, whenever a response handler is executed, the returned value or error will not just be returned to the request, they will also be published as separate events.

To create a configuration, you can use the generic type EventBusConfiguration:

1import { EventBus, EventBusConfiguration } from "@mcastiello/event-bus";
2
3const config: EventBusConfiguration<BusDefinition> = {
4  [Channels.Network]: {
5    [NetworkEvents.Update]: { 
6      defaultValue: true,
7    },
8    [NetworkEvents.Message]: {
9      cache: false
10    },
11  },
12  [Channels.Server]: {
13    [ServerEvents.Request]: {
14      responseEvent: ServerEvents.Response,
15      errorEvent: ServerEvents.Error,
16    },
17  },
18}
19
20const bus = new EventBus({ config });

This will use the type definition created earlier, and it will tell the bus that the NetworkEvents.Update will have an initial value of true stored in the cache, the event NetworkEvents.Message won't be cached at all, and the responses to ServerEvents.Request will dispatch their own messages.

1// This will resolve immediately as `true`
2const data = await bus.getChannel(Channels.Network).once(NetworkEvents.Update);
3
4// Whenever a request fails, this subscription will log an error at console.
5bus.getChannel(Channels.Server).subscribe(ServerEvents.Error, (message) => console.error(message));
6// Whenever a request is successful, this subscription will log the response at console.
7bus.getChannel(Channels.Server).subscribe(ServerEvents.Response, (data) => console.debug(data));
8
9try {
10  // This will fail as the payload does not contain a list of numbers.
11  const result = await bus.getChannel(Channels.Server).request(ServerEvents.Request, "Test");
12} catch (error) {
13  // No need for logging here, the subscription will do that
14}

Private channels

Once the bus is typed, you may want to use the same channel types in different streams, for example if you have different modules communicating with a central units, but all sending the same types of messages.

For this particular use case, it is possible to create private channels that rely on an ID shared between different modules. In this way, other part of your application won't be able to listen to events sent in that channel, and their messages won't affect the cache in your private channel.

1import { EventBus } from "@mcastiello/event-bus";
2
3const bus = new EventBus({ config });
4const privateChannel = { channel: Channels.Network, id: "mypricate-id" };
5
6// Adds a subscription to the message event on the private channel
7bus.getChannel(privateChannel).subscribe(NetworkEvents.Message, (data) => console.log(data));
8
9// This will not trigger the subscription.
10bus.getChannel(Channels.Network).publish(NetworkEvents.Message, "Hello Public World!");
11// This will.
12bus.getChannel(privateChannel).publish(NetworkEvents.Message, "Hello Private World!");