Class: Aerospike::Command

Inherits:

Object

• Object
• Aerospike::Command

Defined in:

lib/aerospike/command/command.rb

Overview

:nodoc:

Direct Known Subclasses

BatchCommand, SingleCommand

Instance Method Summary

• #execute ⇒ Object
• #initialize(node) ⇒ Command constructor

  A new instance of Command.

• #set_batch_exists(policy, batch_namespace) ⇒ Object
• #set_batch_get(policy, batch_namespace, bin_names, read_attr) ⇒ Object
• #set_delete(policy, key) ⇒ Object

  Writes the command for delete operations.

• #set_exists(policy, key) ⇒ Object

  Writes the command for exist operations.

• #set_operate(policy, key, operations) ⇒ Object

  Implements different command operations.

• #set_read(policy, key, bin_names) ⇒ Object

  Writes the command for get operations (specified bins).

• #set_read_for_key_only(policy, key) ⇒ Object

  Writes the command for get operations (all bins).

• #set_read_header(policy, key) ⇒ Object

  Writes the command for getting metadata operations.

• #set_scan(policy, namespace, set_name, bin_names) ⇒ Object
• #set_touch(policy, key) ⇒ Object

  Writes the command for touch operations.

• #set_udf(policy, key, package_name, function_name, args) ⇒ Object
• #set_write(policy, operation, key, bins) ⇒ Object

  Writes the command for write operations.

• #write_bins ⇒ Object

  List of all bins that this command will write to - sub-classes should overrite this as appropriate.

Constructor Details

#initialize(node) ⇒ Command

# File 'lib/aerospike/command/command.rb', line 78

def initialize(node)
  @node = node

  self
end

Instance Method Details

#execute ⇒ Object

Raises:

• (Aerospike::Exceptions::Timeout)

# File 'lib/aerospike/command/command.rb', line 390

def execute
  iterations = 0

  # set timeout outside the loop
  limit = Time.now + @policy.timeout

  # Execute command until successful, timed out or maximum iterations have been reached.
  while true
    # too many retries
    iterations += 1
    break if (@policy.max_retries > 0) && (iterations > @policy.max_retries+1)

    # Sleep before trying again, after the first iteration
    sleep(@policy.sleep_between_retries) if iterations > 1 && @policy.sleep_between_retries > 0

    # check for command timeout
    break if @policy.timeout > 0 && Time.now > limit

    begin
      @conn = @node.get_connection(@policy.timeout)
    rescue => e
      # Socket connection error has occurred. Decrease health and retry.
      @node.decrease_health

      Aerospike.logger.error("Node #{@node.to_s}: #{e}")
      next
    end

    # Draw a buffer from buffer pool, and make sure it will be put back
    begin
      @data_buffer = Buffer.get

      # Set command buffer.
      begin
        write_buffer
      rescue => e
        Aerospike.logger.error(e)

        # All runtime exceptions are considered fatal. Do not retry.
        # Close socket to flush out possible garbage. Do not put back in pool.
        @conn.close if @conn
        raise e
      end

      # Reset timeout in send buffer (destined for server) and socket.
      @data_buffer.write_int32((@policy.timeout * 1000).to_i, 22)

      # Send command.
      begin
        @conn.write(@data_buffer, @data_offset)
      rescue => e
        # IO errors are considered temporary anomalies. Retry.
        # Close socket to flush out possible garbage. Do not put back in pool.
        @conn.close if @conn

        Aerospike.logger.error("Node #{@node.to_s}: #{e}")
        # IO error means connection to server @node is unhealthy.
        # Reflect cmd status.
        @node.decrease_health
        next
      end

      # Parse results.
      begin
        parse_result
      rescue => e
        case e
        # do not log the following exceptions
        when Aerospike::Exceptions::ScanTerminated
        when Aerospike::Exceptions::QueryTerminated
        else
          Aerospike.logger.error(e)
        end

        # close the connection
        # cancelling/closing the batch/multi commands will return an error, which will
        # close the connection to throw away its data and signal the server about the
        # situation. We will not put back the connection in the buffer.
        @conn.close if @conn
        raise e
      end

      # Reflect healthy status.
      @node.restore_health

      # Put connection back in pool.
      @node.put_connection(@conn)

      # command has completed successfully.  Exit method.
      return
    ensure
      Buffer.put(@data_buffer)
    end

  end # while

  # execution timeout
  raise Aerospike::Exceptions::Timeout.new(limit, iterations)
end

#set_batch_exists(policy, batch_namespace) ⇒ Object

# File 'lib/aerospike/command/command.rb', line 260

