How can I improve the publish/subscribe model after reading the Node source code?

“This article is participating in the technical topic essay node.js advanced road, click to see details”

preface

What is publish and subscribe? Based on an event (topic) channel, the object Subscriber that wants to receive the notification subscribes to the topic by defining the event, and the object Publisher that is activated notifies each Subscriber that subscribes to the topic by publishing the topic event.

For example, a popular chestnut – watching drama. There is a TV series that xiao Bao likes very much on a certain platform, and she wants to see the latest progress all the time, but xiao Bao is still very busy and cannot refresh the platform all the time. The platform found this problem and provided a subscription function. Bao chose to subscribe to the TV series. After the update, the platform would immediately send a message to notify Bao. Bao can happily follow the drama.

In the above cases, THE TV series is Publisher, the small package is Subscriber, and the platform assumes the intermediary role of Event Channel.

A few months ago, packet wrote an observer pattern vs a subscription model, don’t be confused again, through the Angle of the martial arts explained the difference between the observer pattern and release a subscription model, dating back to the way certain aspects may increase the understanding of the cost, the article also caused some controversy, packet feel released at the beginning of a subscription model code implementation is not perfect.

Nodejs provides the Event.EventEmitter module. The core of this module is the encapsulation of event triggering and event listener functions. Based on the EventEmitter module, it is relatively convenient to implement the publish and subscribe mode, so the package decided to absorb the essence of EventEmitter source to improve the publish and subscribe mode.

By studying this article, you will learn:

• 🌟 Master the publish and subscribe model
• 🌟 understandNode 中 EventEmitterThe implementation and use of
• 🌟 Master the handwritten publishing and subscription model

EventEmitter

First of all, we will take a look at the source code for EventEmitter. There are a lot of sources in this package.

The init method

There are three main objects in the publish-subscribe pattern, and the event (topic) channel is responsible for maintaining a queue of handlers under an event. So we first need to maintain an event channel, defined in the constructor.

// Store the format of the event channel
const EventChannel = {
  event1: [func1, func2],
  event2: [func3, func4],
};
Copy the code

EventEmitter initializes the event channel properties using EventEmitter. Init. Instead, initialize ObjectCreate(NULL) — object.create.

So why does that happen? Object.create(null) Creates empty objects that have no prototype methods and are pure objects, thus avoiding contamination by prototypes. Object literals {} create empty objects in the same way as new Object(), inheriting the properties of Object objects.

function EventEmitter(opts) {
  EventEmitter.init.call(this, opts);
}

EventEmitter.init = function (opts) {
  if (
    this._events === undefined ||
    this._events === ObjectGetPrototypeOf(this)._events
  ) {
    this._events = ObjectCreate(null);
    this._eventsCount = 0; }};Copy the code

addListener/on

The addListener/on method registers a listener for the specified event, taking a string event and a callback function.

Interestingly, EventEmitter provides two pairs of methods for subscribing and unsubscribing: addListener/on and removeListener/off. When learning the module, the package was especially struggling, but when reading the source code, it suddenly became clear that the two pairs of methods are essentially the same.

EventEmitter.prototype.on = EventEmitter.prototype.addListener;
EventEmitter.prototype.off = EventEmitter.prototype.removeListener;
Copy the code

The on method is internally based on the _addListener method, so the package mainly reads the _addListener method.

Knowledge 1: newListener event

NewListener is an artificially specified event in NodeJS that is triggered when a new monitor is added. It is used in the same way as a regular binding monitor, except that the monitor name is forcibly set to newListener.

var events = require("events");
var eventEmitter = new events.EventEmitter();
eventEmitter.on("newListener".() = > {
  console.log("Bound to a new event");
});
eventEmitter.on("click".() = > {
  console.log("click");
});

// New events are bound
Copy the code

Knowledge 2: The prepend property

The prepend attribute controls the order of different handlers of the same event. For example :(this property is not exposed for external use)

/ / the prepend to false
event.on("click", fn1);
event.on("click", fn2);
event.on("click", fn3);

// The three handlers for the click event in the event channel should run from top to bottom
{
  click: [fn1, fn2, fn3];
}
Copy the code

