@analogculture/weave

Weave

Weave is a system of event driven generic resources and just one possible solution to a list of problems including fault tolerance, load distribution, predictability and scalability for servers that handle data and connect to a database.

Using weave you can create a horizontally and vertically scalable cluster of workers, which can be used to perform generic actions (find, create, update, delete) on an arbitrary number of resources.

Why use it?

Weave's purpose is simple, to handle database interactions so your servers don't have to.

Use Weave if:

• you want an arbitrarily large number of resources, defined only in your business logic.
• you want to ensure that queries are always delivered and only delivered once.
• you want your servers to handle requests and not database interactions.
• you want a simple way to scale your resource pool.

How does it work?

A weave service is more like a glorified adapter than a server, it uses AMQP (RabbitMQ) to handle messaging and MongoDB for persistence, you simply pass an action to the worker pool and the first available worker will do the job.

If you have three separate queries to make, each one will be handled by an async worker and returned. You can await the result before performing the next action or perform them in parallel. A worker can only be given one job at a time and once the job is complete, its current database connection/s will be closed.

The generic nature of weave means that data sent to a worker is assumed to be "qualified", therefor all checks for validation, existence and errors must be made before a call to a weave worker.

Weave uses RabbitMQ's message acknowledgements and durable messages to provide single delivery of messages to any worker in the pool.

What's in the box?

Weave provides two functions - worker and channel.

Worker

The worker function creates a new weave service based on some simple environment variables, this service doesn't require any business logic directly, it simply creates its workers, connects to RabbitMQ and waits for a message.

Install/Config

• npm init --yes
• npm install --save @analogculture/weave
• Create a new .env file
• Add following config:

CPUS=1
APP_NAME=weave_example
QUEUE_NAME=resource_queue
MESSAGE_HOST=amqp://localhost:5672
MONGODB_HOST=mongodb://localhost:27017

Note: you can skip this and use a config object instead.

Usage

CLI

• Follow the install/config instructions above
• Create a start script in package.json:

{
  ...
  "scripts": {
    "start": "weave"
  }
  ...
}

Programmatic

• Follow the install/config instructions above
• Create an index.js file:

const { worker } = require('@analogculture/weave')
worker()

// or with a config object:

worker({
  cpus: 1,
  appName: 'weave_example',
  mongoHost: 'mongodb://localhost:27017',
  host: 'amqp://localhost:5672',
  q: 'resource_queue'
})

Channel

The channel function provides a simple API for communicating messages to your Weave pool from any other server or service configured. Configure the message host and queue with environment variables. The API is documented below.

Install/Config

• npm init --yes
• npm install --save @analogculture/weave
• Create a new .env file
• Add following config:

QUEUE_NAME=resource_queue
MESSAGE_HOST=amqp://localhost:5672

Usage

const { channel } = require('@analogculture/weave')

const sendCommand = channel({
  host: 'amqp://rabbitmq:5672', // defaults to process.env.MESSAGE_HOST
  queue: 'resource_queue' // defaults to process.env.QUEUE_NAME
})

const getItems = async id => {
  const { data } = await sendCommand({
    resource: 'items',
    action: 'find',
    payload: {
      entityId: id
    }
  })
  return data[0]
}

getItems('abc')

Details

Scaling

Vertical Scaling

Unless otherwise specified, on start, a Weave service will automatically detect and scale to create one worker per available CPU core.

Horizontal Scaling

Given the same message host and same queue name, you can replicate the weave service as many times as you like in your cluster and messages will always be handled by single worker processes.

API Examples / Docs

API actions are defined with the worker, the channel merely sends a message with the corresponding data shape.

Find

const { channel } = require('@analogculture/weave')

const sendCommand = channel({
  host: 'amqp://rabbitmq:5672',
  queue: 'resource_queue'
})

const { data } = await sendCommand({
    resource: 'items',
    action: 'find',
    payload: {
      entityId: 'abc'
    }
  })
  return data[0]

You can include mongodb queries in any call to find eg payload: { _id: { $in: allocationIds } }

Create

const { channel } = require('@analogculture/weave')

const sendCommand = channel({
  host: 'amqp://rabbitmq:5672',
  queue: 'resource_queue'
})

sendCommand({
  resource: 'items',
  action: 'create',
  payload: {
    data: {
      name: 'one'
    },
    meta: {
      entityId: 'abc'
    }
  }
})

Update

const { channel } = require('@analogculture/weave')

const sendCommand = channel({
  host: 'amqp://rabbitmq:5672',
  queue: 'resource_queue'
})

sendCommand({
  resource: 'items',
  action: 'update',
  payload: {
    data: {
      name: 'two'
    },
    meta: {
      entityId: 'abc'
    }
  }
})