SocketCluster. The most underrated framework. Part 3: A Pub/Sub example and middleware

maarteNNNN / sc-underrated-framework-pubsub

SocketCluster. The most underrated framework. Part 3: A Pub/Sub example and middleware

Introduction

In this part we will make a simple chat example to understand how Pub/Sub works in SocketCluster. The app can be tested across multiple browser windows. We will add some simple middlewares. A chat history and censorship for bad-words.

Setup

Let's setup a blank project by running socketcluster create sc-pubsub and cd sc-pubsub. Let's install nodemon to restart the server automatically npm i -D nodemon. And for our bad-words censorship we will use a package called bad-words from NPM. npm i -s bad-words. The server can be run with npm run start:watch.

Client code setup (don't give much attention to this, just copy and paste)

We will use vanilla JavaScript in HTML like part 2 shipped with SocketCluster in public/index.html. Let's delete everything inside the style tag and replace it with:

* {
  margin: 0;
  padding: 0;
}

html {
  height: 100vh;
  width: 100vw;
}

.container {
  height: 100vh;
  display: flex;
  align-items: center;
  justify-content: center;
  flex-direction: column;
}

.chat-history {
  height: 70vh;
  width: 75%;
  border: 1px solid #000;
  display: flex;
  flex-direction: column;
  overflow-y: auto;
}

.chat-input {
  width: 75%;
  height: 5vh;
  border-left: 1px solid #000;
  border-bottom: 1px solid #000;
  border-right: 1px solid #000;
}

input {
  box-sizing: border-box;
  width: 100%;
  height: 100%;
  border: none;
  padding: 0 1em;
}

strong,
small {
  font-size: 11px;
  color: gray;
}

.message {
  padding: 0.25rem 1rem;
}

and delete eveything inside <div class="container"> tag and replace it with:

<div id="chat-history" class="chat-history"></div>
<div class="chat-input">
  <input placeholder="message" onkeyup="sendMessage(event)" />
</div>

Okay. Now we have a basic chat page. Nothing too fancy. Now we can focus on getting actual logic of our chat application.

The Pub/Sub functionality

Client

Pub/Sub in SocketCluster is something that can work without writing any backend logic. We can create a channel on the client and the server makes this channel available for other clients.

(async () => {
  for await (const data of socket.subscribe('chat')) {
    console.log(data);
  }
})();

and we should create the function that listens to the enter key on the input to send the publish the message.

const sendMessage = async (event) => {
  if (event.keyCode === 13) {
    try {
      await socket.transmitPublish('chat', {
        timestamp: Date.now(),
        message: event.target.value,
        socketId: socket.id,
      });
      event.target.value = '';
    } catch (e) {
      console.error(e);
    }
  }
};

The transmitPublish method does not suspect a return value. If you do want a response you can look at invokePublish.

The transmitPublish sends an object with a timestamp, message and the socketId. The socket.subscribe('chat') async iterable will log any new data being pushed. Open two browser windows next to each other and open the Developer Tools in both windows. If you send a message in one window it should output it in both consoles.

We will display the messages in the #chat-history div by creating a function that creates an element, changes the text, adds a class and appends the element.

const createMessage = ({ socketId, timestamp, message }) => {
  const chatHistoryElement = document.getElementById('chat-history');
  const messageElement = document.createElement('div');
  messageElement.className = 'message';
  messageElement.innerHTML = `<strong>${socketId}</strong> <small>${timestamp}:</small> ${message}`;

  chatHistoryElement.appendChild(messageElement);

  // Always scroll to the bottom
  chatHistoryElement.scrollTop = chatHistoryElement.scrollHeight
};

change the previous console.log(data) inside the socket.subscribe('chat') to createMessage(data).

Now if we send messages it should display them in the HTML instead of the Developer Tools. Pretty neat, huh? Upon to this point we still didn't do any server-side code.

Server-side

There's only one problem with our app. Every new window does not have any older messages. This is where the server comes in. We will create a middleware that pushes every message to an array, for simplicity sake. Another thing the middleware will pick-up is bad-words. We can filter them and replace the characters with a *.

const Filter = require('bad-words');
const filter = new Filter();

...

const history = []

agServer.setMiddleware(
  agServer.MIDDLEWARE_INBOUND,
  async (middlewareStream) => {
    for await (const action of middlewareStream) {
      if (action.type === action.PUBLISH_IN) {
        try {
          // Censor the message
          action.data.message = filter.clean(action.data.message);
        } catch (e) {
          console.error(e.message);
        }
        // Push to the array for history
        history.push(action.data);
      }
      // Allow the action
      action.allow();
    }
  },
);

...

We set an inbound middleware, we pass it an async iterable stream. On every action of the stream we check if the action.type equals the constant provided by SC action.PUBLISH_IN. If the conditional is true we filter the message and allow the action. Alternatively we could action.block() the action if we don't want it to go through. More on middleware here

To implement the history it's pretty simple, we just create a constant const history = [] and push every action.data to it. Like shown in the above code.

To initially get the history we transmit the data upon a socket connection (e.g. a new browser window).

(async () => {
  for await (let { socket } of agServer.listener('connection')) {
    await socket.transmit('history', history);
  }
})();

And create a receiver on the client which uses a loop to create the messages.

(async () => {
  for await (let data of socket.receiver('history')) {
    for (let i = 0; i < data.length; i++) {
      const m = data[i];
      createMessage(m);
    }
  }
})();

I will try to add an article every two weeks.

Comments

[Tobias Nickel]

I am not sure if the loops are so much better than registering a listener function. Anyway, If someone prefer an other codestyle it would be easy to implement a single helper method, that takes an Iterable and a callback-function.

However I really thank you showing this project. In has a very interesting architecture. I already installed it and made a first test. because I wanted to know if web-clients can access cross domain. And it worked instandly.

If SocketCluster can hold up to all its promisses, I will have a lot of fun with it.

Thanks for this article.

[maarteNNNN]

Thanks for your reply and insight!

I have covered the benefit of the async iterables briefly in part 1. It's basically queuing actions. This is more advantageous than callbacks that do not care about concurrent logic. An example I gave is that database operations will execute according to the queue. Event listeners actions run concurrently, and therefore could potentially generate errors. Upon talking to the creator of SocketCluster, Jon Dubois, he said that in previous versions of the framework - which did not utilize the async iterable, that it was the most common problem.

As for syntax the (async => { ... })() is indeed not a very beautiful syntax. It can be easily forgotten to execute it by adding the (), overall it needs a ; to be able to format it with Prettier for example. However there are some proposals to ECMAScript to resolve these issues. One of them is the async do proposal. Another is top-level await.