/ / the prepend to true
// This is just for chestnuts
event.on("click", fn1, true);
event.on("click", fn2, true);
event.on("click", fn3, true);

// The three handlers for the click event in the event channel should run from top to bottom
{
  click: [fn3, fn2, fn1];
}
Copy the code

Read the source code below:

Step1: Get the event channel and the listener for the event to be registered

events = target._events;
// Check whether the event channel exists
if (events === undefined) {
  events = target._events = ObjectCreate(null);
} else {
  // If the newListener event has already been registered, the newListener event will be emitted before subsequent registered events
  if(events.newListener ! = =undefined) {
    target.emit(
      "newListener",
      type,
      // Wait until the once section to elaborate
      listener.listener ? listener.listener : listener
    );

    events = target._events;
  }
  // Get a listener for the event
  existing = events[type];
}
Copy the code

Step2: add a new listener to the event

// No subscription for this event has appeared before
if (existing === undefined) {
  There is no need to declare an array if there is only one handler
  events[type] = listener;
} else {
  if (typeof existing === "function") {
    // Press the new handler into the array
    // Prepend determines the pressing order
    existing = events[type] = prepend
      ? [listener, existing]
      : [existing, listener];
  } else if (prepend) {
    existing.unshift(listener);
  } else{ existing.push(listener); }}Copy the code

removeListener/off

RemoveListener /off removes a listener for the specified event. The listener must be a listener already registered for the event.

Corresponding to the newListener event, NodeJS also sets the removeListener event, which is triggered when a listener is deleted.

The code to remove the listener is relatively simple, so we will comment directly on the source code.

EventEmitter.prototype.removeListener = function removeListener(type, listener) {
  const events = this._events;
  // There is no event channel
  if (events === undefined) return this;

  const list = events[type];
  // This event is not registered with a handler
  if (list === undefined) return this;
  // The current event has only one listener
  // There are two cases where the on register listener is deleted and the once register listener is deleted, as discussed in the once section
  if (list === listener || list.listener === listener) {
    delete events[type];
    // Triggers the removeListener event
    if (events.removeListener)
      this.emit("removeListener", type, list.listener || listener);
    // Remove the listener from the array
  } else if (typeoflist ! = ="function") {
    for (let i = list.length - 1; i >= 0; i--) {
      if (list[i] === listener || list[i].listener === listener) {
        position = i;
        break; }}if (position < 0) return this;
    if (position === 0) list.shift();
    else {
      if (spliceOne === undefined)
        spliceOne = require("internal/util").spliceOne;
      spliceOne(list, position);
    }
    // If there is only one listener, there is no need to use array storage
    if (list.length === 1) events[type] = list[0];

    if(events.removeListener ! = =undefined)
      this.emit("removeListener", type, listener);
  }
  return this;
};
Copy the code

once

Once Registers a single listener for a specified event. That is, the listener is fired only once at most. After the listener is fired, the listener is terminated immediately.

There is a pit in once, and we need to note that once will disarm the listener, but we can also disarm the listener before the once event is executed, so we need to deal with two cases in once.

Case1: the listener is terminated after execution

The difference between the once method and the ON method is that once only executes the listener once and then removes it. Therefore, we can borrow the ON method when designing once and pass in a wrapper function wrapFn that contains the listener method and removes the listener.

eventEmitter.on(event, (. args) = >{ listener(... args); eventEmitter.off(event, listener); });Copy the code

Case2: Call removeListener/off to remove the listener

If you call removeListener/off to remove a listener, it is similar to the removeListener/off method, but in Case1, we are listening for the current listener and the removeListener wrapper function wrapFn. When we call removeListener/off, we pass in the listener method, so we cannot remove it successfully.

So to accommodate this situation, we attach an identifier to the wrapFn function, whose value is listener (wrapfn.listener = Listener). Therefore, when we call the remove method, we can identify listener and listener.listener at the same time.

Once method source:

EventEmitter.prototype.once = function once(type, listener) {
  checkListener(listener);
  // Call the _onceWrap method, which implements the wrapping function above
  this.on(type, _onceWrap(this, type, listener));
  return this;
};
Copy the code

