abbitMQ is a powerful message borker allowing engineers to implement complex messaging topologies with relative ease. At the day job we used RabbitMQ as the backbone of our real time data infrastructure. In the previous post we setup a simple PostgreSQL trigger to send change capture messages to a RabbitMQ exchange. Conceptually, this is where we left off:

In this early stage, we basically have a fire-hose that we can selectively tap into. But we have no way to control the flow of data.

To recap a bit before we get too deep, we had a simple and manual way of handling real time operations. Effectively, we just baked all of the logic in the specific application code path. This became complex and difficult to manage. We wanted to automate as much as possible and stop doing it manually.

We had 3 primary goals:

1. Continue sending writes to PG + not manage "real-time" inline
2. Capture and funnel changes through a single point
3. Apply various actions in response to the changes

Staying with the articles theme of the previous post we can talk in terms of some basic business rules

• Anytime someone adds  or modifies an article we want to cache its current state
• Anytime someone adds or modifies an article we want to push a websocket message
• Anytime someone adds a comment to an article, we want to push a websocket message
• When an article goes live the first time, we want to send some data to a reporting system.

Topic Routing

RabbitMQ takes the Smart Server / Dumb Client approach. Most of your interactions with rabbit are with the server. Creating resources, configuring them, stitching them together, etc. It has two basic constructs, exchange and queue and provides a way to bind them together in interesting ways.

Exchanges allow for complex message routing patterns like pub-sub, pattern matching and point-to-point. It is not uncommon to see rather elaborate server setups. RabbitMQ clients read messages from a queue, but it is possible for a message to arrive at a queue from multiple sources. RabbitMQ server do not have any other dependencies.

