node_clusters.js

const Aerospike = require('aerospike')
const cluster = require('cluster')
const http = require('http')
const url = require('url')
const debug = require('util').debuglog('server')

const config = {
  sharedMemory: {
    key: 0xa5000000
  }
}
const client = Aerospike.client(config)

const noWorkers = 4 // pick this no. based on number of CPUs, size of RAM, etc.

if (cluster.isMaster) {
  // spawn new worker processes
  for (var i = 0; i < noWorkers; i++) {
    cluster.fork()
  }
} else {
  // in spawend worker process
  var id = cluster.worker.id
  debug('worker %s starting', id)
  startServer()
  process.on('SIGINT', () => {
    debug('worker %s exiting', id)
    stopServer()
    process.exit()
  })
}

function startServer () {
  client.connect((err) => {
    if (err) throw err
    debug('Connected to Aerospike cluster')
  })
  http.createServer((req, res) => {
    debug('incoming request on worker %s', cluster.worker.id)
    var key = keyFromPath(req)
    var responder = sendResponse.bind(null, res)
    switch (req.method) {
      case 'GET':
        client.get(key, responder)
        break
      case 'POST':
        var body = ''
        req.on('data', (chunk) => { body += chunk })
        req.on('end', () => {
          var record = JSON.parse(body)
          client.put(key, record, responder)
        })
        break
    }
  }).listen(8000)
}

function stopServer () {
  client.close()
}

function keyFromPath (req) {
  var path = url.parse(req.url)['pathname']
  var key = path.slice(1) // remove leading '/'
  return new Aerospike.Key('test', 'demo', key)
}

function sendResponse (res, error, body) {
  if (error) {
    switch (error.code) {
      case Aerospike.status.ERR_RECORD_NOT_FOUND:
        res.writeHead(404, error.message)
        break
      default:
        res.writeHead(500, error.message)
    }
  } else if (body) {
    res.writeHead(200, { 'Content-Type': 'application/json' })
    res.write(JSON.stringify(body))
  }
  res.end()
}