function _onceWrap(target, type, listener) {
  const state = { fired: false.wrapFn: undefined, target, type, listener };
  const wrapped = onceWrapper.bind(state);
  // Case2: Call off to remove the listener and mount the listener on the enclosing function
  wrapped.listener = listener;
  state.wrapFn = wrapped;
  return wrapped;
}
Copy the code

function onceWrapper() {
  if (!this.fired) {
    // Case1 listener execution and listener removal
    this.target.removeListener(this.type, this.wrapFn);
    this.fired = true;
    if (arguments.length === 0) return this.listener.call(this.target);
    return this.listener.apply(this.target, arguments); }}Copy the code

Emit method

The EMIT method executes each listener in listener order, returning true if the event has a registered listener and false otherwise.

The EMIT method is relatively simple to implement. It takes the listener of the corresponding event and executes it by passing in the parameter.

EventEmitter.prototype.emit = function emit(type, ... args) {
  const events = this._events;
  if(events ! = =undefined) {
  1. None (false) 2. Only one (functional) 3. Multiple (array)
  const handler = events[type];
  // Case1 has no value
  if (handler === undefined) return false;
  // Case2 is a function
  if (typeof handler === "function") {
    const result = ReflectApply(handler, this, args);
  } else { // Case3 is an array
    const len = handler.length;
    const listeners = arrayClone(handler);
    for (let i = 0; i < len; ++i) {
      const result = ReflectApply(listeners[i], this, args); }}return true;
};
Copy the code

The source code to harvest

What can we learn from the Nodejs EventEmitter module to improve our publish and subscribe model?

1. Initial value useObject.create(null)Prototype contamination can be avoided
2. There is no need to use arrays when only one listener exists for the event
3. onceTwo cases of method processing
4. offMethod the processing of boundary case and the processing of two deletion cases

Publish and subscribe implementation

With a base of reading source code, we can implement a complete publish/subscribe model.

EventEmitter constructor

function EventEmitter() {
  this._events = Object.create(null);
}
Copy the code

On methods

EventEmitter.prototype.on = function (type, listener) {
  // Get the event channel
  let events = this._events;
  if (events === undefined) {
    events = this._events = Object.create(null);
  }

  // Check whether the newListener event is listened on and execute the newListener callback function if it is listened on
  if(type ! = ="newListener") {
    if (events.newListener) {
      ethis.emit("newListener", type); }}// It doesn't matter if a single listener uses an array packet, so the packet continues to use arrays
  if(! events[type]) { events[type] = [listener]; }else{ events[type].push(listener); }};Copy the code

Off method

In the off method, we need to deal with the removal of the ON register listener and the removal of the once register listener, and at the same time do a good job of boundary case processing.

EventEmitter.prototype.off = function (type, listener) {
  const events = this._events;
  // The boundary condition
  if (events === undefined) {
    return this;
  }
  const listenerList = events[type];
  if (listenerList === undefined) {
    return this;
  }

  // Handle two cases
  events[type] = events[type].filter((fn) = > {
    returnfn ! == listener && fn.listener ! == listener; }); };Copy the code

Once method

As mentioned in the source code section, once handles two cases.

EventEmitter.prototype.once = function (type, listener) {
  // The listener is removed after execution
  const onceApply = (. args) = > {
    listener.call(this. args);this.off(type, listener);
  };
  // Bind identifier, identified as listener
  onceApply.listener = listener;
  // Register the listener
  this.on(type, onceApply);
};
Copy the code

Emit method

EventEmitter.prototype.emit = function (type, ... args) {
  const events = this._events[type];
  // boundary case processing
  if (events === undefined) {
    return false;
  }
  const handler = events[type];
  if (handler === undefined) {
    return false;
  }
  // Execute the listener corresponding to the EMIT event
  handler.forEach((fn) = > {
    fn.call(this. args); });return true;
};
Copy the code

After the language

I am battlefield small bag, a fast growing small front end, I hope to progress together with you.

If you like xiaobao, you can pay attention to me in nuggets, and you can also pay attention to my small public number – Xiaobao learning front end.

All the way to the future!!

An early end to the epidemic will restore peace to the world