Let’s Learn: TCP Servers in NodeJS Using Observables

Who needs HTTP anyways?

Over the years I’ve had to learn a handful of protocols and formats for exchanging data between client and server or between services in an internal system. They all have their own pros and cons and they all help us send a formatted message between workers.

The one that web developers come into contact with the most is HTTP as the protocol. HyperText Transfer Protocol is an application layer protocol or a protocol for how we can communicate between applications. It defines the headers, the status codes, and all of the other parts that every ajax request is familiar with. REpresentational State Transfer ( REST ) is a style to build applications, to format our requests and logic, using the HTTP protocol. These are the basic building blocks and ideas of the current API design and communication between applications.

REST is the perfect solution for a call->response ecosystem that the AJAX revolution brought to light. We could reasonably assume that the client/service could understand what it needed our application to do ( POST to a resource, PATCH a specific resource, etc ) and we could express that clearly in the HTTP request.

But as our applications have grown larger and more complex, the call->response paradigm is starting to turn into a state->action->state paradigm. We no longer can make the assumption that any one service understands how any other service gets updated or even that any other service exists in the first place. We no longer can assume a monolithic interface and instead must build our services to be islands unto themselves. We have finally come to the future of microservices.

Microservices are often built as reactive systems, which are predominately systems built around event-based architecture. This event-based paradigm can employ REST and HTTP communication for gathering and updating state but more interesting to me are the ones that employ a pub/sub relationship between services.

Instead of using HTTP or REST to update services/databases, we are going to create a hub for services to submit events to and allow workers to register to those events. We could add an http server to these files as well but that is a tutorial for another day.

Today we are going to be using the net module in order to create a TCP , or Transmission Control Protocol, server. We are doing this so that our communication mirrors the state->action->state of our architecture, each message being a simple JSON string of basically a redux action. We are also going to use rxjs to do this all using Observables. Just because. Let’s get started!

If you are having trouble copy/pasting or understanding how to run the gists below, check out the github repo of some examples using them here.

First we are going to create our base abstraction for creating a server. Let’s start by creating our module.exports like so:

We create a Map of sockets to ids and a Map of ids to sockets. This is just so that no matter what part of the puzzle we have, we can get the other one. We should probably use a WeakMap here but for toy problems it works.

We also create a helper function called setSocket that will, when given a net.Socket instance, set it inside of the Maps from above and create a uuid for it. We then create a new new.Server instance, allowing for sockets to act similar to a WebSocket. We also create an Observable of connection events from the server. This socketStream will emit a new socket for each connection event on the server object.

Now that we have an observable of every socket being created, let’s set up our logic for handling of the socket:

removeSocket is a lazy function that, given a socket will delay the execution of the returned function, which removes the socket from the sockets and ids maps.

socketObservable takes a socket and adds the socket to the ids and sockets map along with returning a new observable. This observable is a single value ( of({...}) ) that I then merge with all of the values from the same socket's data action. I massage the message a little bit, ensuring a common interface/decoding via the passed config. I also stop taking these actions once the socket emits a close event. And when that event is emitted, I removeSocket, which is the reason I wanted that function to be lazy.

startServer takes a port and starts the process, returning an observable of the socketStream , which is a stream of connection sockets from the server object, which then get turned into socketObservables from the above function.

We have all of the needed pieces for our server to accept and handle requests, now we need to return them to the code that is running this server:

We return an interface to these values, allowing the end user to subscribe to the streams and also to startServer and stop it as well. Now we can listen for socket connections and messages as so:

We simply require the module, configure it with a port to listen on, and subscribe to events that socket s emit. Now we just need a way to send messages to this socket via a TCP Client. Similar to how we focused on the above server in chunks, let’s start with the module export of the client :

client is a TCP connection at config.port . We create a way to decode messages from the client connection via getData. After we have set up our helpers, we get to the meat and potatoes: stream.

stream is an Observable of a single value data: { action: 'CONNECTED' } } which we then merge with the client data event, massaging the msg data via getData. We also tell our worker to stop listening for events when the client emits an error event. Now, we need to export the consumer some way to interface with this:

A simple send function that encodes the message and sends it to the server. Then we create a tearDown function that we can house any logic needed to destroy this connection and reset our state. We can add the following bits of code and let this client send messages from the command line, to make it easier to test later on:

Now we have a way to create a client to talk to our TCP server that we created in the first step. Now let’s create a new client and test it out!

Now if we run both our client instance ( node client.js ) and our server instance ( node server.js ), we should be able to see the following in the server terminal:

CONNECTION this is action
undefined 'this is socket'
Message from the client! this is action
4780f05d-46a2-474a-b32c-f20035827437 this is socket

This is because we get the initial CONNECTION event being sent to the server. We then also send the Message from the client! action once we have connected. If you included the stdIn code in your client module, you can type in valid JSON and it should print the same inside of the server console.

What Next?

So what can we do with this? What can we do in this stream paradigm, this event-based system, that wasn’t easy in HTTP or REST land? We can easily set up a pub/sub system using workers, use the same paradigm/communication for both Browser services ( web/mobile apps ) and Server services ( databases, 3rd party services, etc ). We can totally decouple our consumption of the actions with the creation of the actions.

I have made a simple demo of creating a master that has workers along with a websocket example using the above ideas which you can find here. It’s not meant to be a working prototype of a system using this but a toy project to show how this idea could be implemented. Check it out and leave any feedback/open any issues you may have!