def set_batch_exists(policy, batch_namespace)
  # Estimate buffer size
  begin_cmd
  keys = batch_namespace.keys
  byte_size = keys.length * DIGEST_SIZE

  @data_offset += (batch_namespace ? batch_namespace.namespace.bytesize : 0)  +
    FIELD_HEADER_SIZE + byte_size + FIELD_HEADER_SIZE

  size_buffer

  write_header(policy, INFO1_READ|INFO1_NOBINDATA, 0, 2, 0)
  write_field_string(batch_namespace.namespace, Aerospike::FieldType::NAMESPACE)
  write_field_header(byte_size, Aerospike::FieldType::DIGEST_RIPE_ARRAY)

  keys.each do |key|
    @data_buffer.write_binary(key.digest, @data_offset)
    @data_offset += key.digest.bytesize
  end
  end_cmd
end

#set_batch_get(policy, batch_namespace, bin_names, read_attr) ⇒ Object

# File 'lib/aerospike/command/command.rb', line 282

def set_batch_get(policy, batch_namespace, bin_names, read_attr)
  # Estimate buffer size
  begin_cmd
  byte_size = batch_namespace.keys.length * DIGEST_SIZE

  @data_offset += batch_namespace.namespace.bytesize +
    FIELD_HEADER_SIZE + byte_size + FIELD_HEADER_SIZE

  if bin_names
    bin_names.each do |bin_name|
      estimate_operation_size_for_bin_name(bin_name)
    end
  end

  size_buffer

  operation_count = 0
  if bin_names
    operation_count = bin_names.length
  end

  write_header(policy, read_attr, 0, 2, operation_count)
  write_field_string(batch_namespace.namespace, Aerospike::FieldType::NAMESPACE)
  write_field_header(byte_size, Aerospike::FieldType::DIGEST_RIPE_ARRAY)

  batch_namespace.keys.each do |key|
    @data_buffer.write_binary(key.digest, @data_offset)
    @data_offset += key.digest.bytesize
  end

  if bin_names
    bin_names.each do |bin_name|
      write_operation_for_bin_name(bin_name, Aerospike::Operation::READ)
    end
  end

  end_cmd
end

#set_delete(policy, key) ⇒ Object

Writes the command for delete operations

# File 'lib/aerospike/command/command.rb', line 112

def set_delete(policy, key)
  begin_cmd
  field_count = estimate_key_size(key)
  size_buffer
  write_header_with_policy(policy, 0, INFO2_WRITE|INFO2_DELETE, field_count, 0)
  write_key(key)
  end_cmd
end

#set_exists(policy, key) ⇒ Object

Writes the command for exist operations

# File 'lib/aerospike/command/command.rb', line 134

def set_exists(policy, key)
  begin_cmd
  field_count = estimate_key_size(key)
  size_buffer
  write_header(policy, INFO1_READ|INFO1_NOBINDATA, 0, field_count, 0)
  write_key(key)
  end_cmd
end

#set_operate(policy, key, operations) ⇒ Object

Implements different command operations

# File 'lib/aerospike/command/command.rb', line 196

def set_operate(policy, key, operations)
  begin_cmd
  field_count = estimate_key_size(key, policy)
  read_attr = 0
  write_attr = 0
  read_header = false

  operations.each do |operation|
    case operation.op_type
    when Aerospike::Operation::READ
        read_attr |= INFO1_READ

      # Read all bins if no bin is specified.
      read_attr |= INFO1_GET_ALL unless operation.bin_name

    when Aerospike::Operation::READ_HEADER
        # The server does not currently return record header data with _INFO1_NOBINDATA attribute set.
        # The workaround is to request a non-existent bin.
        # TODO: Fix this on server.
        # read_attr |= _INFO1_READ | _INFO1_NOBINDATA
        read_attr |= INFO1_READ
      read_header = true

    else
      write_attr = INFO2_WRITE
    end

    estimate_operation_size_for_operation(operation)
  end
  size_buffer

  if write_attr != 0
    write_header_with_policy(policy, read_attr, write_attr, field_count, operations.length)
  else
    write_header(policy, read_attr, write_attr, field_count, operations.length)
  end
  write_key(key, policy)

  operations.each do |operation|
    write_operation_for_operation(operation)
  end

  write_operation_for_bin(nil, Aerospike::Operation::READ) if read_header

  end_cmd
end

#set_read(policy, key, bin_names) ⇒ Object

Writes the command for get operations (specified bins)

# File 'lib/aerospike/command/command.rb', line 154

def set_read(policy, key, bin_names)
  if bin_names && bin_names.length > 0
    begin_cmd
    field_count = estimate_key_size(key)

    bin_names.each do |bin_name|
      estimate_operation_size_for_bin_name(bin_name)
    end

    size_buffer
    write_header(policy, INFO1_READ, 0, field_count, bin_names.length)
    write_key(key)

    bin_names.each do |bin_name|
      write_operation_for_bin_name(bin_name, Aerospike::Operation::READ)
    end

    end_cmd
  else
    set_read_for_key_only(policy, key)
  end
end

#set_read_for_key_only(policy, key) ⇒ Object

Writes the command for get operations (all bins)