Think of an exchange as a post office, and a queue as a mailbox. In the case of a topic exchange, a routing pattern can be used to define how messages are routed to queues based on a routing key. Routing keys are words separated by a period ( . ). They can also contain positional wildcards ( * ) or a pound ( # ) which will match zero or more words

For our purposes, this results in having to write a lot less code. Most of what we are going to be building sits on top of RabbitMQ topic routing. In a nutshell, topic routing allows us to route a single message through multiple exchanges and arrive at multiple queues for consumers to handle.  In our setup, PostgreSQL publishes everything the single pg-messaging exchange, and all we need to do is define how things get routed around and what needs to happen when a consumer receives a message.

Realtime Engine

Internally, we referred to the system that defined this logic as the Realtime Engine (RTE).  It is collection of "pipeline" instances. Each pipeline is a RabbitMQ connection that holds one or more node.js streams housed in a Node.js Pipeline instance. The pipeline receives a message from a RabbitMQ Queue, and sends it through its streams.

A Simplified RTE pipeline class looks like this:

const {pipeline, Writable, Passthrough} = require('stream')

class RTEPipeline extends RabbitMQConnection {
  constructor(opts) {
    super(opts)
    if (!Array.isArray(opts.streams)) {
      throw new TypeError(
        'streams is required and must be an array'
      )
    }
     
    // setup exchanges
    // setup queues
    
    this.writer = new PassThrough()
    this.pipeline = pipeline(
      this.writer
    , ...this.opts.streams
    , new Writable({objectMode: true, write: this._drain.bind(this)})
    , this.onFinish.bind(this)
    )
    
    for(const queue of this.opts.queues) {
      queue.onMessage(this._onMessage.bind(this)
    }
  }

  write(buffer) {
    if (buffer === undefined) return true
    return this.writer.write(buffer)
  }

  onFinish(err) {
    // called when pipeline finishes or errors
    if (err) {
      console.error(err)
    } else {
      console.log(`${this.name} pipeline shutdown success`)
    }
  }

  _onMessage(buffer, msg) {
    this.ack(msg)
    return this.write(buffer)
  }

  _drain(buffer, enc, cb) {
    // things to do as the last stage of the pipeline
    // metrics
    // logging
    // other clean up
    
    cb()
  }
}

1. Sets up any declared RabbitMQ Exchanges + bindings
2. Given an array of Duplex Streams, create a stream pipeline. A dummy stream is used as the writable stream to push messages to as the come in from RabbitMQ. Add a single write stream to the end of the pipeline to drain the pipeline
3. Sets up any declared queues, and adds a handler
4. Adds all queue bindings from the primary exchange to each of the queues.

In practice, a pipeline instance would look like this:

'use strict'
const Pipeline = require('./pipeline')
const {Passthrough} = require('stream')

new Pipeline({
  name: 'CRUD-Pipeline'
, exchange: 'cdc'
, exchange_routing: { // exchange (pg-messaging) <> exchange (cdc) routing
    from: 'pg-messaging'
  , to: 'rte-cdc'
  , pattern: 'cdc.*.*'
  }  
, queues: [{
    name: 'rte-pipeline-noop'
  , pattern: [ // cdc -> rte-pipeline-crud bindings
      'cdc.article.*'
    , 'cdc.author.*'
    , 'cdc.comment.*'
    ]
  }]
, streams: [
    new Passthrough()
  , new Passthrough()
  , new Passthrough()
  ]
})

This does the following:

1. Create 2 exchanges if they do not exist cdc and pg-messaging.
2. Route all messages with a cdc routing key prefix to cdc exchange
3. Set up and consume from a queue named rte-pipeline-crud
4. Route all CDC messages for article, author, and comment resource to the rte-pipeline-crud queue.
5. Push every message received through each of the defined streams.

In this case all of the streams are pass-through streams, so nothing actually happens. Visually, this snippet of code looks like this.

Here is a slightly more involved and concrete setup to illustrate some use cases a bit better:

'use strict'

new Pipeline({
  name: 'crud-Pipeline'
, queues: [{
    name: 'rte-pipeline-crud'
  , pattern: [
      'cdc.article.*'
    , 'cdc.author.*'
    , 'cdc.comment.*'
    ]
  }]
, streams: [
    new AMQPParseStream()
  , new CacheStream(['article', 'comment'])
  , new ParallelStream([
      new EmailStream(['author'])
    , new GraphQLSubscriptionStream(['article', 'comment'])
    ])
  ]
})

1. Parse the incoming message
2. Cache all articles + comments
3. Send an update message via GraphQL Subscriptions for article + comment updates
4. Send and email notification to the authors.

This is also very easy to scale out. By running more processes with the same pipelines, we get more consumers on the same queues which gives use higher throughput and more workers to process messages.

With the Pipeline class we are able to automate real time data workflows in a rather modular way. More importantly, it frees us from having to juggle business logic, database transactions and real time messaging logic throughout the codebase. Once a pipeline is setup, we concern ourselves with the business logic; anytime a particular database record is touched (successfully), it will automatically trigger all of the associated data pipelines.

In the next part we'll closer look at writing streams to handle data.

At the day job, we use postgres as the primary database. As communication platform (chat) we do a good deal of real-time whiz-bangery. Being that we are an early stage start-up, we try to follow the keep it simple, stupid (k.i.s.s.) approach. Do the simplest thing possible until it isn't simple anymore.  The simple approach to real time was a very manual and you might see something like this quite a bit:

async function doMessage(user, opts, db) {
  const {message, resource_id} = opts
  const txn = await db.transaction()
  try {
    const resource = await txn.query({
      text: 'INSERT INTO table ...'
    , values: [resource_id, message]
    })
    await tx.commit()
    realtime.publish(user, resource)
    return resource
  } catch (err) {
    txn.rollback()
    log(err)
    // handle errors
  }
  // Other assorted logics
}

This is simple - I can clearly see what it is trying to do. It adds a message by inserting it into a table, and publishes a message over some real-time channel. There are a couple of problems here. The biggest being that the act of adding a message and dispatching some real-time event are tied together. This innocent little function is a function that cannot really be reused. For example:

const doMessage = require('./do-message')

async function doThingAndAddMessage(user, opts, db) {
  const txn = await db.transaction()
  try {
    const result = await txn.query(...)
    await doMessage(user, {
      resource_id: result.id
    , message: opts.message
    }, txn)
    await someOtherThing(user, ...)
  } catch (err) {
    txn.rollback()
    log(err)
    // handle errors
  }
}

If someOtherThing fails, there is a pretty good chance the real-time message is still sent even though there is an error. This generally leads to people writing different but very similar functions and queries to get around this problem. When this happens, there is more than one function that can add messages, and they don't always do the real-time bits; which is another problem. At the end of the day, the basic pattern is pretty simple:

1. modify something in the database
2. Push some message to end-users in real-time telling them what happened so they don't have to reload pages.

Generally speaking these real-time messages tell end users what changed / how it changed. This is usually pretty application specific so it made sense in the simple phase to implement this logic in the code paths that handled user interactions. At some point we needed a to understand how things changed. The previous value and the current value. Which usually requires a read of the data before the write. You can also do this in a single query if you are using a database that implements MVCC, but it gets complicated and cumbersome.

We came to realize that the best way to de-couple and simplify these problems would be the event sourcing pattern. At the time, our applications were interacting with PostgreSQL so heavily, inverting the application to send writes to kafka, would have been too much of an undertaking. What we could do, however, is use PostgreSQL as an event source - set up debezium, funnel data into Kafka, layer on KSQL to pull data back together and sync it back into a database or some other process. That would be a pretty sustainable way to "do real-time".  

So, why not kafka connect? The idea is sound, but as a small team at a small start-up we had a number of issues with the approach.

1. We aren't really comfortable with the java stack / ecosystem. Nor do we want to be
2. We only had 1 devops guy and running kafka in production requires people
3. We were already using RabbitMQ and had people with operational experience
4. The number of moving parts, and complexity that kafka + kafka connect adds didn't seem worth it (added ~8 servers for a small cluster).
5. We had a number of partitioned tables that made things complicated with debezium; topic per table partition which is hard to reason about down-stream

Postgres has facilities for sending messages to remote connections with LISTEN / NOTIFY. However, there are some scalability concerns there. Namely, it requires a single connection ( no pooling ). It also not supported by background workers, which can be a bit problematic. It isn't really intended for high scale messaging. Additionally, we were already using RabbitMQ and we wanted to keep our messaging there. As luck would have it there is a PostgreSQL extension for sending messages via RabbitMQ. With this single extension we can implement basic change capture and use PostgreSQL and RabbitMQ as a real time engine.  

A change capture message is fairly simple, it tells you where the change is coming from ( a table name or resource name ), and how it changed. The HOW can be thought of as a diff  of data. All we want to do is get this diff out of PostgreSQL and into some remote service automatically so we don't have to do it by hand all the time. Here is how we did it

PostgreSQL Setup

To start we need to install the amqp extension and the hstore extension, which ships with most distributions of PostgreSQL. More on this later - it will make more sense in a moment. The amqp package can be installed from a source build or from PGXN.

CREATE EXTENSION IF NOT EXISTS "amqp";
CREATE EXTENSION IF NOT EXISTS "hstore";
GRANT USAGE ON SCHEMA amqp to <USER>;
GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA amqp to <USER>;

The amqp extension will set up some table where config for your RabbitMQ cluster will go. The easiest thing to do is add these settings to your PostgreSQL config and read them in the DB setup migrations

## postgres.conf

amqp.host = rabbitmq # rabbit host hame
amqp.port = 5672
amqp.vhost = '/'
amqp.username = guest
amqp.password = guest
amqp.exchange = 'pg-messaging'

You can use the CURRENT_SETTING sql function to read arbitrary values out of the PostgreSQL config.

INSERT INTO amqp.broker (host, port, vhost, username, password)(
  SELECT
    COALESCE(current_setting('amqp.host'), NULL)
  , COALESCE(current_setting('amqp.port')::INT, NULL)
  , COALESCE(current_setting('amqp.vhost'), NULL)
  , COALESCE(current_setting('amqp.username'), NULL)
  , COALESCE(current_setting('amqp.password'), NULL)
);

If you have ever written a ROW level trigger for PostgreSQL, you are probably familiar with the special variables NEW and OLD which represent the previous state of the row being modified, and the current state of the row being modified. We convert this data to JSON, and send it to a RabbitMQ exchange. Our change capture message looks like this:

{
  "previous": Object // JSON representation of the previous row
, "current": Object // JSON representation of the current row
, "targets": Array //  A list of columns that actually changed
, "timestamp": Date // Date / time of the operation
, "operation": String // WHAT happened (insert, update, delete)
, "resource": String // An application specific name of the thing being chagned
, "tablename": String // The name of the table
, "routing_key": String // The rabbitmq routing key that was used
}

This gives us everything we need to know

1. WHAT changed ( the operation, table + resource )
2. WHEN it changed ( a timestamp )
3. HOW it changed ( previous / current + targets )

Change Data Trigger

To make intelligent decisions in applications that may want to consume this. All that is left is to write the Trigger function to send it out into the world. I'm going to focus on the update operation case, because its the most interesting.

CREATE OR REPLACE FUNCTION AMQP_CHANGE_DATA_CAPTURE()
  RETURNS TRIGGER AS $$
    DECLARE
      routing_key_prefix TEXT
      routing_key
      resource TEXT
    BEGIN
      resource := COALESCE(TG_ARGV[0], TG_TABLE_NAME);
      routing_key_prefix := COALESCE(TG_ARGV[1], 'cdc');
      routing_key := LOWER(FORMAT('%s.%s.%s', routing_key_prefix, resource, TG_OP));
    
      IF TG_OP = 'INSERT' THEN
      -- SNIP
      RETURN NEW;
      END IF;

      IF TG_OP = 'UPDATE' THEN
        PERFORM amq.publish(
          1 -- id of broker in amqp.broker table
        , CURRENT_SETTING('amqp.exchange') -- from postgres.conf
        , routing_key
        , JSON_BUILD_OBJECT(
            'previous', ROW_TO_JSON(OLD)
          , 'current',  ROW_TO_JSON(NEW)
          , 'targets',  AKEYS(HSTORE(NEW) - HSTORE(OLD))
          , 'timestamp', CURRENT_TIMESTAMP
          , 'operation', LOWER(TG_OP)
          , 'resource', resource
          , 'table', TG_TABLE_NAME
          , 'routing_key', routing_key
          )::TEXT;
        RETURN NEW;
      END IF;

      IF TG_OP = 'DELETE' THEN
      -- SNIP
      RETURN NULL;
      END IF;

    END;
  $$ LANGUAGE plpgsql;

There are a couple of things to note here.

, 'targets',  AKEYS(HSTORE(NEW) - HSTORE(OLD))

• HSTORE is used because it implements subtract  hstore - hstore, which removes matching K/V pairs.  The AKEYS function just returns an array of keys of the hstore value. In other words, this gives us the columns that were actually changed.

resource := COALESCE(TG_ARGV[0], TG_TABLE_NAME);

• TG_ARGV is an array of arguments passed to the trigger definition. In this case you can optionally pass a resource name and a routing key prefix if you want to route messages do a different destination. This solves our partitioned table problem. We can give them a common name. For example, Instead of _table_1 and _table_2 we can can name and handle them in code as just table.

trigger [trig'er] -n, noun

1. a device, as a lever, the pulling or pressing of which releases a detent or spring

2. procedural code that is automatically executed in response to certain events on a particular table

3. special stored procedure that is run when specific actions occur within a database.

With the trigger function defined, we can selectively add change capture to any table in our database. We can specify what events that trigger the messages ( or all of them ), what they are called and how they get routed around.

CREATE TRIGGER article_change_data_capture_trig
  AFTER INSERT OR DELETE OR UPDATE
  ON blog_articles
  FOR EACH ROW
    EXECUTE PROCEDURE AMQP_CHANGE_DATA_CAPTURE('article');

NOTE: Always use AFTER events for these triggers. This ensures that the message only is sent when the transaction succeeds.

Our RabbitMQ routing keys will look like this

• cdc.article.insert
• cdc.article.delete
• cdc.article.update

Now for every insert, update and delete on the blog_articles table, A CDC messages will be sent to the pg-messaging exchange on our RabbitMQ server.  This falls inline with the Keep It Simple, Stupid mentality. Not only did we successfully avoid any additional servers, complicated infrastructure or expensive hosted services, it only took about ~80 lines of SQL.  There are certainly some trade-offs you'll need to consider with a set up like this:

The biggest downside here is that it makes use of a non-standard ( albeit a very stable one ) extension. In most cases, managed services ( AWS Aurora, Google Cloud SQL, etc. ) do not support these types of extensions, nor do they give you access to install them. If you are currently using or do not want to run your own database servers, this is likely not the most appropriate option.

With 1 trigger and 1 extension we've made a data firehose. In Part 2, we'll take a look at how to make the data work for us with a little bit of help from Node.js

Between the Foreign key, which verifies values in another table, and the Check constraint, which verifies values in a specific column, you can typically accomplish just about everything you need to do rather easily. The main problem the you'll run into is that these kinds of constraints are restricted to a single column on a table. Additionally they only apply the the current value in that column. This restriction doesn't apply to custom constraint, which is a special kind of trigger. This effectively gives you the full power of SQL to implement much more complex constraint conditions.

CREATE TYPE states AS ENUM ('one', 'two', 'three');
create table example (
  id INT GENERATED ALWAYS AS IDENTITY
, state states NOT NULL  
);

A simple example to illustrate the utility of a constraint trigger is a simple state machine. Here, our table has a state field that has a custom ENUM column. This ensures that the column can only contain one, two, or three. However, I want to make sure that the initial value is always one. Additionally, I want to make sure that, during update operations, that the state value can only be set to two if the previous value was one and can only be set to three if the previous value was two. If any of those conditions are not met, that operation should fail and abort the transaction.

+ -- ONE --> -- TWO --> -- THREE -- +

This kind of logic is generally pretty difficult to capture with traditional check constraints. At this point this is usually where most application developers will turn to doing this work at the application layer by pulling the current record, comparing the input value to the existing value and making a decision. Careful, of course, to account for the situation where there is no record and taking the appropriate action. The problem here is that doing it this way creates a data race between the running instances of the application in the time it take to read / check the record and updating or inserting value. It is an unnecessary bit of complexity when postgres can do all of that in a single step with a constraint

Trigger [tri'ger] -n., --noun

1. special stored procedure that is run when specific actions occur within a database.
2. cause an event or situation to happen or exist.

To do this we create a function like any other function that RETURNS TRIGGER. The function must do one of three things:

1. Raise an exception to abort the transaction
2. Return a record/row value having exactly the structure of the table the trigger was fired for
3. Return NULL

CREATE OR REPLACE FUNCTION example_fsm_check()
RETURNS TRIGGER AS 
$$
  BEGIN
    RETURN NEW;
  END;
$$ LANGUAGE PLPGSQL VOLATILE;

Right now, this is a no-op. What we need to do is determine if the transaction is an INSERT and check for the value one or check for the transition values if it is an UPDATE. Postgres injects some additional information into trigger functions that you can inspect to make these kinds of determinations. In this case we want TG_OP.

CREATE OR REPLACE FUNCTION example_fsm_check()
RETURNS TRIGGER AS 
$$
  BEGIN
    IF TG_OP = 'INSERT' AND NEW.state != 'one' THEN
      RAISE EXCEPTION 'Invalid FSM State %s', NEW.state
    END IF;
    RETURN NEW;
  END;
$$ LANGUAGE PLPGSQL VOLATILE;

Now things are getting interesting. In the above snippet, if the operation is an insert, and the state column is NOT one, raise an exception. We can make use of  TG_OP to tell us that the operation is an insert, and the special NEW object to inspect the data being inserted. This prevents the initial state from being anything other than the value we want and raises an exception if it is.

INSERT INTO example (state)
VALUES ('two')

ERROR:  Invalid FSM State twos
CONTEXT:  PL/pgSQL function example_fsm() line 5 at RAISE
SQL state: P0001

The next thing to do is handle the allowable state transitions during updates. Like most other programming languages, plpgsql has some facilities for logic branching. Additionally, during an UPDATE operation, postgres gives trigger functions access to the OLD record as well as the NEW record

-- on update
IF TG_OP = 'UPDATE' THEN
  -- if state changes
  IF OLD.status IS DISTINCT FROM NEW.status THEN
    -- only from one -> two
    IF OLD.status = 'one' AND NEW.status = 'two' THEN
      RAISE EXCEPTION 'Invalid state transition';
    END IF;
		
    -- only from two -> three
    IF OLD.status = 'two' AND NEW.status != 'three' THEN
      RAISE EXCEPTION 'Invalid state transition';
    END IF;
  END IF;
END IF;
-- end update

Here, we add a couple of if checks. The main if condition is checking if status column has changed at all - if not, there is nothing to do. However, if it has changed, there are two additional conditions that make up the final logic of our little state machine. If the old status value was ONE, and the new value is not TWO, then raise an exception. Similarly if the value was TWO and the new value is not THREE, also raise an exception.

Lastly, we can attach the function to a table as a CONSTRAINT trigger like normal.

CREATE CONSTRAINT TRIGGER check_sample_fsm
AFTER INSERT OR UPDATE
ON example
FOR EACH ROW
EXECUTE procedure example_fsm_constraint();

That's it! we've defined our own constraint. Now unlike a regular trigger procedure, a constraint trigger can define when it is triggered as well as conditions that define when the trigger is executed. For example, if we want the constraint to fire at the end of a transaction and only when the state is set to one, we can easily add these instructions to the trigger using the DEFERRABLE option and the WHEN condition clause.

CREATE CONSTRAINT TRIGGER check_sample_fsm
AFTER INSERT OR UPDATE
ON example
DEFERRABLE INITIALLY DEFERRED 
FOR EACH ROW
WHEN NEW.state = 'one'
EXECUTE procedure example_fsm_constraint();

Trigger constraints give us the ability to define very complex triggers that extend well beyond what one could do with CHECK constraints. They could even be used, for example, to build out FOREIGN KEY like constraints that check multiple tables or other points of data that can't be accomplished otherwise.

Trigger Constraints - They are pretty cool.

At the day job, I am in the process of migrating a number of applications off of a document store to postgres. One of the features we had was custom attributes which effectively allowed end users to associate arbitrary data with various entities.  In addition to just porting the ability to store user defined data, we now need to allow end users to be able to query for the associated entity via the attributes they have defined. This includes ability to sort by, apply range (numeric or date) filters and do full text phrase matching over text values.

For example, ACME Co., may want to define attributes about their users like this:

• ip_address (text)
• name (text)
• login_date (timestamp)
• max_sale_value(number)
• total_purchases(number)
• vip_customer (boolean)

From this data they might want ask arbitrary questions of this data to find and gain insights of users. For example

Find all users who have logged in within the last 3 months, have more than 10 total purchases but are not vip and whose name starts with J

Translated roughly to SQL:

SELECT *
FROM user_attributes
WHERE
  login_date BETWEEN NOW() - INTERVAL '90 DAY' AND NOW()
AND total_purchases > 10
AND name ILIKE 'j%'
AND vip = FALSE

Another company, WizzBang LLC., want to store a different set of data

• name (text)
• phone_number (text)
• dog_lover (boolean)
• cat_lover (boolean)
• pet_count (number)
• first_pet (timestamp)
• favorite_dog_breed (text)
• favorite_cat_breed (text)
• award_count (number)

Not only is the data they store very different both physically and contextually, but the are going to ask very different questions of it than ACME. Of course, they are both large companies and estimated ~10M users. This means we are going to need a way to index their data to support the queries they will need to run so the application doesn't grind to a stop. There are some options out there.

EAV

The Entity Attribute Value (EAV) design has been around for a while. It has pros and cons. For our situation, on the upside, the column that stores the actual value is a physical, typed column that we can perform a variety of operations on. Additionally, because it is a column, we could index it for better performance.

EAV schemas try to make the rigid relational model and make it bit more flexible. An entities table defines basic things (Car). An attributes table defines the properties of an entity (weight), and the attribute_values table holds the value for a property associated with an entity. For our needs, the primary problem is that there is only a single value column, meaning they all have to be the same type. That makes it hard to support dates, boolean, or numeric types.

JSON (JAY'sun) -n., noun

We can serialize our data as JSON and everything will be easy

1. An open-standard file format that uses human-readable text to transmit data consisting of attribute-value pairs and array data types

JSON

You might be thinking JSON! think again. If all we needed to do was store the data, you might be on to something. But we have to do a bit more with it and this gets hard. In particular:

1. It is really just text under the hood which means we have to do a lot of manual type casting of specific fields and doing that won't use indexes (slow)
2. To be able to do any kind of searching, range queries, etc, we need to know what kind of data is in there anyway. And if we have to control that - we might as well just use a table
3. Doing joins off of values in JSON columns is tough. We either have to add columns to the table to help out, or put the JSON column on the source table to avoid having to do join - like a doc store

The real deal breaker is doing queries that target specific fields within the JSON need to know what kind of data is in the field and explicitly cast it to perform any kind of filtering other than exact matching.

-- SAMPLE TABLE w/ JSON COLUMN
CREATE TABLE sample(
  id SERIAL
, word TEXT
, json_data JSONB NOT NULL
);
CREATE INDEX json_data_gin_indx ON sample USING GIN(json_data);

--- INSERT SOME RANDOM DATA
INSERT INTO sample (word, json_data)(
  SELECT
    encode(gen_random_bytes(10), 'hex')
  , JSON_BUILD_OBJECT(
      'value', iter
    , 'sample', encode(gen_random_bytes(10), 'hex')
  ) from generate_series(1, 2000000) as iter
)
-- > Affected rows: 2000000
-- > Time: 76.108s

-- TRY TO QUERY FOR SOME VALUES
select *
from table_b
where json_data->>'a' < 2

-- > ERROR:  operator does not exist: text < integer

while the ->> operator gets you access to field values within the JSON, they are accessed by the string representation, which greatly limits the ways it can be queried. To get around the issue you would have to manually cast all fields

select *
from table_b
where 
  CAST(json_data->>'a' AS INTEGER) < 2

This isn't very complicated to do itself, and it will work a short while. The problem here is that this kind of query can't make use of indexes. While it is possible to index JSON columns with GIN indexes, You are fairly limited when it comes to the kinds of queries will make use of indexes. Once the table grows large enough, we are going to need a different solution

What we really want is an actual postgres table with typed columns that gives us all of the powerful query predicate and indexing power that we know and love. I want a flexible schema. To get the functionality we need to implement the described feature, we're going to need to rethink this.

Attribute Table

This is actually a pretty good use case for table inheritance. PostgreSQL table inheritance takes a page directly from object orientated programming practices. Tables can inherit columns and attributes from a common parent table. Every child table will have all of the columns that are added or removed from the parent but are still able to define their own columns to make up a unique schema. This is how we can achieve a flexible schema with physical tables. We can make this process mostly automatic with triggers.

-- base table
CREATE TABLE custom_attributes (
  id UUID NOT NULL default gen_random_uuid()
, user_id UUID NOT NULL
, created_at TIMESTAMPTZ DEFAULT NOW()::TIMESTAMPTZ
, company_id UUID NOT NULL
);

-- TRIGGER: When a new company record is inserted
-- Generate an attribute table for there custom data
CREATE OR REPLACE FUNCTION generate_custom_attribute_table()
  RETURNS TRIGGER AS $$
  DECLARE
    tablename TEXT;
    ident TEXT
    company_id TEXT
  BEGIN
    company_id := COALESCE(NEW.id, gen_random_uuid())::TEXT;
    ident := REPLACE(company_id, E'\-', ''::TEXT);
    tablename := 'user_attributes_' || ident;
    EXECUTE FORMAT(
      'CREATE TABLE %s (
        CONSTRAINT %s_pkey PRIMARY KEY (user_id, company_id)
      ) INHERITS (custom_attributes)'
    , tablename, tablename
    )
    RETURN NULL;
  END;
  $$ LANGUAGE plpgsql VOLATILE;

CREATE TRIGGER on_company_insert
AFTER INSERT
ON company FOR EACH ROW
EXECUTE procedure generate_custom_attribute_table();

When a new record is inserted into the company table, a trigger creates a table that inherits from the custom_attribute table. This is the table that will contain the physical columns to store data points defined by a company. Initially, this table doesn't have any columns that deviate from the base table.

Now, we are going to need a way to allow administrators to manage their schema with out giving them direct access to the internal table. For that we are going to need a catalog that keeps record of the field names and their respective types that can be used as the public interface.

Catalog Table

We'll store the attributes definitions our companies want to store in a table that we can simply add / remove records from. Basically all we need is the name of the field and the type of data that will be stored - In this case we are going to constrain that to just number, boolean, text and date fields.

CREATE DOMAIN user_attribute_type AS TEXT
  CONSTRAINT user_attribute_field_chk
    NOT NULL CHECK(VALUE IN ('boolean', 'date', 'number', 'text'));

CREATE TABLE IF NOT EXISTS user_attribute_catalog (
  id UUID NOT NULL gen_random_uuid()
, company_id UUID NOT NULL REFERENCES company(id)
, field_name text NOT NULL
, field_type user_attribute_type
, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
, CONSTRAINT catalog_attribute_name_uniq (company_id, field_name)
);

When we want to show all of the defined attributes for a company, it is a pretty simple query

SELECT field_name, field_type
FROM user_attribute_catalog
WHERE company_id = $1

The last piece to this is going to be a couple of triggers to add and remove the appropriate columns to the child table when records are added or removed from the catalog table.

Inserting a record into the catalog table adds a column to the respective attribute table for the company. The type of the column is determined by the field_type column of the inserted record

-- ADD ATTRIBUTE
CREATE OR REPLACE FUNCTION public.add_user_attribute()
RETURNS TRIGGER AS $$
  DECLARE
    statements TEXT;
    tablename TEXT;
    ident TEXT;
    organization_id TEXT;
    field TEXT;
    BEGIN
    ident := REPLACE(NEW.company_id::TEXT, E'\-', ''::TEXT);
    tablename = 'user_attributes_' || ident;
    field := NEW.field_name || '_' || ident

    IF
      NEW.field_type = 'boolean' THEN
        statements := FORMAT('
          ALTER TABLE %s
              ADD COLUMN IF NOT EXISTS %s BOOLEAN
        ', tablename, field);

      ELSEIF NEW.field_type = 'number' THEN
        statements := FORMAT('
          ALTER TABLE %s
              ADD COLUMN IF NOT EXISTS %s FLOAT8
        ', tablename, field);

      ELSEIF NEW.field_type = 'date' THEN
        statements := FORMAT('
          ALTER TABLE %s
              ADD COLUMN IF NOT EXISTS %s TIMESTAMPTZ
        ', tablename, field);
      ELSEIF NEW.field_type = 'text' THEN
        statements := FORMAT('
          ALTER TABLE %s
              ADD COLUMN IF NOT EXISTS %s zdb.fulltext
                CONSTRAINT ua_%s_chk CHECK(char_length(%s) <= 512)
        ', tablename, field, field, field);
      ELSE
        RAISE EXCEPTION 'Unknown attribute type: %s', NEW.field_type;
    END IF;
    EXECUTE statements;
    RETURN NULL;
  END;
  $$
  LANGUAGE plpgsql VOLATILE;

CREATE TRIGGER on_insert_user_attribute_fields
 AFTER INSERT
 ON user_attribute_fields  FOR EACH ROW
 EXECUTE PROCEDURE add_user_attribute();

-- REMOVE ATTRIBUTE
CREATE OR REPLACE FUNCTION public.remove_user_attribute()
  RETURNS TRIGGER AS $$
  DECLARE
    tablename TEXT;
    ident TEXT;
    field TEXT;
    BEGIN
      ident := REPLACE(OLD.company_id::TEXT, E'\-', ''::TEXT);
      tablename := 'user_attributes_' || ident;
      field := OLD.field_name || '_' || UUID_TO_IDENT(OLD.id);
      EXECUTE FORMAT('
        ALTER TABLE %s
          DROP COLUMN IF EXISTS %s
      ', tablename, field);
    RETURN NULL;
    END;
  $$ LANGUAGE plpgsql VOLATILE;

CREATE TRIGGER on_delete_user_attribute_fields
  AFTER DELETE
  ON user_attribute_fields
  FOR EACH ROW
  EXECUTE PROCEDURE remove_user_attribute();

These two trigger functions handle managing the internal table definition that will hold the actual attribute data. When a record is inserted into the catalog table, a column is added. Conversely, when record is deleted from the catalog, it corresponding column is removed.

It is important to note that the column being added and removed have no constraints and no defaults. Doing it this way avoids having to re-write entire tables and is mostly a change in the metadata about the table. These operations can be preformed on production tables with multiple millions of records in milliseconds.

Inserting into the Company table will generate an attributes table specifically for that company

INSERT INTO company (id, name)
VALUES ('02335c58-a66f-4b53-9801-cb8045e2e848', 'ACME');

# list tables
\dt
company
custom_attributes
custom_attribute_catalog
user_attributes_02335c58a66f4b539801cb8045e2e848

By default, it will only have the columns that were defined by the parent attributes table.

# user_attributes_02335c58a66f4b539801cb8045e2e848

+-------------+------+
| Column Name | Type |
| ----------- | ---- |
| id          | UUID |
| company_id  | UUID |
| user_id     | UUID |
+-------------+------+

Adding a field definition to the catalog adds a column of the appropriate type to the child table that was set up for the specific company.

INSERT INTO user_attribute_catalog
(company_id, field_name, field_type)
VALUES (
  '02335c58-a66f-4b53-9801-cb8045e2e848'
, 'fake_number'
, 'number' 
);

# user_attributes_02335c58a66f4b539801cb8045e2e848

+----------------------------------------------+--------+
| Column Name                                  | Type   |
|----------------------------------------------|--------|
| id                                           | UUID   |
| company_id                                   | UUID   |
| user_id                                      | UUID   |
| fake_number_6ad3e88314e04832b39daef8fa7ff730 | DOUBLE |
+----------------------------------------------+--------+

Removing records for from the catalog table will additionally drop the column from the attribute table

Indexing

We've solved the flexible schema problem, but we still have the issue of needing to perform complex user defined queries on a large data set. We need to index this, and to do that we are going to use elasticsearch via the fantastic zombodb extension.

Once the extension is installed and configured, we need to update the first trigger that generates the attributes table to add an elasticsearch index.

It is an index like any other postgres index, it just happens to live in elasticsearch

  BEGIN
    company_id := COALESCE(NEW.id, gen_random_uuid())::TEXT;
    ident := REPLACE(company_id, E'\-', ''::TEXT);
    tablename := 'user_attributes_' || ident;
    EXECUTE FORMAT(
      'CREATE TABLE %s (
        CONSTRAINT %s_pkey PRIMARY KEY (user_id, company_id)
      ) INHERITS (custom_attributes);
     CREATE INDEX IF NOT EXISTS %s_zdx ON %s USING zombodb((%s.*)) WITH (alias = ''%s'')
      '
    , tablename, tablename, tablename, tablename, tablename, tablename
    );
    RETURN NULL;
  END;

That is all there is  to it. The simplest way to think about how zombodb and postgres work together is this - It is an index like any other postgres index, it just happens to live in elasticsearch; In the same way postgres manages a BTREE index on disk, it will manage a zombodb index in elasticsearch. Zombodb indexes the entire row, so as columns are added and values are populated, the index is kept up to date. This means we can perform just about any kind of search query elasticsearch supports. We can update the SQL query for the data set defined for ACME to use zombodb

SELECT *
FROM user_attributes_02335c58a66f4b539801cb8045e2e848
WHERE user_attributes_02335c58a66f4b539801cb8045e2e848 ==>
dsl.limit(25, 
  dsl.must(
    dsl.term('vip', 'false')
  , dsl.range(field=>'total_purchases', gt=>10)
  , dsl.range(
      field=>'login_date'
    , lte=>'2018-12-30'
    , gte=>'2018-09-30'
    )
  , dsl.wildcard('name', 'j*')
  )
)

This tells PG to use the zombodb index which makes the appropriate query to your elasticsearch cluster, and will still return the results out of the table. You can still use joins, procedures and everything else postgres provides. We get a strongly typed flexible schema in a relational database backed by a blazing fast elasticsearch index.

It may not be magic - but it is damn close

lasticsearch is an extremely powerful searchable document store. The more you use it, the more you learn about the more realize how deep the rabbit hole of possibility goes. Except for when it comes null values. The Achilles heel of elasticsearch. What it really boils down to is elasticsearch doesn't index null values or fields that are missing from a document. Even if you have set up an index mapping and told elasticsearch about your field.

As you might think it can get a little tricky searching document for a field value or where that field is null / missing - especially when combined with other field queries. For example lets say I have some documents that looks something like this

[{
  "name": "bill"
, "age": 25
, "fav_color": "blue"
}, {
  "name": "sam"
, "age": "30"
, "fav_color": null
}, {
  "name": "fred"
, "age": "21"
, "fav_color": "red"
}]

If I want to find all documents where age greater than or equal to 25 and fav_color is blue or null you could think of this in terms of SQL

SQL ˈsē-kwəl -n --noun

1. domain-specific language used in programming and designed for managing data held in a relational database management system
2. the next installment (as of a speech or story); especially : a literary, cinematic, or televised work continuing the course of a story begun in a preceding one

SELECT 
 id
, name
, age
, fav_color
FROM
  users
WHERE
  age > 25
AND
  (fav_color = blue OR fav_color IS NULL)

You could do this with bool queries, but it gets a little complex - even for small queries.

{
  "query": {
    "constant_score": {
      "filter": {
        "bool": {
          "must": [{
            "range": {
             "age": {
               "gte": 25
             }
            }
          }, {
            "bool": {
             "should": [{
               "bool": {
                "must": {
                  "term": {
                   "fav_color": "blue"
                  }
                }
               }
             }, {
               "bool": {
                 "must_not": {
                   "exist": "fav_color"
                 }
              }
             }]
            }
          }]
        }
      }
    }
  }
}

This style of queries is really nice if you want to create a generic query builder for elasticsearch as everything can be reduced to a bool query at the end of the day. But elasticsearch has a short cut for this with the dis_max query which allows you to combine queries. When the query uses the same field, it is OR'd together, and you can even specify tie breaker values if you need to denote which values are more important. So we could combine the above like so

{
  "dis_max": {
    "queries": [{
      "term": {
        "fav_color": "blue"
      }
    }, {
      "bool": {
        "must_not": {
          "exists": "fav_color"
        }
      }
    }]
  }
}

Unfortunately, there is not a missing query type, so we still have to use a bool with a must_not clause. But, now we can replace most of the nested bool queries with the dis_max query and elasticsearch will do the work of translating it into a bunch of bool queries for us.

{
  "query": {
    "constant_score": {
      "filter": {
        "bool": {
          "must": [{
            "range": {
              "age": {
                "gte": 25
              }
            }
          }, {
            "dis_max": {
              "queries": [{
                "term":{
                  "fav_color":"blue"
                }
              }, {
                "bool": {
                  "must_not": {
                    "exists": "fav_color"
                  }
                }
              }]
            }
          }]
        }
      }
    }
  }
}

lasticsearch is an amazing piece of technology. Built on top of luecine it offers all of he incredible search facilities that you'd expect from a full featured search. What makes elasticsearch so powerful, however, is the fact that it stores the actual data that was originally index as JSON documents. Basically, it is a Full Text Search Database more so than a search engine. This allows elastic search to do things that other search engines can't do like aggregations, scripted queries, multi-query searches, etc; All in addition to the expected searching capabilities like suggestions, spelling corrections, faceting, and so on. For these reasons people are using elasticsearch as the primary data store for massive amounts of data.

One thing that is also uniquely different with elasticsearch is that it's query language is just JSON objects. You'll see examples like this:

{
  "query": {
    "match" : {
      "message" : "this is a test"
    }
  }
}

This is a pretty simple search query. match is a full text search that support wildcard matching, partial word matching, fuzzy matching and the like. Many queries work this way and this is great when your search criteria are broad. Very much like how you would expect any search engine to work today. If you are using elasticsearch as a primary data store, you'll want to do more complicated things like you are probably used to doing with SQL. At first glance it is not apparently obvious that elasticsearch can do this. However, the query DSL has a fantastic gem of a query type - compound bool that lets you combine multiple query fragments together. The meat of it looks like this:

{
  "query": {
    "constant_score": {
      "filter": {
        "bool": {
          "must": []
          "should": []
        }
      }
    }
  }
}

Search (sərCH) -v, verb., -n, noun;

1. try to find something by looking or otherwise seeking carefully and thoroughly
2. an act of searching for someone or something

The two important parts of the bool query, are the should and must. These keys have very SQL-like behaviors behind them.

• should <> sql OR
• must <> sql AND

Let's take a look at some more concrete examples.

Shirt Document

Lets say we have an index of shirts for some made up e-commerce application. It's a basic data set, but we want to let our uses find exactly what they are looking for with laser like precision.

{
  "brand": "Gildan"
, "size": "XL"
, "color": "black"
, "type": "long-sleeve"
, "price": 19.00
, "gender": "men"
, "age": "adult"
}, {
  "brand": "Hanes"
, "size": "M"
, "color": "purple"
, "type": "hooded-sweatshrt"
, "price": 22.50
, "gender": "female"
, "age": "toddler"
}

If we want to find all of the Large mens apparel we could add 2 term queries to the must block.

{
  "query": {
    "constant_score": {
      "filter": {
        "bool": {
          "must": [{
            "term": {
              "size": "L"
            }
          }, { 
            "term": {
              "gender": "men"
            }
          }]
        }
      }
    }
  }
}

This translates to SQL almost literally, and will return documents for Men's Large apparel.

SELECT *
FROM shirt
WHERE (size = 'L' AND  gender = 'men')

The queries are going to get a little complex, so we can leave off the query envelope bits and deal with the filter block itself. From here, if we just move the query fragments from the must block to the should block and we'll get a slightly a different query.

Elasticsearch

{
  "bool": {
    "should": [{
      "term": {
        "size": "L"
      }
    }, {
      "term": {
        "gender": "men"
      }
    }]
  }
}

SQL

This will return documents for apparel of size Large, OR apparel for men

SELECT *
FROM shirt
WHERE (size = 'L' OR  gender = 'men')

Red Pill (rəd' pil) -n., --noun;

Take the red pill - you stay in wonderland and we shall see how deep the rabbit hole goes

1. A dose of the truth
2. An eye opening experience

Simple enough, and there is a lot that you can do with this. Now, there is one simple fact that isn't apparently obvious, and it is a bit of a red pill. The should and must arrays take any number of query fragments. Bool is a query fragment. This means they are nest-able, and the rabbit hold runs deep.

Elasticsearch

{
  "bool": {
    "should": [{
      "bool": {
        "must": [{
          "term": {
            "color": "black"
          }
        , {
            "term": {
              "gender": "men"
            }
          }
        }]
      }
    , {
        "bool": {
          "must": [{
            "term": {
              "size": "XL"
            }
          , {
              "term": {
                "type": "hooded-sweatshirt"
              }
            }
          }]
        }      
      }
    }]
  }
}

SQL

SELECT *
FROM shirt
WHERE  ( 
  ( color = 'black' AND gender = 'male') 
  OR
  ( size = 'XL' AND type = 'hooded-sweatshirt')
)

You aren't limited to single term queries either. You can use any legal query - terms, multi_match, range, etc., at any level.

Elasticsearch

{
  "bool": {
    "should" [{
      "bool": {
        "must": [{
          "terms": {
            "color": [ "red", "black", "white"]
          }
        }, {
          "term": {
            "gender": "men"
          }
        }, {
          "terms": {
            "type": ["hooded-sweatshirts", "long-sleeve"]
          }
        }]
      }      
    }, {
      "bool": {
        "must": [{
          "term": {
            "color": "blue"
          }
        }]
      , "should": [{
          "bool": {
            "must": [{
              "gender": "female"
            }, {
              "type": "hooded-sweatshirt"
            }, {
              "range": {
                "lte": 45.00
              }
            }]
          }
        }, {
          "bool": {
            "must": [{
              "term": {
                "age": "toddler"
              }
            }, {
              "term": {
                "brand": "Hanes"
              }
            }]
          }
        }]
      }
    }]
  }
}

Complex, but still legal. This translates rather easily and still does what you think it would.

SQL

SELECT * 
FROM shirt
WHERE (
    ( 
      color IN ('red', 'black', 'white') 
      AND
        gender = 'men'
      AND
        type IN ('hooded-sweatshirt', 'long-sleeve')
    )
    OR 
    (
      (color = 'blue')
      AND (
        ( gender = 'female' AND type = 'hooded-sweatshirt' AND price <= 45.00)
        OR 
        (age = 'todler' AND brand = 'Hanes')
      )
    )
)

Crazy! Moreover, these are still search queries using your indexed content. Your queries will return in under ~10ms over millions of records. It's power and speed make elasticsearch one of most popular the data stores of choice for real-time big data analytics, monitoring and sensor data aggregations and big commerce.

Elasticsearch - It's pretty good.

oing to start doing some live coding on twitch! I've been in the open source space for a long while and it seems fitting that this is the next step. I don't really have a schedule in mind or much else lined out. Other than - It's probably happening

Check out my channel on twitch. I'll do my best to send out a heads up when I plan on doing something. I'll probably work on Skyring, Tastypie or maybe dust off some stuff from my megadoomer collection of things.

I'm Twitchin'!

kyring is is a distributed system for managing timers, or delayed execution similar to `setTimeout` in javascript. The difference being that it is handled in a reliable and fault tolerant way. setTimeout in javascript is transient. If the running application is restarted or crashes, any pending timers are lost forever. The one guarantee that skyring provides is that a timer will execute after the specified delay, and that it only executes once. Exactly once is an interesting challenge in distributed systems, and Skyring makes use of a number of mechanisms at the node level to achieve this. From a high level, this is what the behavior on individual nodes looks like.

Shared Nothing

Skyring follows the shared nothing mantra similar to Cassandra and other distributed software systems. There is no master / slave setup. There is not a central controller or broker. Every node in a cluster is the same as the next and effectively on an island all to itself and has little insight into what other nodes might be doing. It really just knows that there are other nodes in the cluster, the ip address of the node and if it is currently active.

Consistent Hashring

Skyring ring uses a consistent hashing to assign a specific subset of keys to specific nodes in the cluster. Nodes connect to each other using an internal tcp mesh called tchannel. When a node in the cluster handles an HTTP request for a timer, it hashes the ID and looks up the node responsible for that timer. If the node that handled the request is not the owner, it will immediately proxy the request to the node that is, and it is handled on the respective node just like every other HTTP request. The benefits gained from this are many. Most notably, is the ability horizontally scale by just starting more instances. More instances = better performance. There are no noisy neighbors in a skyring cluster. Additionally, as nodes are added, they evenly distribute the workload (memory, CPU, disk usage, etc).

Write Ahead Persistence

Internally, each skyring node uses a levelup database (w/ a leveldown backend) where timer data is stored. Before the timer is actually started, it writes a record to leveldb so there is a disk backed record of the timer. Updates to existing timers follow the same pattern - Write to disk first. Aside from a simple persistence layer, levelup gives skyring some limited backend abstraction as levelup allows the persistence layer to be configured. Skyring itself, ships with memdown(the default) and leveldown support. But you could configure it to use rocksdb, lmdb, or any of the levelup plugins.

Using an embedded database is a critical component of the system. To really achieve a shared nothing model, each node must have it's own database. More traditional central database server solutions, by their very nature, are shared; in the way that all clients connect to the same server where the data lives. If the data is shared, there is a risk of timers being executed more than once, or ownership slipping across node boundaries. With a dedicated database for each node, this is an impossibility. It also greatly simplifies the application logic as all of the data can be trusted, and all operations are reduced to simple key operations.

Internal Rebalancing

Distributing the actual timers throughout the cluster is a little tricky as timers in javascript are maintained entirely in memory. There is no in-code representation that could be used to re-create a timer. And, timers cannot be serialized in a way that would make them transferable. This makes the act of moving them from Node A to Node B a challenge. Each node will handle this in two distinct ways. The first is at the edge and is handled by the HTTP api and the internal tchannel. When An HTTP request comes with the node checks for or assigns a unique ID for the prospective timer. It hashes the ID and determines if it is responsible for that ID. if it is not, it will proxy the request over the internal tchannel. The receiving node injects the request proxy and injects it into the HTTP server, and it is handled like every other HTTP request. In that respect, timers over the HTTP API are distributed as they are handled.

This is very similar to how a rebalance of timers is handled when a new node is added to the cluster. Say for example, we have a two node cluster and they each have some pending timers on them.

When a new node is added, we need to re-evaluate the existing timer ids in the cluster to determine if the new node should take them on. Similar to as we talked about previously, because all nodes are currently active members of the tchannel mesh we can do the following:

• Hash the timer ID
• Clear the timer from memory
• Pull & Delete the record from leveldb
• Wrap it in a mock HTTP request
• Proxy it to the right node via tchannel

As before, when the new node receives a message over the tchannel, it is injected into the HTTP router and the server handles it as normal. When a node is removed ( intentionally or not ) some interesting things have to happen. Most importantly, the node is no longer an active member of the hashring, cannot accept HTTP requests, and is no longer connected to the tchannel mesh. It does this so it will not be sent any more work during the shutdown process that would just be lost. It also means that it cannot send other nodes the pending timers it does have over the normal avenues.

To circumvent the problem, the each node is connected to a nats cluster of one or more nats servers that it uses to publish reblanace events with timer data. Nats will round robin each event to a different node in the cluster which will perform the normal operations as laid out above. If it is responsible for the timer, it will just funnel in back through the create operations setting the timeout to the remaining time left, or executing immediately if that timer has lapsed. In the case the the node is not responsible for that timer, it will just wrap the timer data in a mock HTTP request of proxy it over the tchannel mesh. Business as usual.

As nodes leave the cluster, timers are redistributed between the active nodes in the cluster. When the last node in the cluster shuts down ( or crashes for that matter ) it will skip over the rebalance phase leaving the data in leveldb intact. When the node starts again, it will start the recovery phase.

Restart Recovery

When a node starts ( or restarts ) for the first time, before anything else, it starts streaming any and all records out of the leveldb instance. If there are any records in the DB, it is safe to assume that this node was, at one time responsible for these timers and it will immediately start passing them through the normal create operations. If the initial timer has lapsed, it will just execute the timer immediately.

Without the balance and proxy logic, the nuts and bolts of what a skyring node does, is actually very simple. In fact it comes down to about 10 lines of code - literally! It is simple enough that a first year developer could understand.

const timers = new Map()
function set(id, delay, fn) {
  if (!timers.has(id)) {
    timers.set(id, setTimeout( fn, delay ).unref());
  }
}
function cancel(id) {
  const ref = timers.get(id);
  clearTimeout( ref );
  return timers.delete(id);
}

Simple. Fast. Scalable

kyring is a distributed system for managing timers. When a timer lapses, a message is delivered to destination that you have defined. *How* that message is delivered is configurable. Out of the box, Skyring comes with an `HTTP` transport, and there is an official package enabling tcp delivery of messages with connection pooling. They are pretty easy to write, and you can use any of the tools you are currently used to using.

STDOUT Transport

To illustrate the process, we're going to make a simple transport handler to write the data to stdout. Basically, speaking a transport is just a node.js module that exports a named function

Module [ˈmäjo͞ol] -n., --noun

any of a number of distinct but interrelated units from which a program may be built up or into which a complex activity may be analyzed.

'use strict'

const os = require('os')

module.exports = function stdout(method, url, payload, id, storage) {
  // deliver the message
  process.stdout.write(payload);
  process.stdout(os.EOL);

  // clear the timer
  storage.remove(id);
};

Pretty simple. We just write the data to the stdout out stream attached to the process, of course, be sure to remove the timer reference from the skyring internal storage. To load your transport into a skyring server, you can pass an array of transports when instantiating as server instance. The array can contain references to the transport itself, or as string that can be passed to require

'use strict'

const Skyring = require('skyring')
const stdout = require('./transports/stdout')
const server = new Skyring({
  transports: [require('./transports/stdout')]
, seeds: ['localhost:3455']
})

server.load().listen(3000)

Done. Just be sure that every skyring node in the cluster has all of the same transports loaded so they have the capability to execute all of the timers. Other than that, we can start using our new transport by referencing it by name in the transport field of the request to create a new timer.

curl -XPOST http://localhost:3000/timer -H 'Content-Type: application/json' -d '{
  "timeout":3000
, "data":"hello world!"
, "callback": {
    "transport": "stdout"
  , "method":"unused"
  , "uri": "unused"
  }
}'

ZMQ Transport

We've got our feet wet with this simple example. Let's expand a little bit. If you've read through the summer of sockets, you may have noticed that I am fond of ZeroMQ. So instead of just writing to stdout, let's write a transport for zmq. I would want to be able to have it send timer notifications to 1 or more servers, and specify a message distribution patterns (pub, sub, push, pull, etc).

ZeroMQ [ˈzirō emkyo͞o] -n., --noun

A messaging library, which allows you to design a complex communication system without much effort.

For starters, lets start a new npm package.

$ mkdir -p skyring-zmq-transport/lib
$ cd skyring-zmq-transport
$ touch index.js lib/zmq.js README.md
$ npm init
$ npm install zmq debug --save
$ npm install skyring --save-dev

You should have something this

|-- lib
|   `-- zmq.js
|-- package.json
|-- README.md
`-- index.js

Lets start easy and fill in the primary transport handler

'use strict'

const ZMQ = require('zmq')
const debug = require('debug')('skyring:transports:zmq')

module.exports = function zmq(method, url, payload, id, storage) {
  debug('execute zmq timer', 'timeout', payload)

  const conn = ZMQ.socket('push')
  conn.connect(url)
  conn.send('timeout', ZMQ.ZMQ_SNDMORE)
  conn.send(payload)
  storage.remove(id);
}

A simplistic start, but it does complete the basic task. When a timer lapses and was configured to use the zmq transport ( the name of our function ), we'll make a new zmq socket, connect to the requested url, and send the payload. The first problem with this is it will make a new connection for every timer execution. We really need to keep track of previously created connections. We can do that pretty easily with a Map.

'use strict'

const ZMQ = require('zmq')
const debug = require('debug')('skyring:transports:zmq')
const connections = new Map()

module.exports = function zmq(method, url, payload, id, storage) {
  debug('execute zmq timer', 'timeout', payload);

  const conn = getConnection(url);
  conn.send('timeout', ZMQ.ZMQ_SNDMORE);
  conn.send(payload);
  storage.remove(id);
}

function getConnection(addr) {
  // if we've have an existing connection, return it
  if (connections.has(addr)) return connections.get(addr);

  // otherwise make a new one
  debug(`creating socket to ${addr}`);
  const socket = ZMQ.socket('push');
  socket.on('error', (err) => {
    console.error('destroying socket %s', addr, err.message);
    socket.removeAllListeners();
    socket.disconnect();
    socket.close();
    connections.delete(addr);
  });
  connections.set(addr, socket);
  socket.connect(addr);
  return socket;
}

Now, we've added a little helper function getConnection. It takes a connection url, makes a new ZMQ push socket connected to the url, and returns it. Now we can reuse our connections, and all timers configured with the same url will use the same connection. This does the job, but we could expand a bit to allow for different socket types. ZMQ has comes with PUB / SUB, REQ / REP, and a few others. But in the context of skyring, it doesn't make sense to use a SUB socket as it's read-only. And there is no real need to wait for a response, so both req and rep are out. We should restrict our types to just push and pub. Additionally, we could make it configurable if the socket should connect or bind to the address.

'use strict'

const ZMQ = require('zmq');
const debug = require('debug')('skyring:transports:zmq');
const METHODS = new Set(['push', 'pub']);
const ZMQ_BIND = !!process.env.ZMQ_BIND;
const connections = new Map();

module.exports = function zmq(method, url, payload, id, storage) {
  debug('execute zmq timer', 'timeout', payload);

  const conn = getConnection(url, method);
  conn.send('timeout', ZMQ.ZMQ_SNDMORE);
  conn.send(payload);
  storage.remove(id);
}

function getConnection(addr, type) {
  const key = `${type}:${addr}`;
  if (connections.has(key)) return connections.get(key);

  if (!METHODS.has(type)) {
    const error = new Error(`unsupported transport method ${type}`);
    error.code = 'EZMQMETHOD';
    throw error;
  }

  debug(`creating ${type} socket to ${addr}`);
  const socket = ZMQ.socket(type);
  socket.on('error', (err) => {
    console.error('destroying socket %s', key, err.message);
    socket.removeAllListeners();
    socket.disconnect();
    socket.close();
    connections.delete(key);
  });

  connections.set(key, socket);

  if (ZMQ_BIND) {
    socket.bindSync(addr);
  } else {
    socket.connect(addr);
  }
  return socket;
}

Just a couple of small adjustments. Before creating the socket we check to see of the the socket type is allowed and throw if it is not. On line 18, we use a composite key for storing connections, so it is possible to have multiple socket types for the same url.

We can plug this into a single node skyring cluster to see it in action.

'use strict'

const Skyring = require('skyring');
const server = new Skyring({
  seeds: ['localhost:3455']
, transports: [require('./lib/zmq')]
});

server.load().listen(3000);

process.once('SIGINT', onSignal);
process.once('SIGTERM', onSignal);

function onSignal() {
  server.close(() => {
    console.log('skyring closed');
  });
}

That's it! We can start setting timers with using our zmq transport

curl -XPOST http://localhost:3000/timer -H 'Content-Type: application/json' -d '{
  "timeout":3000
, "data":"hello world!"
, "callback": {
    "transport": "zmq"
  , "method":"push"
  , "uri": "tcp://0.0.0.0:5555"
  }
}'

It is pretty simple and straight forward to build your own transport layer for Skyring. It can be a simple function, or for more complex use cases, you can build out entire packages using whatever tools you want. This makes the transport system in skyring very flexible, powerful, and easy to use.

he very idea of distributed timers is complex. Conceptually is full of race conditions and edge cases. Skyring for Node.js boils the problems space down to a simple to use library and API for building scalable service that need to perform time sensitive, actions. That is a mouthful - Think An email gateway, a web-hook service, auto-dialers for telephony systems. Or in the most practical sense, anytime you might need functionality like setTimeout but needs to survive restarts / crashes; Or are using a language that doesn't support non-blocking timers. Skyring fills that gap, and it is easy to use. We can get something going in less that 20 lines of code.

To start, we just install the skyring package from npm

$ npm install skyring --save

The only external dependency is a running nats instance. The defaults will do. You can certainly run nats as a cluster if you need high availability, but it isn't necessary for this

docker run -d --name=nats -p 4222:4222 nats:latest

Now we can just require skyring and start it.

// index.js
'use strict'

const Skyring = require('skyring')
const HTTP_PORT = +(process.env.PORT || 3000)
const node = new Skyring({
  node: {
    app: 'demo'
  }
}).load().listen( HTTP_PORT )

function onSignal () {
  // trigger triggers a rebalance and shuts down
  node.close((err) => {
    if (err) process.exitCode = 1
  })
}

process.on('SIGINT', onSignal)
process.on('SIGTERM', onSignal)

Pretty simple! We care about two ports here - the port for the http server which we have defaulted to 3000, and the port for the skyring cluster, which defaults to 3456. The default configuration is a 2 node cluster on ports3456 and 3455.

# Start Node 1
$ DEBUG=skyring:* PORT=3001 node index.js

# Start Node 2
$ DEBUG=skyring:* PORT=3002 node index.js --channel:port=3455

That is it! You just made one of those! You have a running skyring cluster, and we can start making timers.

$ curl -i -XPOST http://localhost:3000/timer -d '{
  "timeout": 5000,
  "data": "hello world!",
  "callback": {
    "transport": "http",
    "method": "post",
    "uri": "http://0.0.0.0:5555/"
  }
}'

If everything goes as planned, you should get back a 201 and a location header to the timer instance if you need to cancel it, or update it.

HTTP/1.1 201 Created
location: /timer/040a8101-9543-40a8-9634-e66de573abbc
Date: Tue, 02 May 2017 03:04:21 GMT
Connection: keep-alive
Content-Length: 0

For this to all come together, you'll want something to handle the timer when it is triggers. We can do that, again, with a simple node echo server that listens on port 5555

// echo.js
'use strict';

const http = require('http')

http.createServer( (req, res) => {
  res.writeHead(200)
  res.end();
  req.pipe(process.stdout)
}).listen(5555);

You can play around with hammering the skyring cluster with requests, stopping/starting one of the nodes, and any pending timers will find their way to a running node. The unit tests do just that! The skyring api is very fast, which means it can be used efficiently from synchronous languages like python or ruby with out the complexity of threading, or introducing bloated async frameworks. Skyring is http <> http which most all languages support out of the box. And with out too much additional work, the default skyring server could be a pretty good web hooks microservice

Skyring: Distributed, reliable timers made easy.
The code for this example can be found here

agination is a recurring problem that developers have to deal with when implementing data access layers for APIs. It can be particularly tricky with the more traditional RDMS like MySQL or Postgresql. For example, let's say we had an API endpoint that allowed consumers to search a data base of moves. We could search by title, director, starring actors, etc. Our data base has millions of movies, and we know we don't want to return all all the potential matches for every search request.

We only want to return the top 25 or so records and indicate in the response that there are more results to query for:

{
  meta: {
    total: 12000
  , limit: 25
  , next: <URL TO NEXT PAGE>
  , previous: <URL TO PREVIOUS PAGE>
  }
, data: [ ... ]
}

This sounds fairly simple and straight forward at first glance, however, in order to make pagination possible through the API, we need to indicate how many records would have been returned by the original query prior to any paging. Which effectively means we need to run the query twice as efficiently as possible. We don't want do this at the application level - pulling potentially hundreds of thousands of records into memory only to discard most of them and risk crashing the application. There are a couple of approaches.

We could just run two queries in parallel.

SELECT
  count(*)
FROM
  movies
WHERE
  director = "Michael Bay";

SELECT
  title
, release_date
, director
FROM
  movies
WHERE
  director = "Michael Bay";
ORDER BY title
LIMIT 25
OFFSET 50

Simple, easy to reason about. But it is two separate queries and round trips to the DB creating double the load on the DB. We will also still have to construct the response manually combining the result from the count query and aggregate a data array.

Alternatively, we push the count query down into a sub query.

SELECT
  title
, release_date
, director
, (
    SELECT
      count(*)
    FROM
      movies
    WHERE
      director = "Michael Bay"
  ) as total
FROM
  movies
WHERE
  director = "Michael Bay";
ORDER BY title
LIMIT 25
OFFSET 50

This is a bit better, where we can send a single query, and every row will have a column called total indicating the total number of records in the query prior to pagination. But we will still have to check one of the records to grab the count and construct the data array. We can do better though.

Common Table Expressions

Provide a way to write auxiliary statements for use in a larger query. These statements, which are often referred to as Common Table Expressions or CTEs, can be thought of as defining temporary tables that exist just for one query.

Postgres gives us a couple of interesting features that Make this easy. We can use an inline Common Table Expression to get our count information, and we can use the JSON_AGG and ROW_TO_JSON function to compile the data array of json serialized results.

WITH 
  movie_count AS (
    SELECT COUNT(*), movie_id
    FROM movies
    GROUP BY movie_id
  )
, movies AS (
    SELECT
      title
    , release_date
    , director
    FROM
      movies
    WHERE
      director = "Michael Bay";
    ORDER BY title
    LIMIT 25
    OFFSET 50
  )

SELECT
  movie_count.count,
  COALESCE(
    (
      JSON_AGG(
        ROW_TO_JSON(movies.*)
      ) FILTER ( where movies.movie_id is not null )
    )
    , '[]'::json
  ) as data
  FROM movie_count
LEFT JOIN movies ON movies.movie_id = movies.movie_id
GROUP BY movie_count.count

This will give us a single result very close to the way we want it, and make sure that we still get back an empty array if there are no rows that match the query. This will keep our in app processing down to a minimum.

{
  count: 12000
  data: [{
    "title": "Transformaers"
  , "director": "Michael Bay"
  , "release_date": 2007
  }
  , ... ]
}

The SQL isn't terribly difficult to read or understand, and we can use the original parameters to determine if there are any additionally pages based on the count which keeps the vast majority of the work at the database layer, and in a single query

const prefix = '/api/v1/movies'
const next_val = offset + limit
const prev_val = offset - limit

const next = next_val < total
  ? `${prefix}?limit=${limit}&offset=${next_val}`
  : null

const prev = prev_val < 0
  ? null
  : `${prefix}?limit=${limit}&offset=${prev_val}`

Not bad! It isn't always apparently obvious, but if you find yourself trying to manipulate data structures in your application, it is probably a job for your database. In this case we were able to condense a tabular data structure spread across 2-3 queries in to a single query that returned a nested JSON object with and array of object. That is pretty cool

postgres++

orking with timers a distributed system is a really nasty problem that pops up more often than most people would like. Something as simple an useful as setTimeout / clearTimeout becomes brittle, unreliable and a bottle neck in today's stateless, scalable server mindset. Basically, I need to be able to set a timer somewhere in a cluster with out knowing or caring about what server. And reversely, I need to be able to cancel that timer **without** having to know where in the cluster that timer lives. But before we can start to understand possible solutions, let's dive into a use case to understand the problem and why existing solutions aren't suitable replacements.

Scenarios

Un-send an email  - A simple example might be providing an email service that allows people to cancel an email they have sent if it is within a 10 second window. We can't really un-send an email, but we can just not send it until our window has lapsed. Or more accurately, send it after 10 seconds giving the end user a way to cancel the act of sending. There are many ways we could do this, but for the purpose of this post - timers.

Telemarketer auto dial - A more complex example, we could consider a telemarketer using an automated dialer. Once the telemarketer has finished talking with Person A, They are given a small window of time to take some notes and dial Person B. If they don't start dialing with in the given window, the system dials the next person on the list of numbers. The act of manual dialing stops the auto-dial timer so Person C isn't dialed the same time Person B is dialed. Timers make a lot of sense here.

For these short to medium delays setTimeout is a really good fit. They are kept in memory, asynchronous, give us a rather high level of precision, deal with an arbitrary delay on an arbitrary action. More importantly we can cancel it if we want to. Simple, to the point and very easy to reason about. However, in distributed systems, we loose track of which servers are managing which timers in the event we want to cancel it. Which is the critical criteria of our solution. Let's take a look at a couple of existing solutions out there that we could utilize to work around this.

Existing solutions

Message brokers

RabbitMQ is one of the go to solutions for effectively queuing up work. But for situations like above, there are a couple of inherent problems. For one, messages are virtually impossible to delete. It just isn't apart of the AMQP protocol. Once a message goes into a queue, the only way to get it out is consume the message, or delete / purge to queue. Not really what we are after.

Message Broker

ˈmesij ˈbrōkər -n, --noun

1. arrange or negotiate (a settlement, deal, or plan).
2. an intermediary program module that translates a message from the formal messaging protocol of the sender to the formal messaging protocol of the receiver

Secondly, there isn't a way to delay the delivery of a message - they are consumed as fast as a consumer can pull them off the queue. This is also not really great for our use cases. Another pain point is the scalability of message brokers. While it is possible, it tends to be a bit complex.

PUB-SUB / Message Bus

This can come in various forms. I've talked about doing this with multicast. Projects I have worked on and similar projects out there take an approach with some inherent flaws.

If you are thinking that this sounds like an intentional race condition - it is

When a timer is set, the same message is broadcast to all or a limited number of nodes in the cluster, and a duplicate timer is created with a random amount of skew on the timeout. The hope here is when one of them triggers, we have enough time to broadcast another message to cancel the timer which have not triggered yet and only one of them executes. If you are thinking that this sounds like an intentional race condition - It is. So we have the ability to delay execution, but now we run the risk of executing the same timer more than once.

Another problem with using pub sub or the message bus pattern here is that as we add more nodes, the amount of traffic and work that needs to be done also goes up. If every node has to do the work of setting the timer and also look it up to cancel a timer if it is a duplicate - we don't get any performance gains by adding more nodes. It actually makes it worse.

Redis Queue

Redis has proven itself to be both very reliable and very versatile. I can be used for a wide range of technological problems. In our use case, we could use redis as a queue for timers to be set, canceled or moved in a cluster. We could have all the nodes in the cluster call BLPOP on a list key waiting for work. That would start a timer, cancel a timer or whatever it might be. BLPOP picks the least idle connection and sends it the next key in the list. This gives us exactly once delivery of messages, which is what we are after.

However, canceling becomes a problem again. We have no way of ensuring that the node that gets the message is the node that has the pending timer to be canceled, or moved to a different server, etc. You could use this in combination with pubsub or multiple list keys so certain servers are responsible for a sub set of actions. The complexity of these solutions greatly out weighs any benefit and still doesn't solve our problem. Additionally, BLPOP doesn't work like you would expect it to when you need to run redis as a cluster or behind sentinel. In these situations you run the risk of duplicated messages or missing messages all together, which puts us right back where we started. And, personally speaking, I find using your data store as a messaging layer a bit misplaced.

Skyring

These sorts of problems and frustrations are what have lead me to build Skyring - Distributed Timers as a service. Simply put - Skyring exposes a simple API into a hashring of servers to manage timers

When I started building skyring, I had only a couple simple requirements.

1. Simple and easy to reason about
2. Ability to cancel timers
3. Horizontally scalable
4. Plain old setTimeout where possible
5. Reliable - removing a node should not remove timers

Simplicity

Simplicity was a major driving factor. Complexity sucks. Complex systems ultimately buckle under there own weight. A dead simple API is all I want.

Set a timer

Setting a timer is as simple is making an HTTP request to the /timer endpoint to any of the nodes in a Skyring cluster.

curl -XPOST http://localhost:8080/timer -d '{
  "timeout":2000,
  "data": "hello world"
  "callback": {
    "transport": "http",
    "uri": "http//api.someservice.com",
    "method": "POST"
  }
}'

Complexity Sucks

For performance considerations and simplicity sake, skyring will not return any data payloads. Important information is always returned in the HTTP headers. In this situation, the URI for the timer is returned in the location header.

HTTP/1.1 201 CREATED
location: /timer/4adb026b-6ef3-44a8-af16-4d6be0343ecf
Date: Fri, 23 Dec 2016 00:19:13 GMT
Connection: keep-alive
Content-Length: 0

Cancel a timer

The other critical piece to the equation is being able to cancel a timer. And it as just as simple as sending a DELETE request to the URI return in the location header

curl -XDELETE http://localhost:8080/timer/4adb026b-6ef3-44a8-af16-4d6be0343ecf

Horizontal scale

Another problem that I wanted to resolve was being able to scale the service out horizontally. Adding servers shouldn't make things slower as it tends to do with pub sub and message bus. Skyring, is a consistent hashring connected by a persistent tcp channel. As new nodes join the ring, they are responsible for only a specific portion of the timers.

Requests can be issued to any node in the cluster. If the node the request lands on is not responsible for the key in question, it is immediately proxied to the node that is responsible over the persistent connection. That means that nothing is ever more than a single network hop away. One node or one thousand nodes - still one hop away. This is very similar to the way cassandra and other distributed data stores work. This allows skyring to keep the work load evenly spread across the cluster - CPU load, network load, event loop load, etc.

Reliable

Another problem that needs to be solved is how to deal with a node that leaves the cluster - intentionally or not. The real fly in the gravy here is we have to, both, stop accepting http requests and remove it from the internal hasring so it can stop accepting work. And we still need a way to redistribute any pending timers to the still active nodes. Under the hood, this is achieved over a nats queue group so things stay fast, but ensures exactly once delivery of messages, even in clustered setups. When a timer comes in over the nats channel it is funneled back through the normal internal machinery so the node that is now responsible for that timer gets it in, at most, 1 network hop.

Skyring makes it easy to use timers with confidence in a distributed system with a simple API providing functionality in line with setTimeout and clearTimout

eroMQ has its fair share of quirks and oddities. It manages everything in a global state, requires things be manually grouped into `Contexts`, allocates a thread per context (making it not thread-save) , transports are baked into the library, and so on. It can be a bit clunky to work with at times. As a result, one of the original developers on the ZMQ project, Martin Sustrik, started a new project that evolved into a complete re-write / re-realization of the ZMQ project, called nanomsg.

Nanomsg aims to resolve many of the underlying short comings of the zeromq library, but remain compliant with the ZMTP spec. It provides many of the messaging patterns ( which are refereed to as scalability patterns ) as found in ZMQ - like Pair, Pipeline, Pub-Sub and Req/Rep.

It also introduces two new patterns, Bus - which is basically the peer-to-peer pattern ( many to many ) and Survey - which is, much like pub-sub with replies. What is even more interesting, is that the transport mechanism is plugin based, opposed to zermq where transports are vertically integrated, or baked in. This allows for others to create transport outside of the library itself. What is even more interesting, is that HTML5 Websockets is one of the transports that ships with nanomsg making the browser a viable and active participant in the network topology. The node project for nanomsg comes with a simple example of using websockets - It basically boils down to this:

// server.js
// modified example found in the nanomsg project
'use strict';
const nanomsg = require('nanomsg');
const socket = nanomsg.socket('pair');
const http   = require('http');

// Websocket
socket.bind('ws://0.0.0.0:3001');
socket.on('data', function(msg){
    pair.send(`received message ${'' + msg}`);
})

// Serve a webpage
http.createServer(function( req, res ){
  fs.createReadStream( 'index.html' ).pipe( res )
}).listen( 3000 )

Make a new Websocket, and give it a nanomessage socket type mapping, and wait for data.

<!DOCTYPE html>
<html>
<head></head>
<body>
   <div id="response"> 
      <span> </span> 
   </div>
   <input type="text">
   <button>send</button>
   <script>
      var ws = new WebSocket('ws://127.0.0.1:440',[
         'pair.sp.nanomsg.org'
      ])

      ws.onmessage = function( e ){
          var reader = new FileReader() // handle binary messages
          reader.addEventListener('loadend', function(){
             var result = reader.result;
             document.querySelector('span').textContent = result
          });
          reader.readAsText( e.data );
      }

      var button = document.querySelector('button');
      var input  = document.querySelector('input');

      button.addEventListener('click', funtion( e ){
         ws.send(input.value);
         input.value = '';
      },false)
   </script>
</body>
</html>

Done. 2-way communication ( via the Pair socket type ) over websocket using nanomsg. The server code ( the important part anyway ) is only 5 lines. Event The node package is less than 500 lines. In contrast to something like socket.io which has 3 external dependent libraries ( a parser, adapter, and underlying engine) and also requires a client side library before you can even make a connection.

dəˈvīs -n., --noun

A piece of software that connects one or more sockets together to deliver messages; An application

Additionally, all nanomsg sockets are duplex streams. So, in the case of a proxy or forwarding device, we really don't need to do anything with the messages, could just pipe them to another nanomsg socket. Similar to the pair socket type in ZeroMQ, nanomsg will not load balance multiple clients; which means our example only works with the first open connection. But you could just as easily use pub/sub. req and reply socket types or open a second socket to achieve something more practical.

Chat

Chat is really a sync problem more than it is a real-time problem

Chat is a rather common use case for websockets these days. Unlike most, I like to think of chat as a sync problem more than it is a real-time problem. Person A writes to a log, Person B and Person C get a copy of the log, or an incremental update of the log. Pub-Sub makes these updates rather easy to do, Lets tweak our example to make use of the Pub-Sub model.

We need to change the socket type on the server.

// server.js
const socket = nanomsg.socket('pub');

We also need to change the websocket mapping in the browser code

var ws = new WebSocket('ws://127.0.0.1:440',[
   `pub.sp.nanomsg.org'
]);

Yes, change both to pub. It feels backwards, but the websocket transport mapping needs to match that of the server and the library does the rest. But pretty simple. Now every browser that connects will get every message sent in a fire and forget style. As you might be thinking, subscribers can't send messages! And you would be right.

This breaks our 2-way communication. Unfortunately, nanomsg websocket protocol does not double as an http server as you are probably used to with libraries like socket.io. However, this fits in with the way nanomsg works in general. Transports are pretty agnostic as to the use case - browser or not, it is just another participant in the messaging pattern and network topology.

Topology (təˈpäləjē) -n., --noun.

1. the way in which constituent parts are interrelated or arranged.

We could open up another socket connection to handle one way push back to the server, or start an http server - As far as I can tell, you are going to need two ports or multiple servers to handle each scenario. Out of curiosity, I have implemented a silly chat app using Kafka as the keeper of the log. Basically an HTTP POST is used to write to the log, and Kafka notifies any consumers ( the same app in this case ) which publish the message down the websocket. This could easily be expanded to use Kafka topics to mirror additional rooms and IMs and tying server instances to partitions and it would scale out pretty easily.

Chat. Done.

Nanomsg builds on many of the great idea that came out of ZeroMQ and opends the door for more interesting ideas with far less work. As we've seen here, the amount of code required to include client side applications as a member of otherwise complex networking applications is rather easy to do.
You can find the code for this example here along with the rest of the summer of sockets posts