Init

.gitignore

@@ -0,0 +1,12 @@
+/.bundle/
+/.yardoc
+/_yardoc/
+/coverage/
+/doc/
+/pkg/
+/spec/reports/
+/tmp/
+/lua/.rocks
+/lua/*.snap
+/lua/*.xlog
+/.rocks/

README.md

@@ -0,0 +1,41 @@
+# Qube
+
+HTTP API layer over Tarantool Queue.
+
+## Installation
+
+1. Install [Tarantool](https://github.com/tarantool/tarantool) or `brew install tarantool`
+2. Install [Queue](https://github.com/tarantool/queue) or `tarantoolctl rocks install queue`
+2. `git clone https://github.com/tnt-qube/qube`
+3. Edit config:
+
+```lua
+-- config.lua
+
+Config.http = {
+  root  = '/api/v1',
+  host  = '127.0.0.1',
+  port  = '5672',
+  token = '77c04ced3f915240d0c5d8d5819f84c7',
+  log_requests = true,
+  log_errors   = true
+}
+
+Config.tarantool = {
+  access = {
+    user     = 'qube',
+    password = '77c04ced3f915240d0c5d8d5819f84c7',
+  },
+  node = {
+    pid_file     = '/var/run/tarantool',
+    memtx_memory = 1024 * 1024 * 1024 * 2,
+    memtx_dir    = '/var/backup/qube',
+    wal_dir      = '/var/backup/qube',
+    log          = '/var/log/qube/qube.log',
+    background   = false,
+    custom_proc_title = 'qube'
+  }
+}
+```
+
+4. Run server `tarantool init.lua`

bin/dev-run

@@ -0,0 +1,5 @@
+#!/usr/bin/env bash
+
+tarantool init.lua
+rm -rf *.snap *.xlog *.log
+clear

config.lua

@@ -0,0 +1,28 @@
+local Config = {}
+
+Config.http = {
+  root  = '/api/v1',
+  host  = '127.0.0.1',
+  port  = '5672',
+  token = '77c04ced3f915240d0c5d8d5819f84c7',
+  log_requests = true,
+  log_errors   = true
+}
+
+Config.tarantool = {
+  access = {
+    user     = 'qube',
+    password = '77c04ced3f915240d0c5d8d5819f84c7',
+  },
+  node = {
+    pid_file     = '/var/run/tarantool',
+    memtx_memory = 1024 * 1024 * 1024 * 2,
+    memtx_dir    = '/var/backup/qube',
+    wal_dir      = '/var/backup/qube',
+    log          = '/var/log/qube/qube.log',
+    background   = false,
+    custom_proc_title = 'qube'
+  }
+}
+
+return Config

init.lua

@@ -0,0 +1,73 @@
+local json   = require('json')
+local config = require('config')
+
+box.cfg(config.tarantool.node) -- or box.cfg{} for local dev
+
+local http_router = require('http.router')
+local http_server = require('http.server')
+local tsgi        = require('http.tsgi')
+
+local qube = require('lib.qube')
+-- local xray = require('lib.xray')
+
+box.once('access', function()
+   box.schema.user.create(config.tarantool.user, { password = config.tarantool.password })
+   box.schema.user.grant(config.tarantool.user,  'read,write,execute', 'universe')
+end)
+
+local function send_response(code, payload)
+  if not type(payload) == 'table' then
+    return { status = code, body = json.encode({ message = tostring(payload) }) }
+  else
+    return { status = code, body = json.encode(payload) }
+  end
+end
+
+local function auth_request(env)
+  local request_token = env:header('x-auth-token')
+  if not request_token == config.http.token then
+    return send_response(403, 'Failed to authorize request')
+  else
+    return tsgi.next(env)
+  end
+end
+
+local function forward_request(controller, request)
+  local success, result = pcall(qube[controller], request)
+  if not success then
+    return send_response(500, result)
+  else
+    return send_response(200, result)
+  end
+end
+
+local routes = {
+  { method = 'GET',    path = '/tubes',                    controller = 'tube_list'   },
+  { method = 'POST',   path = '/tubes',                    controller = 'create_tube' },
+  { method = 'DELETE', path = '/tubes/:tube',              controller = 'delete_tube' },
+  { method = 'POST',   path = '/tubes/:tube',              controller = 'add_task'    },
+  { method = 'GET',    path = '/tubes/:tube',              controller = 'take_task'   },
+  { method = 'PUT',    path = '/tubes/:tube/:task_id/ack', controller = 'ack_task'    },
+}
+
+local router = http_router.new()
+local auth_opts = {
+  preroute = true,  name = 'auth',
+  method   = 'GET', path = '/api/.*'
+}
+router:use(auth_request, auth_opts)
+
+local server_opts = {
+  log_requests = config.http.log_requests,
+  log_errors   = config.http.log_errors
+}
+local server = http_server.new(config.http.host, config.http.port, server_opts)
+
+for _, r in ipairs(routes) do
+  router:route({ method = r.method, path = config.http.root .. r.path }, function(request)
+    return forward_request(r.controller, request)
+  end)
+end
+
+server:set_router(router)
+server:start()

lib/qube.lua

@@ -0,0 +1,59 @@
+local json = require('json')
+
+local M = {}
+M.queue = require('queue')
+M.queue.cfg({ ttr = 300 })
+
+function M.create_tube(request)
+  local name = request:post_param('tube')
+  local type = request:post_param('type')
+  local opts = request:post_param('options')
+  if M.queue.create_tube(name, type, opts) then
+    return true
+  end
+end
+
+function M.delete_tube(request)
+  local name = request:stash('tube')
+  local tube = M.queue.tube[name]
+        tube:truncate()
+  return tube:drop()
+end
+
+function M.add_task(request)
+  local name = request:stash('tube')
+  local body = json.decode(request:read())
+  local task = body['task']
+  local opts = body['options'] or {}
+  if M.queue.tube[name]:put(task, opts) then
+    return true
+  end
+end
+
+function M.take_task(request)
+  local name = request:stash('tube')
+  local opts = json.decode(request:read())
+  local task = M.queue.tube[name]:take(opts['timeout'])
+  if not task then
+    return {}
+  else
+    return { task_id = task[1], data = task[3] }
+  end
+end
+
+function M.ack_task(request)
+  local name    = request:stash('tube')
+  local task_id = tonumber(request:stash('task_id'))
+  local status  = M.queue.tube[name]:ack(task_id)
+  return status
+end
+
+function M.tube_list(_)
+  local list = {}
+  for tube, _ in pairs(M.queue.tube) do
+    table.insert(list, tube)
+  end
+  return list
+end
+
+return M

lib/xray.lua

@@ -0,0 +1,19 @@
+local M = {}
+
+-- Taken from https://t.me/tarantoolru/35012
+M.dd = function(...)
+   local x = debug.getinfo(2)
+   local dbg = string.format('[%s:%d][%s]', x.source, x.currentline, x.name)
+   local formatter_yaml = require('yaml').new()
+   formatter_yaml.cfg {
+      encode_invalid_numbers = true;
+      encode_load_metatables = true;
+      encode_use_tostring = true;
+      encode_invalid_as_nil = true;
+   }
+   require('log').info('\n ++VAR DUMP++ %s %s \n  %s', dbg,
+      require('json').encode({ debug.traceback() }),
+   formatter_yaml.encode({ ... }))
+end
+
+return M