class BigRecord::ConnectionAdapters::CassandraAdapter

Constants

CHARSET

string charset

LOST_CONNECTION_ERROR_MESSAGES
NULL

utility constants

TYPE_BINARY
TYPE_BOOLEAN
TYPE_NULL

data types

TYPE_STRING

Public Class Methods

new(connection, logger, connection_options, config) click to toggle source
Calls superclass method
# File lib/big_record/connection_adapters/cassandra_adapter.rb, line 42
def initialize(connection, logger, connection_options, config)
  super(connection, logger)
  @connection_options, @config = connection_options, config
end

Public Instance Methods

build_serialized_value(type, value) click to toggle source

Serialize an object in a given type

# File lib/big_record/connection_adapters/cassandra_adapter.rb, line 227
def build_serialized_value(type, value)
  type.chr + value
end
configuration() click to toggle source
# File lib/big_record/connection_adapters/cassandra_adapter.rb, line 47
def configuration
  @config.clone
end
delete(table_name, row) click to toggle source
# File lib/big_record/connection_adapters/cassandra_adapter.rb, line 203
def delete(table_name, row)
  result = nil
  log "DELETE FROM #{table_name} WHERE ROW=#{row};" do
    result = @connection.remove(table_name.to_s, row, {:consistency => Cassandra::Consistency::QUORUM})
  end
  result
end
delete_all(table_name) click to toggle source
# File lib/big_record/connection_adapters/cassandra_adapter.rb, line 211
def delete_all(table_name)
  raise NotImplementedError
end
deserialize(str) click to toggle source

Deserialize the given string. This method supports both the pure YAML format and the type header format.

# File lib/big_record/connection_adapters/cassandra_adapter.rb, line 233
def deserialize(str)
  return unless str

  #       stay compatible with the old serialization code
  #       YAML documents start with "--- " so if we find that sequence at the beginning we
  #       consider it as a serialized YAML value, else it's the new format with the type header
  if str[0..3] == "--- "
    YAML::load(str) if str
  else
    deserialize_with_header(str)
  end
end
deserialize_with_header(data) click to toggle source

Deserialize the given string assumed to be in the type header format.

# File lib/big_record/connection_adapters/cassandra_adapter.rb, line 247
def deserialize_with_header(data)
  return unless data and data.size >= 2

  # the type of the data is encoded in the first byte
  type = data[0];

  case type
  when TYPE_NULL then nil
  when TYPE_STRING then data[1..-1]
  when TYPE_BINARY then data[1..-1]
  else data
  end
end
disconnect!() click to toggle source

CONNECTION MANAGEMENT ====================================

# File lib/big_record/connection_adapters/cassandra_adapter.rb, line 61
def disconnect!
  @connection.disconnect!
  super
end
get(table_name, row, column, options={}) click to toggle source
# File lib/big_record/connection_adapters/cassandra_adapter.rb, line 92
def get(table_name, row, column, options={})
  serialized_result = get_raw(table_name, row, column, options)
  result = nil
  if serialized_result.is_a?(Array)
    result = serialized_result.collect{|e| deserialize(e)}
  else
    result = deserialize(serialized_result)
  end
  result
end
get_columns(table_name, row, columns, options={}) click to toggle source
# File lib/big_record/connection_adapters/cassandra_adapter.rb, line 138
def get_columns(table_name, row, columns, options={})
  row_cols = get_columns_raw(table_name, row, columns, options)
  return nil unless row_cols && !row_cols.empty?

  result = {}
  row_cols.each do |key,value|
    begin
      result[key] =
        if key == 'id'
          value
        else
          deserialize(value)
        end
    rescue Exception => e
      puts "Could not load column value #{key} for row=#{row.name}"
    end
  end
  result
end
get_columns_raw(table_name, row, columns, options={}) click to toggle source
# File lib/big_record/connection_adapters/cassandra_adapter.rb, line 103
def get_columns_raw(table_name, row, columns, options={})
  result = {}

  log "SELECT (#{columns.join(", ")}) FROM #{table_name} WHERE ROW=#{row};" do
    prefix_mode = false
    prefixes = []

    columns.each do |name|
      prefix, name = name.split(":")
      prefixes << prefix+":" unless prefixes.include?(prefix+":")
      prefix_mode = name.blank?
    end

    if prefix_mode
      prefixes.sort!
      values = @connection.get(table_name, row, {:start => prefixes.first, :finish => prefixes.last + "~"})

      result["id"] = row if values && values.size > 0

      values.each do |key,value|
        result[key] = value unless value.blank?
      end
    else
      values = @connection.get_columns(table_name, row, columns)

      result["id"] = row if values && values.compact.size > 0

      columns.each_index do |id|
        result[columns[id].to_s] = values[id] unless values[id].blank?
      end
    end
  end
  result
