Class: Aerospike::Client
- Inherits:
-
Object
- Object
- Aerospike::Client
- Defined in:
- lib/aerospike/client.rb
Overview
Examples:
# connect to the database client = Client.new(‘192.168.0.1’)
#=> raises Aerospike::Exceptions::Timeout if a :timeout
is specified and :fail_if_not_connected
set to true
Instance Attribute Summary collapse
-
#cluster ⇒ Object
readonly
Returns the value of attribute cluster.
-
#default_admin_policy ⇒ Object
Returns the value of attribute default_admin_policy.
-
#default_batch_policy ⇒ Object
Returns the value of attribute default_batch_policy.
-
#default_info_policy ⇒ Object
Returns the value of attribute default_info_policy.
-
#default_operate_policy ⇒ Object
Returns the value of attribute default_operate_policy.
-
#default_query_policy ⇒ Object
Returns the value of attribute default_query_policy.
-
#default_read_policy ⇒ Object
Returns the value of attribute default_read_policy.
-
#default_scan_policy ⇒ Object
Returns the value of attribute default_scan_policy.
-
#default_write_policy ⇒ Object
Returns the value of attribute default_write_policy.
Instance Method Summary collapse
-
#add(key, bins, options = nil) ⇒ Object
Examples:.
-
#append(key, bins, options = nil) ⇒ Object
Examples:.
-
#batch_exists(keys, options = nil) ⇒ Object
Check if multiple record keys exist in one batch call.
-
#batch_get(keys, bin_names = nil, options = nil) ⇒ Object
Read multiple record headers and bins for specified keys in one batch call.
-
#batch_get_header(keys, options = nil) ⇒ Object
Read multiple record header data for specified keys in one batch call.
-
#change_password(user, password, options = nil) ⇒ Object
Change user’s password.
-
#close ⇒ Object
Closes all client connections to database server nodes.
-
#connect ⇒ Object
Connect to the cluster.
-
#connected? ⇒ Boolean
Determines if there are active connections to the database server cluster.
-
#create_index(namespace, set_name, index_name, bin_name, index_type, collection_type = nil, options = nil, ctx: nil) ⇒ Object
Create secondary index.
-
#create_role(role_name, privileges = [], allowlist = [], read_quota = 0, write_quota = 0, options = nil) ⇒ Object
Create a user-defined role.
-
#create_user(user, password, roles, options = nil) ⇒ Object
Create user with password and roles.
-
#delete(key, options = nil) ⇒ Object
Examples:.
-
#drop_index(namespace, set_name, index_name, options = nil) ⇒ Object
Delete secondary index.
-
#drop_role(role_name, options = nil) ⇒ Object
Remove a user-defined role.
-
#drop_user(user, options = nil) ⇒ Object
Remove user from cluster.
-
#execute_udf(key, package_name, function_name, args = [], options = nil) ⇒ Object
Execute user defined function on server and return results.
-
#execute_udf_on_query(statement, package_name, function_name, function_args = [], options = nil) ⇒ Object
execute_udf_on_query applies user defined function on records that match the statement filter.
-
#exists(key, options = nil) ⇒ Object
Determines if a record key exists.
-
#get(key, bin_names = nil, options = nil) ⇒ Object
Read record header and bins for specified key.
-
#get_header(key, options = nil) ⇒ Object
Read record generation and expiration only for specified key.
-
#grant_privileges(role_name, privileges, options = nil) ⇒ Object
Grant privileges to a user-defined role.
-
#grant_roles(user, roles, options = nil) ⇒ Object
Add roles to user’s list of roles.
-
#initialize(hosts = nil, policy: ClientPolicy.new, connect: true) ⇒ Client
constructor
A new instance of Client.
-
#list_udf(options = nil) ⇒ Object
ListUDF lists all packages containing user defined functions in the server.
-
#node_names ⇒ Object
Returns list of active server node names in the cluster.
-
#nodes ⇒ Object
Returns array of active server nodes in the cluster.
-
#operate(key, operations, options = nil) ⇒ Object
Perform multiple read/write operations on a single key in one batch call.
-
#prepend(key, bins, options = nil) ⇒ Object
Examples:.
-
#put(key, bins, options = nil) ⇒ Object
Examples:.
-
#query(statement, options = nil) ⇒ Object
Query executes a query and returns a recordset.
-
#query_execute(statement, operations = [], options = nil) ⇒ Aerospike::ExecuteTask
QueryExecute applies operations on records that match the statement filter.
-
#query_partitions(partition_filter, statement, options = nil) ⇒ Object
Executes a query for specified partitions and returns a recordset.
-
#query_role(role, options = nil) ⇒ Object
Retrieve privileges for a given role.
-
#query_roles(options = nil) ⇒ Object
Retrieve all roles and their privileges.
-
#query_user(user, options = nil) ⇒ Object
Retrieve roles for a given user.
-
#query_users(options = nil) ⇒ Object
Retrieve all users and their roles.
-
#register_udf(udf_body, server_path, language, options = nil) ⇒ Object
Register package containing user defined functions with server.
-
#register_udf_from_file(client_path, server_path, language, options = nil) ⇒ Object
Register package containing user defined functions with server.
-
#remove_udf(udf_name, options = nil) ⇒ Object
RemoveUDF removes a package containing user defined functions in the server.
- #request_info(*commands, policy: nil) ⇒ Object
-
#revoke_privileges(role_name, privileges, options = nil) ⇒ Object
Revoke privileges from a user-defined role.
-
#revoke_roles(user, roles, options = nil) ⇒ Object
Remove roles from user’s list of roles.
-
#scan_all(namespace, set_name, bin_names = nil, options = nil) ⇒ Object
Reads all records in specified namespace and set from all nodes.
-
#scan_node(node, namespace, set_name, bin_names = nil, options = nil) ⇒ Object
ScanNode reads all records in specified namespace and set, from one node only.
-
#scan_node_partitions(node, namespace, set_name, bin_names = nil, options = nil) ⇒ Object
Reads all records in specified namespace and set for one node only.
-
#scan_partitions(partition_filter, namespace, set_name, bin_names = nil, options = nil) ⇒ Object
Reads records in specified namespace and set using partition filter.
-
#set_quotas(role_name, read_quota, write_quota, options = nil) ⇒ Object
Set or update quota for a role.
- #supports_feature?(feature) ⇒ Boolean
-
#touch(key, options = nil) ⇒ Object
Examples:.
-
#truncate(namespace, set_name = nil, before_last_update = nil, options = {}) ⇒ Object
Removes records in the specified namespace/set efficiently.
Constructor Details
#initialize(hosts = nil, policy: ClientPolicy.new, connect: true) ⇒ Client
Returns a new instance of Client.
49 50 51 52 53 54 55 56 57 58 |
# File 'lib/aerospike/client.rb', line 49 def initialize(hosts = nil, policy: ClientPolicy.new, connect: true) hosts = ::Aerospike::Host::Parse.(hosts || ENV["AEROSPIKE_HOSTS"] || "localhost") policy = create_policy(policy, ClientPolicy) set_default_policies(policy.policies) @cluster = Cluster.new(policy, hosts) @cluster.add_cluster_config_change_listener(self) self.connect if connect self end |
Instance Attribute Details
#cluster ⇒ Object
Returns the value of attribute cluster.
47 48 49 |
# File 'lib/aerospike/client.rb', line 47 def cluster @cluster end |
#default_admin_policy ⇒ Object
Returns the value of attribute default_admin_policy.
39 40 41 |
# File 'lib/aerospike/client.rb', line 39 def default_admin_policy @default_admin_policy end |
#default_batch_policy ⇒ Object
Returns the value of attribute default_batch_policy.
40 41 42 |
# File 'lib/aerospike/client.rb', line 40 def default_batch_policy @default_batch_policy end |
#default_info_policy ⇒ Object
Returns the value of attribute default_info_policy.
41 42 43 |
# File 'lib/aerospike/client.rb', line 41 def default_info_policy @default_info_policy end |
#default_operate_policy ⇒ Object
Returns the value of attribute default_operate_policy.
46 47 48 |
# File 'lib/aerospike/client.rb', line 46 def default_operate_policy @default_operate_policy end |
#default_query_policy ⇒ Object
Returns the value of attribute default_query_policy.
42 43 44 |
# File 'lib/aerospike/client.rb', line 42 def default_query_policy @default_query_policy end |
#default_read_policy ⇒ Object
Returns the value of attribute default_read_policy.
43 44 45 |
# File 'lib/aerospike/client.rb', line 43 def default_read_policy @default_read_policy end |
#default_scan_policy ⇒ Object
Returns the value of attribute default_scan_policy.
44 45 46 |
# File 'lib/aerospike/client.rb', line 44 def default_scan_policy @default_scan_policy end |
#default_write_policy ⇒ Object
Returns the value of attribute default_write_policy.
45 46 47 |
# File 'lib/aerospike/client.rb', line 45 def default_write_policy @default_write_policy end |
Instance Method Details
#add(key, bins, options = nil) ⇒ Object
Examples:
client.add key, {'bin', -1}, :timeout => 0.001
180 181 182 183 184 |
# File 'lib/aerospike/client.rb', line 180 def add(key, bins, = nil) policy = create_policy(, WritePolicy, default_write_policy) command = WriteCommand.new(@cluster, policy, key, hash_to_bins(bins), Aerospike::Operation::ADD) execute_command(command) end |
#append(key, bins, options = nil) ⇒ Object
Examples:
client.append key, {'bin', 'value to append'}, :timeout => 0.001
138 139 140 141 142 |
# File 'lib/aerospike/client.rb', line 138 def append(key, bins, = nil) policy = create_policy(, WritePolicy, default_write_policy) command = WriteCommand.new(@cluster, policy, key, hash_to_bins(bins), Aerospike::Operation::APPEND) execute_command(command) end |
#batch_exists(keys, options = nil) ⇒ Object
Check if multiple record keys exist in one batch call.
The returned boolean array is in positional order with the original key array order.
The policy can be used to specify timeouts and protocol type.
357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 |
# File 'lib/aerospike/client.rb', line 357 def batch_exists(keys, = nil) policy = create_policy(, BatchPolicy, default_batch_policy) results = Array.new(keys.length) if policy.use_batch_direct key_map = BatchItem.generate_map(keys) execute_batch_direct_commands(policy, keys) do |node, batch| BatchDirectExistsCommand.new(node, batch, policy, key_map, results) end else execute_batch_index_commands(policy, keys) do |node, batch| BatchIndexExistsCommand.new(node, batch, policy, results) end end results end |
#batch_get(keys, bin_names = nil, options = nil) ⇒ Object
Read multiple record headers and bins for specified keys in one batch call.
The returned records are in positional order with the original key array order.
If a key is not found, the positional record will be nil.
The policy can be used to specify timeouts and protocol type.
318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 |
# File 'lib/aerospike/client.rb', line 318 def batch_get(keys, bin_names = nil, = nil) policy = create_policy(, BatchPolicy, default_batch_policy) results = Array.new(keys.length) info_flags = INFO1_READ case bin_names when :all, nil, [] info_flags |= INFO1_GET_ALL bin_names = nil when :none info_flags |= INFO1_NOBINDATA bin_names = nil end if policy.use_batch_direct key_map = BatchItem.generate_map(keys) execute_batch_direct_commands(policy, keys) do |node, batch| BatchDirectCommand.new(node, batch, policy, key_map, bin_names, results, info_flags) end else execute_batch_index_commands(policy, keys) do |node, batch| BatchIndexCommand.new(node, batch, policy, bin_names, results, info_flags) end end results end |
#batch_get_header(keys, options = nil) ⇒ Object
Read multiple record header data for specified keys in one batch call.
The returned records are in positional order with the original key array order.
If a key is not found, the positional record will be nil.
The policy can be used to specify timeouts and protocol type.
350 351 352 |
# File 'lib/aerospike/client.rb', line 350 def batch_get_header(keys, = nil) batch_get(keys, :none, ) end |
#change_password(user, password, options = nil) ⇒ Object
Change user’s password. Clear-text password will be hashed using bcrypt before sending to server.
804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 |
# File 'lib/aerospike/client.rb', line 804 def change_password(user, password, = nil) raise Aerospike::Exceptions::Aerospike.new(INVALID_USER) unless @cluster.user && @cluster.user != "" policy = create_policy(, AdminPolicy, default_admin_policy) hash = LoginCommand.hash_password(password) command = AdminCommand.new if user == @cluster.user # Change own password. command.change_password(@cluster, policy, user, hash) else # Change other user's password by user admin. command.set_password(@cluster, policy, user, hash) end @cluster.change_password(user, hash) end |
#close ⇒ Object
Closes all client connections to database server nodes.
70 71 72 |
# File 'lib/aerospike/client.rb', line 70 def close @cluster.close end |
#connect ⇒ Object
Connect to the cluster.
63 64 65 |
# File 'lib/aerospike/client.rb', line 63 def connect @cluster.connect end |
#connected? ⇒ Boolean
Determines if there are active connections to the database server cluster.
Returns +true+ if connections exist.
78 79 80 |
# File 'lib/aerospike/client.rb', line 78 def connected? @cluster.connected? end |
#create_index(namespace, set_name, index_name, bin_name, index_type, collection_type = nil, options = nil, ctx: nil) ⇒ Object
Create secondary index.
This asynchronous server call will return before command is complete.
The user can optionally wait for command completion by using the returned
IndexTask instance.
This method is only supported by Aerospike 3 servers.
index_type should be :string, :numeric or :geo2dsphere (requires server version 3.7 or later)
collection_type should be :list, :mapkeys or :mapvalues
ctx is an optional list of context. Supported on server v6.1+.
567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 |
# File 'lib/aerospike/client.rb', line 567 def create_index(namespace, set_name, index_name, bin_name, index_type, collection_type = nil, = nil, ctx: nil) if .nil? && collection_type.is_a?(Hash) , collection_type = collection_type, nil end policy = create_policy(, Policy, default_info_policy) str_cmd = "sindex-create:ns=#{namespace}" str_cmd << ";set=#{set_name}" unless set_name.to_s.strip.empty? str_cmd << ";indexname=#{index_name};numbins=1" str_cmd << ";context=#{CDT::Context.base64(ctx)}" unless ctx.to_a.empty? str_cmd << ";indextype=#{collection_type.to_s.upcase}" if collection_type str_cmd << ";indexdata=#{bin_name},#{index_type.to_s.upcase}" str_cmd << ";priority=normal" # Send index command to one node. That node will distribute the command to other nodes. response = send_info_command(policy, str_cmd).upcase if response == "OK" # Return task that could optionally be polled for completion. return IndexTask.new(@cluster, namespace, index_name) end if response.start_with?("FAIL:200") # Index has already been created. Do not need to poll for completion. return IndexTask.new(@cluster, namespace, index_name, true) end raise Aerospike::Exceptions::Aerospike.new(Aerospike::ResultCode::INDEX_GENERIC, "Create index failed: #{response}") end |
#create_role(role_name, privileges = [], allowlist = [], read_quota = 0, write_quota = 0, options = nil) ⇒ Object
Create a user-defined role. Quotas require server security configuration “enable-quotas” to be set to true. Pass 0 for quota values for no limit.
867 868 869 870 871 |
# File 'lib/aerospike/client.rb', line 867 def create_role(role_name, privileges = [], allowlist = [], read_quota = 0, write_quota = 0, = nil) policy = create_policy(, AdminPolicy, default_admin_policy) command = AdminCommand.new command.create_role(@cluster, policy, role_name, privileges, allowlist, read_quota, write_quota) end |
#create_user(user, password, roles, options = nil) ⇒ Object
Create user with password and roles. Clear-text password will be hashed using bcrypt before sending to server.
789 790 791 792 793 794 |
# File 'lib/aerospike/client.rb', line 789 def create_user(user, password, roles, = nil) policy = create_policy(, AdminPolicy, default_admin_policy) hash = LoginCommand.hash_password(password) command = AdminCommand.new command.create_user(@cluster, policy, user, hash, roles) end |
#delete(key, options = nil) ⇒ Object
Examples:
existed = client.delete key, :timeout => 0.001
202 203 204 205 206 207 |
# File 'lib/aerospike/client.rb', line 202 def delete(key, = nil) policy = create_policy(, WritePolicy, default_write_policy) command = DeleteCommand.new(@cluster, policy, key) execute_command(command) command.existed end |
#drop_index(namespace, set_name, index_name, options = nil) ⇒ Object
Delete secondary index.
This method is only supported by Aerospike 3 servers.
598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 |
# File 'lib/aerospike/client.rb', line 598 def drop_index(namespace, set_name, index_name, = nil) policy = create_policy(, Policy, default_info_policy) str_cmd = "sindex-delete:ns=#{namespace}" str_cmd << ";set=#{set_name}" unless set_name.to_s.strip.empty? str_cmd << ";indexname=#{index_name}" # Send index command to one node. That node will distribute the command to other nodes. response = send_info_command(policy, str_cmd).upcase return if response == "OK" # Index did not previously exist. Return without error. return if response.start_with?("FAIL:201") raise Aerospike::Exceptions::Aerospike.new(Aerospike::ResultCode::INDEX_GENERIC, "Drop index failed: #{response}") end |
#drop_role(role_name, options = nil) ⇒ Object
Remove a user-defined role.
874 875 876 877 878 |
# File 'lib/aerospike/client.rb', line 874 def drop_role(role_name, = nil) policy = create_policy(, AdminPolicy, default_admin_policy) command = AdminCommand.new command.drop_role(@cluster, policy, role_name) end |
#drop_user(user, options = nil) ⇒ Object
Remove user from cluster.
797 798 799 800 801 |
# File 'lib/aerospike/client.rb', line 797 def drop_user(user, = nil) policy = create_policy(, AdminPolicy, default_admin_policy) command = AdminCommand.new command.drop_user(@cluster, policy, user) end |
#execute_udf(key, package_name, function_name, args = [], options = nil) ⇒ Object
Execute user defined function on server and return results.
The function operates on a single record.
The package name is used to locate the udf file location:
udf file = <server udf dir>/<package name>.lua
This method is only supported by Aerospike 3 servers.
502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 |
# File 'lib/aerospike/client.rb', line 502 def execute_udf(key, package_name, function_name, args = [], = nil) policy = create_policy(, WritePolicy, default_write_policy) command = ExecuteCommand.new(@cluster, policy, key, package_name, function_name, args) execute_command(command) record = command.record return nil if !record || record.bins.empty? result_map = record.bins # User defined functions don't have to return a value. key, obj = result_map.detect { |k, _| k.include?("SUCCESS") } return obj if key key, obj = result_map.detect { |k, _| k.include?("FAILURE") } = key ? obj.to_s : "Invalid UDF return value" raise Aerospike::Exceptions::Aerospike.new(Aerospike::ResultCode::UDF_BAD_RESPONSE, ) end |
#execute_udf_on_query(statement, package_name, function_name, function_args = [], options = nil) ⇒ Object
execute_udf_on_query applies user defined function on records that match the statement filter. Records are not returned to the client. This asynchronous server call will return before command is complete. The user can optionally wait for command completion by using the returned ExecuteTask instance.
This method is only supported by Aerospike 3 servers. If the policy is nil, the default relevant policy will be used.
531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 |
# File 'lib/aerospike/client.rb', line 531 def execute_udf_on_query(statement, package_name, function_name, function_args = [], = nil) policy = create_policy(, WritePolicy, default_write_policy) nodes = @cluster.nodes if nodes.empty? raise Aerospike::Exceptions::Aerospike.new(Aerospike::ResultCode::SERVER_NOT_AVAILABLE, "Executing UDF failed because cluster is empty.") end statement = statement.clone statement.set_aggregate_function(package_name, function_name, function_args, false) # Use a thread per node nodes.each do |node| Thread.new do Thread.current.abort_on_exception = true begin command = ServerCommand.new(@cluster, node, policy, statement, true, statement.task_id) execute_command(command) rescue => e Aerospike.logger.error(e) raise e end end end ExecuteTask.new(@cluster, statement) end |
#exists(key, options = nil) ⇒ Object
Determines if a record key exists.
The policy can be used to specify timeouts.
280 281 282 283 284 285 |
# File 'lib/aerospike/client.rb', line 280 def exists(key, = nil) policy = create_policy(, Policy, default_read_policy) command = ExistsCommand.new(@cluster, policy, key) execute_command(command) command.exists end |
#get(key, bin_names = nil, options = nil) ⇒ Object
Read record header and bins for specified key.
The policy can be used to specify timeouts.
293 294 295 296 297 298 299 |
# File 'lib/aerospike/client.rb', line 293 def get(key, bin_names = nil, = nil) policy = create_policy(, Policy, default_read_policy) command = ReadCommand.new(@cluster, policy, key, bin_names) execute_command(command) command.record end |
#get_header(key, options = nil) ⇒ Object
Read record generation and expiration only for specified key. Bins are not read.
The policy can be used to specify timeouts.
303 304 305 306 307 308 |
# File 'lib/aerospike/client.rb', line 303 def get_header(key, = nil) policy = create_policy(, Policy, default_read_policy) command = ReadHeaderCommand.new(@cluster, policy, key) execute_command(command) command.record end |
#grant_privileges(role_name, privileges, options = nil) ⇒ Object
Grant privileges to a user-defined role.
881 882 883 884 885 |
# File 'lib/aerospike/client.rb', line 881 def grant_privileges(role_name, privileges, = nil) policy = create_policy(, AdminPolicy, default_admin_policy) command = AdminCommand.new command.grant_privileges(@cluster, policy, role_name, privileges) end |
#grant_roles(user, roles, options = nil) ⇒ Object
Add roles to user’s list of roles.
823 824 825 826 827 |
# File 'lib/aerospike/client.rb', line 823 def grant_roles(user, roles, = nil) policy = create_policy(, AdminPolicy, default_admin_policy) command = AdminCommand.new command.grant_roles(@cluster, policy, user, roles) end |
#list_udf(options = nil) ⇒ Object
ListUDF lists all packages containing user defined functions in the server.
This method is only supported by Aerospike 3 servers.
464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 |
# File 'lib/aerospike/client.rb', line 464 def list_udf( = nil) policy = create_policy(, Policy, default_info_policy) str_cmd = "udf-list" # Send command to one node. That node will distribute it to other nodes. response_map = @cluster.request_info(policy, str_cmd) _, response = response_map.first vals = response.split(";") vals.map do |udf_info| next if udf_info.strip! == "" udf_parts = udf_info.split(",") udf = UDF.new udf_parts.each do |values| k, v = values.split("=", 2) case k when "filename" udf.filename = v when "hash" udf.hash = v when "type" udf.language = v end end udf end end |
#node_names ⇒ Object
Returns list of active server node names in the cluster.
92 93 94 |
# File 'lib/aerospike/client.rb', line 92 def node_names @cluster.nodes.map(&:name) end |
#nodes ⇒ Object
Returns array of active server nodes in the cluster.
85 86 87 |
# File 'lib/aerospike/client.rb', line 85 def nodes @cluster.nodes end |
#operate(key, operations, options = nil) ⇒ Object
Perform multiple read/write operations on a single key in one batch call.
An example would be to add an integer value to an existing record and then
read the result, all in one database call. Operations are executed in
the order they are specified.
383 384 385 386 387 388 389 390 |
# File 'lib/aerospike/client.rb', line 383 def operate(key, operations, = nil) policy = create_policy(, OperatePolicy, default_operate_policy) args = OperateArgs.new(cluster, policy, default_write_policy, default_operate_policy, key, operations) command = OperateCommand.new(@cluster, key, args) execute_command(command) command.record end |
#prepend(key, bins, options = nil) ⇒ Object
Examples:
client.prepend key, {'bin', 'value to prepend'}, :timeout => 0.001
157 158 159 160 161 |
# File 'lib/aerospike/client.rb', line 157 def prepend(key, bins, = nil) policy = create_policy(, WritePolicy, default_write_policy) command = WriteCommand.new(@cluster, policy, key, hash_to_bins(bins), Aerospike::Operation::PREPEND) execute_command(command) end |
#put(key, bins, options = nil) ⇒ Object
Examples:
client.put key, {'bin', 'value string'}, :timeout => 0.001
115 116 117 118 119 |
# File 'lib/aerospike/client.rb', line 115 def put(key, bins, = nil) policy = create_policy(, WritePolicy, default_write_policy) command = WriteCommand.new(@cluster, policy, key, hash_to_bins(bins), Aerospike::Operation::WRITE) execute_command(command) end |
#query(statement, options = nil) ⇒ Object
Query executes a query and returns a recordset. The query executor puts records on a channel from separate threads. The caller can concurrently pops records off the channel through the record channel.
This method is only supported by Aerospike 3 servers. If the policy is nil, a default policy will be generated.
726 727 728 |
# File 'lib/aerospike/client.rb', line 726 def query(statement, = nil) query_partitions(Aerospike::PartitionFilter.all, statement, ) end |
#query_execute(statement, operations = [], options = nil) ⇒ Aerospike::ExecuteTask
QueryExecute applies operations on records that match the statement filter. Records are not returned to the client. This asynchronous server call will return before the command is complete. The user can optionally wait for command completion by using the returned ExecuteTask instance.
This method is only supported by Aerospike 3+ servers. If the policy is nil, the default relevant policy will be used.
749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 |
# File 'lib/aerospike/client.rb', line 749 def query_execute(statement, operations = [], = nil) policy = create_policy(, WritePolicy, default_write_policy) if statement.nil? raise Aerospike::Exceptions::Aerospike.new(Aerospike::ResultCode::INVALID_COMMAND, "Query failed of invalid statement.") end statement = statement.clone unless operations.empty? statement.operations = operations end task_id = statement.task_id nodes = @cluster.nodes if nodes.empty? raise Aerospike::Exceptions::Aerospike.new(Aerospike::ResultCode::SERVER_NOT_AVAILABLE, "Query failed because cluster is empty.") end # Use a thread per node nodes.each do |node| Thread.new do Thread.current.abort_on_exception = true begin command = ServerCommand.new(@cluster, node, policy, statement, true, task_id) execute_command(command) rescue => e Aerospike.logger.error(e) raise e end end end ExecuteTask.new(@cluster, statement) end |
#query_partitions(partition_filter, statement, options = nil) ⇒ Object
Executes a query for specified partitions and returns a recordset. The query executor puts records on the queue from separate threads. The caller can concurrently pop records off the queue through the recordset.records API.
This method is only supported by Aerospike 4.9+ servers. If the policy is nil, the default relevant policy will be used.
700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 |
# File 'lib/aerospike/client.rb', line 700 def query_partitions(partition_filter, statement, = nil) policy = create_policy(, QueryPolicy, default_query_policy) nodes = @cluster.nodes if nodes.empty? raise Aerospike::Exceptions::Aerospike.new(Aerospike::ResultCode::SERVER_NOT_AVAILABLE, "Query failed because cluster is empty.") end # result recordset recordset = Recordset.new(policy.record_queue_size, 1, :query) tracker = PartitionTracker.new(policy, nodes, partition_filter) Thread.new do Thread.current.abort_on_exception = true QueryExecutor.query_partitions(@cluster, policy, tracker, statement, recordset) end recordset end |
#query_role(role, options = nil) ⇒ Object
Retrieve privileges for a given role.
851 852 853 854 855 |
# File 'lib/aerospike/client.rb', line 851 def query_role(role, = nil) policy = create_policy(, AdminPolicy, default_admin_policy) command = AdminCommand.new command.query_role(@cluster, policy, role) end |
#query_roles(options = nil) ⇒ Object
Retrieve all roles and their privileges.
858 859 860 861 862 |
# File 'lib/aerospike/client.rb', line 858 def query_roles( = nil) policy = create_policy(, AdminPolicy, default_admin_policy) command = AdminCommand.new command.query_roles(@cluster, policy) end |
#query_user(user, options = nil) ⇒ Object
Retrieve roles for a given user.
837 838 839 840 841 |
# File 'lib/aerospike/client.rb', line 837 def query_user(user, = nil) policy = create_policy(, AdminPolicy, default_admin_policy) command = AdminCommand.new command.query_user(@cluster, policy, user) end |
#query_users(options = nil) ⇒ Object
Retrieve all users and their roles.
844 845 846 847 848 |
# File 'lib/aerospike/client.rb', line 844 def query_users( = nil) policy = create_policy(, AdminPolicy, default_admin_policy) command = AdminCommand.new command.query_users(@cluster, policy) end |
#register_udf(udf_body, server_path, language, options = nil) ⇒ Object
Register package containing user defined functions with server.
This asynchronous server call will return before command is complete.
The user can optionally wait for command completion by using the returned
RegisterTask instance.
This method is only supported by Aerospike 3 servers.
413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 |
# File 'lib/aerospike/client.rb', line 413 def register_udf(udf_body, server_path, language, = nil) policy = create_policy(, Policy, default_info_policy) content = Base64.strict_encode64(udf_body).force_encoding("binary") str_cmd = "udf-put:filename=#{server_path};content=#{content};" str_cmd << "content-len=#{content.length};udf-type=#{language};" # Send UDF to one node. That node will distribute the UDF to other nodes. response_map = @cluster.request_info(policy, str_cmd) res = {} response_map.each do |k, response| vals = response.to_s.split(";") vals.each do |pair| k, v = pair.split("=", 2) res[k] = v end end if res["error"] raise Aerospike::Exceptions::CommandRejected.new("Registration failed: #{res["error"]}\nFile: #{res["file"]}\nLine: #{res["line"]}\nMessage: #{res["message"]}") end UdfRegisterTask.new(@cluster, server_path) end |
#register_udf_from_file(client_path, server_path, language, options = nil) ⇒ Object
Register package containing user defined functions with server.
This asynchronous server call will return before command is complete.
The user can optionally wait for command completion by using the returned
RegisterTask instance.
This method is only supported by Aerospike 3 servers.
402 403 404 405 |
# File 'lib/aerospike/client.rb', line 402 def register_udf_from_file(client_path, server_path, language, = nil) udf_body = File.read(client_path) register_udf(udf_body, server_path, language, ) end |
#remove_udf(udf_name, options = nil) ⇒ Object
RemoveUDF removes a package containing user defined functions in the server.
This asynchronous server call will return before command is complete.
The user can optionally wait for command completion by using the returned
RemoveTask instance.
This method is only supported by Aerospike 3 servers.
445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 |
# File 'lib/aerospike/client.rb', line 445 def remove_udf(udf_name, = nil) policy = create_policy(, Policy, default_info_policy) str_cmd = "udf-remove:filename=#{udf_name};" # Send command to one node. That node will distribute it to other nodes. # Send UDF to one node. That node will distribute the UDF to other nodes. response_map = @cluster.request_info(policy, str_cmd) _, response = response_map.first if response == "ok" UdfRemoveTask.new(@cluster, udf_name) else raise Aerospike::Exceptions::Aerospike.new(Aerospike::ResultCode::SERVER_ERROR, response) end end |
#request_info(*commands, policy: nil) ⇒ Object
615 616 617 618 |
# File 'lib/aerospike/client.rb', line 615 def request_info(*commands, policy: nil) policy = create_policy(policy, Policy, default_info_policy) @cluster.request_info(policy, *commands) end |
#revoke_privileges(role_name, privileges, options = nil) ⇒ Object
Revoke privileges from a user-defined role.
888 889 890 891 892 |
# File 'lib/aerospike/client.rb', line 888 def revoke_privileges(role_name, privileges, = nil) policy = create_policy(, AdminPolicy, default_admin_policy) command = AdminCommand.new command.revoke_privileges(@cluster, policy, role_name, privileges) end |
#revoke_roles(user, roles, options = nil) ⇒ Object
Remove roles from user’s list of roles.
830 831 832 833 834 |
# File 'lib/aerospike/client.rb', line 830 def revoke_roles(user, roles, = nil) policy = create_policy(, AdminPolicy, default_admin_policy) command = AdminCommand.new command.revoke_roles(@cluster, policy, user, roles) end |
#scan_all(namespace, set_name, bin_names = nil, options = nil) ⇒ Object
Reads all records in specified namespace and set from all nodes. If the policy’s concurrent_nodes is specified, each server node will be read in parallel. Otherwise, server nodes are read sequentially. If the policy is nil, the default relevant policy will be used.
679 680 681 |
# File 'lib/aerospike/client.rb', line 679 def scan_all(namespace, set_name, bin_names = nil, = nil) scan_partitions(Aerospike::PartitionFilter.all, namespace, set_name, bin_names, ) end |
#scan_node(node, namespace, set_name, bin_names = nil, options = nil) ⇒ Object
ScanNode reads all records in specified namespace and set, from one node only. The policy can be used to specify timeouts.
685 686 687 |
# File 'lib/aerospike/client.rb', line 685 def scan_node(node, namespace, set_name, bin_names = nil, = nil) scan_node_partitions(node, namespace, set_name, bin_names, ) end |
#scan_node_partitions(node, namespace, set_name, bin_names = nil, options = nil) ⇒ Object
Reads all records in specified namespace and set for one node only. If the policy is nil, the default relevant policy will be used.
654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 |
# File 'lib/aerospike/client.rb', line 654 def scan_node_partitions(node, namespace, set_name, bin_names = nil, = nil) policy = create_policy(, ScanPolicy, default_scan_policy) # Retry policy must be one-shot for scans. # copy on write for policy new_policy = policy.clone unless node.active? raise Aerospike::Exceptions::Aerospike.new(Aerospike::ResultCode::SERVER_NOT_AVAILABLE, "Scan failed because cluster is empty.") end tracker = Aerospike::PartitionTracker.new(policy, [node]) recordset = Recordset.new(policy.record_queue_size, 1, :scan) Thread.new do Thread.current.abort_on_exception = true ScanExecutor.scan_partitions(policy, @cluster, tracker, namespace, set_name, recordset, bin_names) end recordset end |
#scan_partitions(partition_filter, namespace, set_name, bin_names = nil, options = nil) ⇒ Object
Reads records in specified namespace and set using partition filter. If the policy’s concurrent_nodes is specified, each server node will be read in parallel. Otherwise, server nodes are read sequentially. If partition_filter is nil, all partitions will be scanned. If the policy is nil, the default relevant policy will be used. This method is only supported by Aerospike 4.9+ servers.
630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 |
# File 'lib/aerospike/client.rb', line 630 def scan_partitions(partition_filter, namespace, set_name, bin_names = nil, = nil) policy = create_policy(, ScanPolicy, default_scan_policy) # Retry policy must be one-shot for scans. # copy on write for policy new_policy = policy.clone nodes = @cluster.nodes if nodes.empty? raise Aerospike::Exceptions::Aerospike.new(Aerospike::ResultCode::SERVER_NOT_AVAILABLE, "Scan failed because cluster is empty.") end tracker = Aerospike::PartitionTracker.new(policy, nodes, partition_filter) recordset = Recordset.new(policy.record_queue_size, 1, :scan) Thread.new do Thread.current.abort_on_exception = true ScanExecutor.scan_partitions(policy, @cluster, tracker, namespace, set_name, recordset, bin_names) end recordset end |
#set_quotas(role_name, read_quota, write_quota, options = nil) ⇒ Object
Set or update quota for a role.
895 896 897 898 899 |
# File 'lib/aerospike/client.rb', line 895 def set_quotas(role_name, read_quota, write_quota, = nil) policy = create_policy(, AdminPolicy, default_admin_policy) command = AdminCommand.new command.set_quotas(@cluster, policy, role_name, read_quota, write_quota) end |
#supports_feature?(feature) ⇒ Boolean
96 97 98 |
# File 'lib/aerospike/client.rb', line 96 def supports_feature?(feature) @cluster.supports_feature?(feature) end |
#touch(key, options = nil) ⇒ Object
Examples:
client.touch key, :timeout => 0.001
267 268 269 270 271 |
# File 'lib/aerospike/client.rb', line 267 def touch(key, = nil) policy = create_policy(, WritePolicy, default_write_policy) command = TouchCommand.new(@cluster, policy, key) execute_command(command) end |
#truncate(namespace, set_name = nil, before_last_update = nil, options = {}) ⇒ Object
Removes records in the specified namespace/set efficiently.
This method is orders of magnitude faster than deleting records one at a time. It requires Aerospike Server version 3.12 or later. See www.aerospike.com/docs/reference/info#truncate for further information.
This asynchronous server call may return before the truncate is complete. The user can still write new records after the server call returns because new records will have last update times greater than the truncate cut-off (set at the time of the truncate call.)
If no policy options are provided, @default_info_policy will be used.
224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 |
# File 'lib/aerospike/client.rb', line 224 def truncate(namespace, set_name = nil, before_last_update = nil, = {}) policy = create_policy(, Policy, default_info_policy) node = @cluster.random_node if set_name && !set_name.to_s.strip.empty? str_cmd = "truncate:namespace=#{namespace}" str_cmd << ";set=#{set_name}" unless set_name.to_s.strip.empty? else if node.supports_feature?(Aerospike::Features::TRUNCATE_NAMESPACE) str_cmd = "truncate-namespace:namespace=#{namespace}" else str_cmd = "truncate:namespace=#{namespace}" end end if before_last_update lut_nanos = (before_last_update.to_f * 1_000_000_000.0).round str_cmd << ";lut=#{lut_nanos}" elsif supports_feature?(Aerospike::Features::LUT_NOW) # Servers >= 4.3.1.4 require lut argument str_cmd << ";lut=now" end response = send_info_command(policy, str_cmd, node).upcase return if response == "OK" raise Aerospike::Exceptions::Aerospike.new(Aerospike::ResultCode::SERVER_ERROR, "Truncate failed: #{response}") end |