The Publisher Subscriber pattern
Tags: javascript, pubsub, patterns, architecture
Published: 2020-04-12 21:02:30

The Publisher Subscriber pattern, also known as PubSub, is an architectural pattern for relaying messages to interested parties through a publisher. The publisher is not generally aware of the subscribers per say but in our implementation it will so that we can ease into the topic.

The PubSub pattern gives us a scalable way to relay messages around our applications but is inflexible in one area and that is the data structure that is sent to each subscriber when a new message is published. Generally this is a good thing though in my opinion since it allows a nice normalised way of transacting data through our applications.

## Tests

For the tests I will be using JavaScript and [the Jest test runner](https://jestjs.io/).

```javascript
const Publisher = require('./publisher');

let publisher;
beforeEach(() => publisher = new Publisher);

describe("Publisher", () => {
  it("Should construct with default values", () => {
    expect(publisher.topic).toEqual("unknown");
    expect(publisher.subscribers).toEqual([]);
  });

  it("Should add subscribers properly", () => {
    const subscriber = jest.fn();
    expect(publisher.subscribers.length).toEqual(0);
    publisher.subscribe(subscriber);
    expect(publisher.subscribers.length).toEqual(1);
  });

  it("Should publish updates to subscribers", () => {
    const subscriber = jest.fn();
    publisher.subscribe(subscriber);
    publisher.publish("test");
    expect(subscriber).toHaveBeenCalledWith({
      topic: "unknown",
      data: "test"
    });
  });

  it("Should unsubscribe from updates as required", () => {
    const subscriber = jest.fn();
    const subscription = publisher.subscribe(subscriber);
    publisher.publish("test");
    expect(subscriber).toHaveBeenCalledTimes(1);
    publisher.unsubscribe(subscription);
    publisher.publish("test");
    expect(subscriber).toHaveBeenCalledTimes(1);
  });

  it("Should not unsubscribe a subscriber from updates unless it exists", () => {
    const subscriber = jest.fn();
    publisher.subscribe(subscriber);
    expect(publisher.subscribers.length).toEqual(1);
    publisher.unsubscribe(() => 24);
    expect(publisher.subscribers.length).toEqual(1);
  });

  it("Generates a consistent subscription id for each subscriber", () => {
    const subscriber = jest.fn();
    const subscription = publisher.subscribe(subscriber);
    const proof = publisher.createSubscriptionId(subscriber);
    expect(subscription).toEqual(proof);
  });
});
```

Here we test that:

1. We begin with sane defaults
2. We can add subscribers
3. We can notify subscribers
4. We can remove subscribers
5. We only remove subscribers when they exist
6. We generate consistent ids for each subscriber that is provided

You can run the tests here:

{% replit @jamesrweb/PubSubExampleTests %}

This covers the bases required of a publisher and subscriber and gives us control over who does and does not get notifications when new content is published. Pretty simple so far, right?

## Implementation

For our implementation I will be using [TypeScript, a typed superset of JavaScript](https://www.typescriptlang.org/). If you are more comfortable with JavaScript you can [compile TypeScript code to JavaScript](https://www.typescriptlang.org/play) in the TypeScript playground.

```typescript
export interface ISubscriberOutput { 
  topic: string; 
  data: any; 
};

export class Publisher {
  public topic: string = "unknown";
  private subscribers: Function[] = [];

  public subscribe(subscriberFn: Function): number {
    this.subscribers = [...this.subscribers, subscriberFn];
    const subscriptionId = this.createSubscriptionId(subscriberFn);
    return subscriptionId;
  }

  public publish(data: any): void {
    this.subscribers.forEach((subscriberFn: Function) => {
      const output: ISubscriberOutput = { topic: this.topic, data };
      subscriberFn(output);
    });
  }

  public unsubscribe(subscriptionId: number): void {
    const subscriberFns = [...this.subscribers];
    subscriberFns.forEach((subscriberFn: Function, index: number) => {
      if(this.createSubscriptionId(subscriberFn) === subscriptionId) {
        subscriberFns.splice(index, 1);
        this.subscribers = [...subscriberFns];
      }
    });
  }

  private createSubscriptionId(subscriberFn: Function): number {
    const encodeString = this.topic + subscriberFn.toString();
    return [...encodeString].reduce((accumulator, char) => {
      return char.charCodeAt(0) + ((accumulator << 5) - accumulator);
    }, 0);
  }
}
```

This class generates a Publisher with a set of methods for us to use for publishing updates, subscribing to those updates and also unsubscribing when the need arises. Let's break things down from top to bottom.

```typescript
export interface ISubscriberOutput { 
  topic: string; 
  data: any; 
};
```

This interface is able to be used by subscribers that will take in messages when the `publish` method is called on the `Publisher` and gives us the structured message output we discussed in the introduction of this article.

```typescript
  public topic: string = "unknown";
  private subscribers: Function[] = [];
```

As we begin to define the `Publisher` class, we first initialise the class with a topic of "unknown" since the topic hasn't been provided or overridden. We also have an array of `subscribers` initialised, each of which should be a `Function`.

Next we create the `subscribe` method. This will add the provided `subscriberFn` function to the `subscribers` array and then return a `subscriptionId` for us to use later should we choose to unsubscribe down the road.

```typescript
  public subscribe(subscriberFn: Function): number {
    this.subscribers = [...this.subscribers, subscriberFn];
    const subscriptionId = this.createSubscriptionId(subscriberFn);
    return subscriptionId;
  }
```

The `createSubscriptionId` generates a unique ID for each subscriber and utilises the same algorithm as the [the Java String hashCode() Method](https://www.tutorialspoint.com/java/java_string_hashcode.htm).

```typescript
  private createSubscriptionId(subscriberFn: Function): number {
    const encodeString = this.topic + subscriberFn.toString();
    return [...encodeString].reduce((accumulator, char) => {
      return char.charCodeAt(0) + ((accumulator << 5) - accumulator);
    }, 0);
  }
```

In short we take the current `topic` and add to that the string representation of the `subscriberFn`. This gives us a somewhat unique string but is not bulletproof by any means. From here we take each character in the `encodeString` and reduce it to a number representation unique to that string.

> Sidenote: We use the `<<` operator which seems to confuse a lot of developers but in this case we are just doing this in essence: `accumulator * (2 ** 5)`.
>
> So let's say it is the second iteration of the `reduce` loop and accumulator is at an arbitrary value of 65 which is the character code for lowercase "a".
> 
> This being the case the `(accumulator << 5) - accumulator` line is the same as writing `(65 * (2 ** 5)) - 65` during that iteration.

> Either way you write it, the resulting number will be 2015 for that example.
>
> Hopefully that makes a bit more sense for you who may not have experienced using bitwise operators before.
>
> I highly recommend [reading the MDN bitwise reference](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Operators/Bitwise_Operators) for more information.

If we want to unsubscribe from a `Publisher` at any time, you can simply call the `unsubscribe` method passing in the return value of the original `subscribe` call.

```typescript
  public unsubscribe(subscriptionId: number): void {
    const subscriberFns = [...this.subscribers];
    subscriberFns.forEach((subscriberFn: Function, index: number) => {
      if(this.createSubscriptionId(subscriberFn) === subscriptionId) {
        subscriberFns.splice(index, 1);
        this.subscribers = [...subscriberFns];
      }
    });
  }
```

Here we clone the current subscribers and loop over the clone until we find one that when it is hashed in the `createSubscriptionId` function, matches the provided `subscriptionId` value.

If we find a match then we remove that function from the `subscriberFns` array and set the `subscribers` to contain only the remaining `subscriberFns`.

Lastly we will look at the `publish` function which takes in some `data` which can be anything you wish to broadcast to the `subscribers`.

```typescript
  public publish(data: any): void {
    this.subscribers.forEach((subscriberFn: Function) => {
      const output: ISubscriberOutput = { topic: this.topic, data };
      subscriberFn(output);
    });
  }
```

We loop over the current `subscribers` and notify each one with an object matching the `ISubscriberOutput` structure.

Overall this implementation keeps things concise and to the point.

## Example usage

An example use case could be an article publisher which notifies subscribers when new articles get published. It could look like this for example:

{% replit @jamesrweb/PubSubExample %}

## Conclusions

I like this pattern and how it allows a scalable and predictable messaging format and how flexible it can be to the needs of what you are building.

I think this ties in nicely with other architectural patterns like the microservices pattern which uses event queues to pass information around in a way that is not too dissimilar to PubSub.

Hopefully you found some value in todays post and you can make use of this pattern in the future!