end
get_consecutive_rows(table_name, start_row, limit, columns, stop_row = nil) click to toggle source
# File lib/big_record/connection_adapters/cassandra_adapter.rb, line 186
def get_consecutive_rows(table_name, start_row, limit, columns, stop_row = nil)
  rows = get_consecutive_rows_raw(table_name, start_row, limit, columns, stop_row)

  result = rows.collect do |row|
    cols = {}
    row.each do |key,value|
      begin
        cols[key] = (key == "id") ? value : deserialize(value)
      rescue Exception => e
        puts "Could not load column value #{key} for row=#{row.name}"
      end
    end
    cols
  end
  result
end
get_consecutive_rows_raw(table_name, start_row, limit, columns, stop_row = nil) click to toggle source
# File lib/big_record/connection_adapters/cassandra_adapter.rb, line 158
def get_consecutive_rows_raw(table_name, start_row, limit, columns, stop_row = nil)
  result = []
  log "SCAN (#{columns.join(", ")}) FROM #{table_name} WHERE START_ROW=#{start_row} AND STOP_ROW=#{stop_row} LIMIT=#{limit};" do
    options = {}
    options[:start] = start_row unless start_row.blank?
    options[:finish] = stop_row unless stop_row.blank?
    options[:count] = limit unless limit.blank?

    keys = @connection.get_range(table_name, options)

    # This will be refactored. Don't make fun of me yet.
    if !keys.empty?
      keys.each do |key|
        row = {}
        row["id"] = key.key

        key.columns.each do |col|
          column = col.column
          row[column.name] = column.value
        end

        result << row if row.keys.size > 1
      end
    end
  end
  result
end
get_raw(table_name, row, column, options={}) click to toggle source
# File lib/big_record/connection_adapters/cassandra_adapter.rb, line 84
def get_raw(table_name, row, column, options={})
  result = nil
  log "SELECT (#{column}) FROM #{table_name} WHERE ROW=#{row};" do
    result = @connection.get(table_name, row, column)
  end
  result
end
serialize(value) click to toggle source

Serialize the given value

# File lib/big_record/connection_adapters/cassandra_adapter.rb, line 218
def serialize(value)
  case value
  when NilClass then NULL
  when String then build_serialized_value(TYPE_STRING, value)
  else value.to_yaml
  end
end
update(table_name, row, values, timestamp) click to toggle source
# File lib/big_record/connection_adapters/cassandra_adapter.rb, line 76
def update(table_name, row, values, timestamp)
  serialized_collection = {}
  values.each do |column, value|
    serialized_collection[column] = serialize(value)
  end
  update_raw(table_name, row, serialized_collection, timestamp)
end
update_raw(table_name, row, values, timestamp) click to toggle source

DATABASE STATEMENTS ======================================

# File lib/big_record/connection_adapters/cassandra_adapter.rb, line 68
def update_raw(table_name, row, values, timestamp)
  result = nil
  log "UPDATE #{table_name} SET #{values.inspect if values} WHERE ROW=#{row};" do
    result = @connection.insert(table_name, row, values, {:consistency => Cassandra::Consistency::QUORUM})
  end
  result
end

Protected Instance Methods

format_log_entry(message, dump = nil) click to toggle source
# File lib/big_record/connection_adapters/cassandra_adapter.rb, line 299
def format_log_entry(message, dump = nil)
  if BigRecord::Base.colorize_logging
    if @@row_even
      @@row_even = false
      message_color, dump_color = "4;36;1", "0;1"
    else
      @@row_even = true
      message_color, dump_color = "4;35;1", "0"
    end

    log_entry = "  \e[#{message_color}m#{message}\e[0m   "
    log_entry << "\e[#{dump_color}m%#{String === dump ? 's' : 'p'}\e[0m" % dump if dump
    log_entry
  else
    "%s  %s" % [message, dump]
  end
end
log(str, name = nil) { || ... } click to toggle source
# File lib/big_record/connection_adapters/cassandra_adapter.rb, line 263
def log(str, name = nil)
  if block_given?
    if @logger and @logger.level <= Logger::INFO
      result = nil
      seconds = Benchmark.realtime { result = yield }
      @runtime += seconds
      log_info(str, name, seconds)
      result
    else
      yield
    end
  else
    log_info(str, name, 0)
    nil
  end
rescue Exception => e
  # Log message and raise exception.
  # Set last_verfication to 0, so that connection gets verified
  # upon reentering the request loop
  @last_verification = 0
  message = "#{e.class.name}: #{e.message}: #{str}"
  log_info(message, name, 0)
  raise e
end
log_info(str, name, runtime) click to toggle source
# File lib/big_record/connection_adapters/cassandra_adapter.rb, line 288
def log_info(str, name, runtime)
  return unless @logger

  @logger.debug(
    format_log_entry(
      "#{name.nil? ? "CASSANDRA" : name} (#{sprintf("%f", runtime)})",
      str.gsub(/ +/, " ")
    )
  )
end