Home

Awesome

This repo is not being actively maintained. I highly recommend that you try lua-cassandra

lua-resty-cassandra

Build Status Coverage Status Module Version Join the chat at https://gitter.im/jbochi/lua-resty-cassandra

Pure Lua Cassandra client using CQL binary protocol v2.

It is 100% non-blocking if used in Nginx/Openresty but can also be used with luasocket.

Installation

Luarocks

Installation through luarocks is recommended:

$ luarocks install cassandra

Manual

Copy the src/ folder and require cassandra.lua.

Usage

Overview:

local cassandra = require "cassandra"

local session = cassandra.new()
session:set_timeout(1000) -- 1000ms timeout

local connected, err = session:connect("127.0.0.1", 9042)

session:set_keyspace("lua_tests")

-- simple query
local table_created, err = session:execute [[
  CREATE TABLE users(
    user_id uuid PRIMARY KEY,
    name varchar,
    age int
  )
]]

-- query with arguments
local ok, err = session:execute([[
  INSERT INTO users(name, age, user_id) VALUES(?, ?, ?)
]], {"John O'Reilly", 42, cassandra.uuid("1144bada-852c-11e3-89fb-e0b9a54a6d11")})


-- select statement
local users, err = session:execute("SELECT name, age, user_id from users")

assert(1 == #users)
local user = users[1]
ngx.say(user.name) -- "John O'Reilly"
ngx.say(user.user_id) -- "1144bada-852c-11e3-89fb-e0b9a54a6d11"
ngx.say(user.age) -- 42

You can check more examples in the tests or here.

Socket methods

session, err = cassandra.new()

Creates a new session. Create a socket with the cosocket API if available, fallback on luasocket otherwise.

Return values:

session:set_timeout(timeout)

Sets timeout (in miliseconds). Uses Nginx tcpsock:settimeout.

Parameters:

ok, err = session:connect(contact_points, port)

Connects to a single or multiple hosts at the given port.

Parameters:

Return values:

ok, err = session:set_keepalive(max_idle_timeout, pool_size) -- Nginx only

Puts the current Cassandra connection immediately into the ngx_lua cosocket connection pool.

Note: Only call this method in the place you would have called the close method instead. Calling this method will immediately turn the current cassandra session object into the closed state. Any subsequent operations other than connect() on the current objet will return the closed error.

Parameters:

Return values:

times, err = session:get_reused_times() -- Nginx only

This method returns the (successfully) reused times for the current connection. In case of error, it returns nil and a string describing the error.

Note: If the current connection does not come from the built-in connection pool, then this method always returns 0, that is, the connection has never been reused (yet). If the connection comes from the connection pool, then the return value is always non-zero. So this method can also be used to determine if the current connection comes from the pool.

Return values:

ok, err = session:close()

Closes the current connection and returns the status.

Return values:

Client methods

All errors returned by functions in this section are tables with the following properties:

Error tables implement the __tostring method and are thus printable. A stringified error table will outputs its message property.

ok, err = session:set_keyspace(keyspace_name)

Sets session keyspace to the given keyspace_name.

Parameters:

Return values:

See :execute()

stmt, err = session:prepare(query, options)

Prepare a statement for later execution.

Parameters:

Return values:

result, err = session:execute(query, args, options)

Execute a query or previously prepared statement.

Parameters:

Return values:

batch, err = cassandra.BatchStatement(type)

Initialized a batch statement. See the example below on how to use batch statements and this for informations about the type of batch to use.

Parameters:

Return values:

batch:add(query, args)

Add an operation to a batch statement. See the example below on how to use batch statements.

Parameters:

trace, err = session:get_trace(result)

Return the trace of a given result, if possible.

Parameters:

Return values:

trace: is a table with the following keys (from system_traces.sessions and system_traces.events system tracing tables:

err: Encountered error if any.

Examples

Batches:

-- Create a batch statement
local batch = cassandra.BatchStatement()

-- Add a query
batch:add("INSERT INTO users (name, age, user_id) VALUES (?, ?, ?)",
          {"James", 32, cassandra.uuid("2644bada-852c-11e3-89fb-e0b9a54a6d93")})

-- Add a prepared statement
local stmt, err = session:prepare("INSERT INTO users (name, age, user_id) VALUES (?, ?, ?)")
batch:add(stmt, {"John", 45, cassandra.uuid("1144bada-852c-11e3-89fb-e0b9a54a6d11")})

-- Execute the batch
local result, err = session:execute(batch)

Pagination might be very useful to build web services:

-- Assuming our users table contains 1000 rows

local query = "SELECT * FROM users"
local rows, err = session:execute(query, nil, {page_size = 500}) -- default page_size is 5000

assert.same(500, #rows) -- rows contains the 500 first rows

if rows.meta.has_more_pages then
  local next_rows, err = session:execute(query, nil, {paging_state = rows.meta.paging_state})

  assert.same(500, #next_rows) -- next_rows contains the next (and last) 500 rows
end

Automated pagination:

-- Assuming our users table now contains 10.000 rows

local query = "SELECT * FROM users"

for _, rows, page, err in session:execute(query, nil, {auto_paging=true}) do
  assert.same(5000, #rows) -- rows contains 5000 rows on each iteration in this case
  -- page: will be 1 on the first iteration, 2 on the second
  -- err: in case any fetch returns an error
  -- _: (the first for argument) is the current paging_state used to fetch the rows
end

Running unit tests

We use busted and require luasocket to mock ngx.socket.tcp(). To run the tests, start a local cassandra instance and run:

$ luarocks install busted
$ make test

Running coverage

$ luarocks install luacov
$ make coverage

Report will be in ./luacov.report.out.

Running linting

$ luarocks install luacheck
$ make lint

Contributors

Juarez Bochi (@jbochi)

Thibault Charbonnier (@thibaultCha) -> Several contributions, including paging support, improved batch statements, better documentation, specs and code style.

Leandro Moreira (@leandromoreira) -> Added support for doubles

Marco Palladino (@thefosk)