# File 'lib/aerospike/command/command.rb', line 144

def set_read_for_key_only(policy, key)
  begin_cmd
  field_count = estimate_key_size(key)
  size_buffer
  write_header(policy, INFO1_READ|INFO1_GET_ALL, 0, field_count, 0)
  write_key(key)
  end_cmd
end

#set_read_header(policy, key) ⇒ Object

Writes the command for getting metadata operations

# File 'lib/aerospike/command/command.rb', line 178

def set_read_header(policy, key)
  begin_cmd
  field_count = estimate_key_size(key)
  estimate_operation_size_for_bin_name('')
  size_buffer

  # The server does not currently return record header data with _INFO1_NOBINDATA attribute set.
  # The workaround is to request a non-existent bin.
  # TODO: Fix this on server.
  #command.set_read(INFO1_READ | _INFO1_NOBINDATA);
  write_header(policy, INFO1_READ, 0, field_count, 1)

  write_key(key)
  write_operation_for_bin_name('', Aerospike::Operation::READ)
  end_cmd
end

#set_scan(policy, namespace, set_name, bin_names) ⇒ Object

# File 'lib/aerospike/command/command.rb', line 321

def set_scan(policy, namespace, set_name, bin_names)
  # Estimate buffer size
  begin_cmd
  field_count = 0

  if namespace
    @data_offset += namespace.bytesize + FIELD_HEADER_SIZE
    field_count += 1
  end

  if set_name
    @data_offset += set_name.bytesize + FIELD_HEADER_SIZE
    field_count += 1
  end

  # Estimate scan options size.
  @data_offset += 2 + FIELD_HEADER_SIZE
  field_count += 1

  if bin_names
    bin_names.each do |bin_name|
      estimate_operation_size_for_bin_name(bin_name)
    end
  end

  size_buffer
  read_attr = INFO1_READ

  if !policy.include_bin_data
    read_attr |= INFO1_NOBINDATA
  end

  operation_count = 0
  if bin_names
    operation_count = bin_names.length
  end

  write_header(policy, read_attr, 0, field_count, operation_count)

  if namespace
    write_field_string(namespace, Aerospike::FieldType::NAMESPACE)
  end

  if set_name
    write_field_string(set_name, Aerospike::FieldType::TABLE)
  end

  write_field_header(2, Aerospike::FieldType::SCAN_OPTIONS)

  priority = policy.priority & 0xFF
  priority <<= 4
  if policy.fail_on_cluster_change
    priority |= 0x08
  end

  @data_buffer.write_byte(priority, @data_offset)
  @data_offset += 1
  @data_buffer.write_byte(policy.scan_percent.to_i.ord, @data_offset)
  @data_offset += 1

  if bin_names
    bin_names.each do |bin_name|
      write_operation_for_bin_name(bin_name, Aerospike::Operation::READ)
    end
  end

  end_cmd
end

#set_touch(policy, key) ⇒ Object

Writes the command for touch operations

# File 'lib/aerospike/command/command.rb', line 122

def set_touch(policy, key)
  begin_cmd
  field_count = estimate_key_size(key)
  estimate_operation_size
  size_buffer
  write_header_with_policy(policy, 0, INFO2_WRITE, field_count, 1)
  write_key(key)
  write_operation_for_operation_type(Aerospike::Operation::TOUCH)
  end_cmd
end

#set_udf(policy, key, package_name, function_name, args) ⇒ Object

# File 'lib/aerospike/command/command.rb', line 243

def set_udf(policy, key, package_name, function_name, args)
  begin_cmd
  field_count = estimate_key_size(key, policy)
  arg_bytes = args.to_bytes

  field_count += estimate_udf_size(package_name, function_name, arg_bytes)
  size_buffer

  write_header(policy, 0, INFO2_WRITE, field_count, 0)
  write_key(key, policy)
  write_field_string(package_name, Aerospike::FieldType::UDF_PACKAGE_NAME)
  write_field_string(function_name, Aerospike::FieldType::UDF_FUNCTION)
  write_field_bytes(arg_bytes, Aerospike::FieldType::UDF_ARGLIST)

  end_cmd
end

#set_write(policy, operation, key, bins) ⇒ Object

Writes the command for write operations

# File 'lib/aerospike/command/command.rb', line 91

def set_write(policy, operation, key, bins)
  begin_cmd
  field_count = estimate_key_size(key, policy)

  bins.each do |bin|
    estimate_operation_size_for_bin(bin)
  end

  size_buffer

  write_header_with_policy(policy, 0, INFO2_WRITE, field_count, bins.length)
  write_key(key, policy)

  bins.each do |bin|
    write_operation_for_bin(bin, operation)
  end

  end_cmd
end

#write_bins ⇒ Object

List of all bins that this command will write to - sub-classes should overrite this as appropriate.

# File 'lib/aerospike/command/command.rb', line 86

def write_bins
  []
end