- * If threadPool is shared, threadPool will not be shutdown when the client instance is
- * closed. This shared threadPool should be shutdown manually before the program
- * terminates. Shutdown is recommended, but not absolutely required if threadPool is
- * constructed to use daemon threads.
- *
- * Default: false
- */
- private final boolean sharedThreadPool;
-
- /**
- * Underlying thread pool used in synchronous batch, scan, and query commands. These commands
- * are often sent to multiple server nodes in parallel threads. A thread pool improves
- * performance because threads do not have to be created/destroyed for each command.
- * The default, null, indicates that the following daemon thread pool will be used:
- *
- * Daemon threads automatically terminate when the program terminates.
- *
- * Default: null (use Executors.newCachedThreadPool)
- */
- private final ExecutorService threadPool;
-
- /**
- * Upper limit of proxy server connection.
- */
- private static final int MAX_CONNECTIONS = 8;
-
- private static final String NotSupported = "Method not supported in proxy client: ";
-
- //-------------------------------------------------------
- // Member variables.
- //-------------------------------------------------------
-
- /**
- * Default read policy that is used when read command policy is null.
- */
- public final Policy readPolicyDefault;
-
- /**
- * Default write policy that is used when write command policy is null.
- */
- public final WritePolicy writePolicyDefault;
-
- /**
- * Default scan policy that is used when scan command policy is null.
- */
- public final ScanPolicy scanPolicyDefault;
-
- /**
- * Default query policy that is used when query command policy is null.
- */
- public final QueryPolicy queryPolicyDefault;
-
- /**
- * Default parent policy used in batch read commands. Parent policy fields
- * include socketTimeout, totalTimeout, maxRetries, etc...
- */
- public final BatchPolicy batchPolicyDefault;
-
- /**
- * Default parent policy used in batch write commands. Parent policy fields
- * include socketTimeout, totalTimeout, maxRetries, etc...
- */
- public final BatchPolicy batchParentPolicyWriteDefault;
-
- /**
- * Default write policy used in batch operate commands.
- * Write policy fields include generation, expiration, durableDelete, etc...
- */
- public final BatchWritePolicy batchWritePolicyDefault;
-
- /**
- * Default delete policy used in batch delete commands.
- */
- public final BatchDeletePolicy batchDeletePolicyDefault;
-
- /**
- * Default user defined function policy used in batch UDF execute commands.
- */
- public final BatchUDFPolicy batchUDFPolicyDefault;
-
- /**
- * Default info policy that is used when info command policy is null.
- */
- public final InfoPolicy infoPolicyDefault;
-
- /**
- * Default multi-record transaction (MRT) policy when verifying record versions in a batch on a commit.
- */
- public final TxnVerifyPolicy txnVerifyPolicyDefault;
-
- /**
- * Default multi-record transaction (MRT) policy when rolling the transaction records forward (commit)
- * or back (abort) in a batch.
- */
- public final TxnRollPolicy txnRollPolicyDefault;
-
- private final WritePolicy operatePolicyReadDefault;
- private final AuthTokenManager authTokenManager;
- private final GrpcCallExecutor executor;
-
- //-------------------------------------------------------
- // Constructors
- //-------------------------------------------------------
-
- /**
- * Initialize proxy client with suitable hosts to seed the cluster map.
- * The client policy is used to set defaults and size internal data structures.
- *
- * In most cases, only one host is necessary to seed the cluster. The remaining hosts
- * are added as future seeds in case of a complete network failure.
- *
- * @param policy client configuration parameters, pass in null for defaults
- * @param hosts array of potential hosts to seed the cluster
- * @throws AerospikeException if all host connections fail
- */
- public AerospikeClientProxy(ClientPolicy policy, Host... hosts) {
- if (policy == null) {
- policy = new ClientPolicy();
- policy.minConnsPerNode = 1;
- policy.maxConnsPerNode = 8;
- policy.asyncMaxConnsPerNode = 8;
- policy.timeout = 5000;
- }
-
- if (policy.threadPool == null) {
- threadPool = Executors.newCachedThreadPool(new ThreadDaemonFactory());
- }
- else {
- threadPool = policy.threadPool;
- }
- sharedThreadPool = policy.sharedThreadPool;
-
- this.readPolicyDefault = policy.readPolicyDefault;
- this.writePolicyDefault = policy.writePolicyDefault;
- this.scanPolicyDefault = policy.scanPolicyDefault;
- this.queryPolicyDefault = policy.queryPolicyDefault;
- this.batchPolicyDefault = policy.batchPolicyDefault;
- this.batchParentPolicyWriteDefault = policy.batchParentPolicyWriteDefault;
- this.batchWritePolicyDefault = policy.batchWritePolicyDefault;
- this.batchDeletePolicyDefault = policy.batchDeletePolicyDefault;
- this.batchUDFPolicyDefault = policy.batchUDFPolicyDefault;
- this.infoPolicyDefault = policy.infoPolicyDefault;
- this.txnVerifyPolicyDefault = policy.txnVerifyPolicyDefault;
- this.txnRollPolicyDefault = policy.txnRollPolicyDefault;
- this.operatePolicyReadDefault = new WritePolicy(this.readPolicyDefault);
-
- GrpcChannelProvider channelProvider = new GrpcChannelProvider();
-
- if (policy.user != null || policy.password != null) {
- authTokenManager = new AuthTokenManager(policy, channelProvider);
- }
- else {
- authTokenManager = null;
- }
-
- try {
- // The gRPC client policy transformed from the client policy.
- GrpcClientPolicy grpcClientPolicy = toGrpcClientPolicy(policy);
- executor = new GrpcCallExecutor(grpcClientPolicy, authTokenManager, hosts);
- channelProvider.setCallExecutor(executor);
-
- // Warmup after the call executor in the channel provider has
- // been set. The channel provider is used to fetch auth tokens
- // required for the warm up calls.
- executor.warmupChannels();
- }
- catch (Throwable e) {
- if(authTokenManager != null) {
- authTokenManager.close();
- }
- throw e;
- }
- }
-
- /**
- * Return client version string.
- */
- private static String getVersion() {
- final Properties properties = new Properties();
- String version = null;
-
- try {
- properties.load(AerospikeClientProxy.class.getClassLoader().getResourceAsStream("project.properties"));
- version = properties.getProperty("version");
- }
- catch (Exception e) {
- Log.warn("Failed to retrieve client version: " + Util.getErrorMessage(e));
- }
- return version;
- }
-
- //-------------------------------------------------------
- // Default Policies
- //-------------------------------------------------------
-
- /**
- * Copy read policy default to avoid problems if this shared instance is later modified.
- */
- public final Policy getReadPolicyDefault() {
- return new Policy(readPolicyDefault);
- }
-
- /**
- * Copy read policy default.
- */
- public final Policy copyReadPolicyDefault() {
- return new Policy(readPolicyDefault);
- }
-
- /**
- * Copy write policy default to avoid problems if this shared instance is later modified.
- */
- public final WritePolicy getWritePolicyDefault() {
- return new WritePolicy(writePolicyDefault);
- }
-
- /**
- * Copy write policy default.
- */
- public final WritePolicy copyWritePolicyDefault() {
- return new WritePolicy(writePolicyDefault);
- }
-
- /**
- * Copy scan policy default to avoid problems if this shared instance is later modified.
- */
- public final ScanPolicy getScanPolicyDefault() {
- return new ScanPolicy(scanPolicyDefault);
- }
-
- /**
- * Copy scan policy default.
- */
- public final ScanPolicy copyScanPolicyDefault() {
- return new ScanPolicy(scanPolicyDefault);
- }
-
- /**
- * Copy query policy default to avoid problems if this shared instance is later modified.
- */
- public final QueryPolicy getQueryPolicyDefault() {
- return new QueryPolicy(queryPolicyDefault);
- }
-
- /**
- * Copy query policy default.
- */
- public final QueryPolicy copyQueryPolicyDefault() {
- return new QueryPolicy(queryPolicyDefault);
- }
-
- /**
- * Copy batch header read policy default to avoid problems if this shared instance is later modified.
- */
- public final BatchPolicy getBatchPolicyDefault() {
- return new BatchPolicy(batchPolicyDefault);
- }
-
- /**
- * Copy batch header read policy default.
- */
- public final BatchPolicy copyBatchPolicyDefault() {
- return new BatchPolicy(batchPolicyDefault);
- }
-
- /**
- * Copy batch header write policy default to avoid problems if this shared instance is later modified.
- */
- public final BatchPolicy getBatchParentPolicyWriteDefault() {
- return new BatchPolicy(batchParentPolicyWriteDefault);
- }
-
- /**
- * Copy batch header write policy default.
- */
- public final BatchPolicy copyBatchParentPolicyWriteDefault() {
- return new BatchPolicy(batchParentPolicyWriteDefault);
- }
-
- /**
- * Copy batch detail write policy default to avoid problems if this shared instance is later modified.
- */
- public final BatchWritePolicy getBatchWritePolicyDefault() {
- return new BatchWritePolicy(batchWritePolicyDefault);
- }
-
- /**
- * Copy batch detail write policy default.
- */
- public final BatchWritePolicy copyBatchWritePolicyDefault() {
- return new BatchWritePolicy(batchWritePolicyDefault);
- }
-
- /**
- * Copy batch detail delete policy default to avoid problems if this shared instance is later modified.
- */
- public final BatchDeletePolicy getBatchDeletePolicyDefault() {
- return new BatchDeletePolicy(batchDeletePolicyDefault);
- }
-
- /**
- * Copy batch detail delete policy default.
- */
- public final BatchDeletePolicy copyBatchDeletePolicyDefault() {
- return new BatchDeletePolicy(batchDeletePolicyDefault);
- }
-
- /**
- * Copy batch detail UDF policy default to avoid problems if this shared instance is later modified.
- */
- public final BatchUDFPolicy getBatchUDFPolicyDefault() {
- return new BatchUDFPolicy(batchUDFPolicyDefault);
- }
-
- /**
- * Copy batch detail UDF policy default.
- */
- public final BatchUDFPolicy copyBatchUDFPolicyDefault() {
- return new BatchUDFPolicy(batchUDFPolicyDefault);
- }
-
- /**
- * Copy info command policy default to avoid problems if this shared instance is later modified.
- */
- public final InfoPolicy getInfoPolicyDefault() {
- return new InfoPolicy(infoPolicyDefault);
- }
-
- /**
- * Copy info command policy default.
- */
- public final InfoPolicy copyInfoPolicyDefault() {
- return new InfoPolicy(infoPolicyDefault);
- }
-
- /**
- * Copy MRT record version verify policy default.
- */
- public final TxnVerifyPolicy copyTxnVerifyPolicyDefault() {
- return new TxnVerifyPolicy(txnVerifyPolicyDefault);
- }
-
- /**
- * Copy MRT roll forward/back policy default.
- */
- public final TxnRollPolicy copyTxnRollPolicyDefault() {
- return new TxnRollPolicy(txnRollPolicyDefault);
- }
-
- //-------------------------------------------------------
- // Client Management
- //-------------------------------------------------------
-
- /**
- * Close GRPC executor and associated resources. The client instance should not
- * be used after this call.
- */
- @Override
- public void close() {
- try {
- executor.close();
- }
- catch (Throwable e) {
- Log.warn("Failed to close grpcCallExecutor: " + Util.getErrorMessage(e));
- }
-
- try {
- if (authTokenManager != null) {
- authTokenManager.close();
- }
- }
- catch (Throwable e) {
- Log.warn("Failed to close authTokenManager: " + Util.getErrorMessage(e));
- }
-
- if (! sharedThreadPool) {
- // Shutdown synchronous thread pool.
- threadPool.shutdown();
- }
- }
-
- /**
- * This method will always return true in the proxy client.
- */
- @Override
- public boolean isConnected() {
- return executor != null;
- }
-
- /**
- * Not supported in proxy client.
- */
- @Override
- public Node[] getNodes() {
- throw new AerospikeException(NotSupported + "getNodes");
- }
-
- /**
- * Not supported in proxy client.
- */
- @Override
- public List getNodeNames() {
- throw new AerospikeException(NotSupported + "getNodeNames");
- }
-
- /**
- * Not supported in proxy client.
- */
- @Override
- public Node getNode(String nodeName) {
- throw new AerospikeException(NotSupported + "getNode");
- }
-
- /**
- * Not supported in proxy client.
- */
- public final void enableMetrics(MetricsPolicy policy) {
- throw new AerospikeException(NotSupported + "enableMetrics");
- }
-
- /**
- * Not supported in proxy client.
- */
- public final void disableMetrics() {
- throw new AerospikeException(NotSupported + "disableMetrics");
- }
-
- /**
- * Not supported in proxy client.
- */
- @Override
- public ClusterStats getClusterStats() {
- throw new AerospikeException(NotSupported + "getClusterStats");
- }
-
- /**
- * Not supported in proxy client.
- */
- public final void getClusterStats(ClusterStatsListener listener) {
- throw new AerospikeException(NotSupported + "getClusterStats");
- }
-
- /**
- * Not supported in proxy client.
- */
- @Override
- public Cluster getCluster() {
- throw new AerospikeException(NotSupported + "getCluster");
- }
-
- //-------------------------------------------------------
- // Multi-Record Transactions
- //-------------------------------------------------------
-
- /**
- * Attempt to commit the given multi-record transaction. First, the expected record versions are
- * sent to the server nodes for verification. If all nodes return success, the transaction is
- * committed. Otherwise, the transaction is aborted.
- *
- * Requires server version 8.0+
- *
- * @param txn multi-record transaction
- * @return status of the commit on success
- * @throws AerospikeException.Commit if verify commit fails
- */
- public final CommitStatus commit(Txn txn)
- throws AerospikeException.Commit {
- return CommitStatus.OK;
- }
-
- /**
- * Asynchronously attempt to commit the given multi-record transaction. First, the expected
- * record versions are sent to the server nodes for verification. If all nodes return success,
- * the transaction is committed. Otherwise, the transaction is aborted.
- *
- * This method registers the command with an event loop and returns.
- * The event loop thread will process the command and send the results to the listener.
- *
- * Requires server version 8.0+
- *
- * @param eventLoop event loop that will process the command. If NULL, the event
- * loop will be chosen by round-robin.
- * @param listener where to send results
- * @param txn multi-record transaction
- * @throws AerospikeException if event loop registration fails
- */
- public final void commit(EventLoop eventLoop, CommitListener listener, Txn txn)
- throws AerospikeException {
- }
-
- /**
- * Abort and rollback the given multi-record transaction.
- *
- * Requires server version 8.0+
- *
- * @param txn multi-record transaction
- * @return status of the abort
- */
- public final AbortStatus abort(Txn txn) {
- return AbortStatus.OK;
- }
-
- /**
- * Asynchronously abort and rollback the given multi-record transaction.
- *
- * This method registers the command with an event loop and returns.
- * The event loop thread will process the command and send the results to the listener.
- *
- * Requires server version 8.0+
- *
- * @param eventLoop event loop that will process the command. If NULL, the event
- * loop will be chosen by round-robin.
- * @param listener where to send results
- * @param txn multi-record transaction
- * @throws AerospikeException if event loop registration fails
- */
- public final void abort(EventLoop eventLoop, AbortListener listener, Txn txn)
- throws AerospikeException {
- }
-
- //-------------------------------------------------------
- // Write Record Operations
- //-------------------------------------------------------
-
- /**
- * Write record bin(s).
- *
- * @param policy write configuration parameters, pass in null for defaults
- * @param key unique record identifier
- * @param bins array of bin name/value pairs
- * @throws AerospikeException if write fails
- */
- @Override
- public void put(WritePolicy policy, Key key, Bin... bins) {
- CompletableFuture future = new CompletableFuture<>();
- WriteListener listener = prepareWriteListener(future);
- put(null, listener, policy, key, bins);
- getFuture(future);
- }
-
- /**
- * Asynchronously write record bin(s).
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results, pass in null for fire and forget
- * @param policy write configuration parameters, pass in null for defaults
- * @param key unique record identifier
- * @param bins array of bin name/value pairs
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void put(EventLoop eventLoop, WriteListener listener, WritePolicy policy, Key key, Bin... bins) {
- if (policy == null) {
- policy = writePolicyDefault;
- }
- WriteCommandProxy command = new WriteCommandProxy(executor, listener, policy, key, bins, Operation.Type.WRITE);
- command.execute();
- }
-
- //-------------------------------------------------------
- // String Operations
- //-------------------------------------------------------
-
- /**
- * Append bin string values to existing record bin values.
- * This call only works for string values.
- *
- * @param policy write configuration parameters, pass in null for defaults
- * @param key unique record identifier
- * @param bins array of bin name/value pairs
- * @throws AerospikeException if append fails
- */
- @Override
- public void append(WritePolicy policy, Key key, Bin... bins) {
- CompletableFuture future = new CompletableFuture<>();
- WriteListener listener = prepareWriteListener(future);
- append(null, listener, policy, key, bins);
- getFuture(future);
- }
-
- /**
- * Asynchronously append bin string values to existing record bin values.
- * This call only works for string values.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results, pass in null for fire and forget
- * @param policy write configuration parameters, pass in null for defaults
- * @param key unique record identifier
- * @param bins array of bin name/value pairs
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void append(EventLoop eventLoop, WriteListener listener, WritePolicy policy, Key key, Bin... bins) {
- if (policy == null) {
- policy = writePolicyDefault;
- }
- WriteCommandProxy command = new WriteCommandProxy(executor, listener, policy, key, bins, Operation.Type.APPEND);
- command.execute();
- }
-
- /**
- * Prepend bin string values to existing record bin values.
- * This call works only for string values.
- *
- * @param policy write configuration parameters, pass in null for defaults
- * @param key unique record identifier
- * @param bins array of bin name/value pairs
- * @throws AerospikeException if prepend fails
- */
- @Override
- public void prepend(WritePolicy policy, Key key, Bin... bins) {
- CompletableFuture future = new CompletableFuture<>();
- WriteListener listener = prepareWriteListener(future);
- prepend(null, listener, policy, key, bins);
- getFuture(future);
- }
-
- /**
- * Asynchronously prepend bin string values to existing record bin values.
- * This call only works for string values.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results, pass in null for fire and forget
- * @param policy write configuration parameters, pass in null for defaults
- * @param key unique record identifier
- * @param bins array of bin name/value pairs
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void prepend(EventLoop eventLoop, WriteListener listener, WritePolicy policy, Key key, Bin... bins) {
- if (policy == null) {
- policy = writePolicyDefault;
- }
- WriteCommandProxy command = new WriteCommandProxy(executor, listener, policy, key, bins, Operation.Type.PREPEND);
- command.execute();
- }
-
- //-------------------------------------------------------
- // Arithmetic Operations
- //-------------------------------------------------------
-
- /**
- * Add integer/double bin values to existing record bin values.
- *
- * @param policy write configuration parameters, pass in null for defaults
- * @param key unique record identifier
- * @param bins array of bin name/value pairs
- * @throws AerospikeException if add fails
- */
- @Override
- public void add(WritePolicy policy, Key key, Bin... bins) {
- CompletableFuture future = new CompletableFuture<>();
- WriteListener listener = prepareWriteListener(future);
- add(null, listener, policy, key, bins);
- getFuture(future);
- }
-
- /**
- * Asynchronously add integer/double bin values to existing record bin values.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results, pass in null for fire and forget
- * @param policy write configuration parameters, pass in null for defaults
- * @param key unique record identifier
- * @param bins array of bin name/value pairs
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void add(EventLoop eventLoop, WriteListener listener, WritePolicy policy, Key key, Bin... bins) {
- if (policy == null) {
- policy = writePolicyDefault;
- }
- WriteCommandProxy command = new WriteCommandProxy(executor, listener, policy, key, bins, Operation.Type.ADD);
- command.execute();
- }
-
- //-------------------------------------------------------
- // Delete Operations
- //-------------------------------------------------------
-
- /**
- * Delete record for specified key.
- *
- * @param policy delete configuration parameters, pass in null for defaults
- * @param key unique record identifier
- * @return whether record existed on server before deletion
- * @throws AerospikeException if delete fails
- */
- @Override
- public boolean delete(WritePolicy policy, Key key) {
- CompletableFuture future = new CompletableFuture<>();
- DeleteListener listener = prepareDeleteListener(future);
- delete(null, listener, policy, key);
- return getFuture(future);
- }
-
- /**
- * Asynchronously delete record for specified key.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results, pass in null for fire and forget
- * @param policy write configuration parameters, pass in null for defaults
- * @param key unique record identifier
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void delete(EventLoop eventLoop, DeleteListener listener, WritePolicy policy, Key key) {
- if (policy == null) {
- policy = writePolicyDefault;
- }
- DeleteCommandProxy command = new DeleteCommandProxy(executor, listener, policy, key);
- command.execute();
- }
-
- /**
- * Delete records for specified keys. If a key is not found, the corresponding result
- * {@link BatchRecord#resultCode} will be {@link ResultCode#KEY_NOT_FOUND_ERROR}.
- *
- * @param batchPolicy batch configuration parameters, pass in null for defaults
- * @param deletePolicy delete configuration parameters, pass in null for defaults
- * @param keys array of unique record identifiers
- * @throws AerospikeException.BatchRecordArray which contains results for keys that did complete
- */
- @Override
- public BatchResults delete(BatchPolicy batchPolicy, BatchDeletePolicy deletePolicy, Key[] keys) {
- CompletableFuture future = new CompletableFuture<>();
- BatchRecordArrayListener listener = prepareBatchRecordArrayListener(future);
- delete(null, listener, batchPolicy, deletePolicy, keys);
- return getFuture(future);
- }
-
- /**
- * Asynchronously delete records for specified keys.
- *
- * If a key is not found, the corresponding result {@link BatchRecord#resultCode} will be
- * {@link ResultCode#KEY_NOT_FOUND_ERROR}.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results
- * @param batchPolicy batch configuration parameters, pass in null for defaults
- * @param deletePolicy delete configuration parameters, pass in null for defaults
- * @param keys array of unique record identifiers
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void delete(
- EventLoop eventLoop,
- BatchRecordArrayListener listener,
- BatchPolicy batchPolicy,
- BatchDeletePolicy deletePolicy,
- Key[] keys
- ) {
- if (keys.length == 0) {
- listener.onSuccess(new BatchRecord[0], true);
- return;
- }
-
- if (batchPolicy == null) {
- batchPolicy = batchParentPolicyWriteDefault;
- }
-
- if (deletePolicy == null) {
- deletePolicy = batchDeletePolicyDefault;
- }
-
- BatchAttr attr = new BatchAttr();
- attr.setDelete(deletePolicy);
-
- CommandProxy command = new BatchProxy.OperateRecordArrayCommand(executor,
- batchPolicy, keys, null, listener, attr);
-
- command.execute();
- }
-
- /**
- * Asynchronously delete records for specified keys.
- *
- * Each record result is returned in separate onRecord() calls.
- * If a key is not found, the corresponding result {@link BatchRecord#resultCode} will be
- * {@link ResultCode#KEY_NOT_FOUND_ERROR}.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results
- * @param batchPolicy batch configuration parameters, pass in null for defaults
- * @param deletePolicy delete configuration parameters, pass in null for defaults
- * @param keys array of unique record identifiers
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void delete(
- EventLoop eventLoop,
- BatchRecordSequenceListener listener,
- BatchPolicy batchPolicy,
- BatchDeletePolicy deletePolicy,
- Key[] keys
- ) {
- if (keys.length == 0) {
- listener.onSuccess();
- return;
- }
-
- if (batchPolicy == null) {
- batchPolicy = batchParentPolicyWriteDefault;
- }
-
- if (deletePolicy == null) {
- deletePolicy = batchDeletePolicyDefault;
- }
-
- BatchAttr attr = new BatchAttr();
- attr.setDelete(deletePolicy);
-
- CommandProxy command = new BatchProxy.OperateRecordSequenceCommand(executor,
- batchPolicy, keys, null, listener, attr);
-
- command.execute();
- }
-
- /**
- * Not supported in proxy client.
- */
- @Override
- public void truncate(InfoPolicy policy, String ns, String set, Calendar beforeLastUpdate) {
- throw new AerospikeException(NotSupported + "truncate");
- }
-
- //-------------------------------------------------------
- // Touch Operations
- //-------------------------------------------------------
-
- /**
- * Reset record's time to expiration using the policy's expiration.
- * Fail if the record does not exist.
- *
- * @param policy write configuration parameters, pass in null for defaults
- * @param key unique record identifier
- * @throws AerospikeException if touch fails
- */
- @Override
- public void touch(WritePolicy policy, Key key) {
- CompletableFuture future = new CompletableFuture<>();
- WriteListener listener = prepareWriteListener(future);
- touch(null, listener, policy, key);
- getFuture(future);
- }
-
- /**
- * Asynchronously reset record's time to expiration using the policy's expiration.
- * Fail if the record does not exist.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results, pass in null for fire and forget
- * @param policy write configuration parameters, pass in null for defaults
- * @param key unique record identifier
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void touch(EventLoop eventLoop, WriteListener listener, WritePolicy policy, Key key) {
- if (policy == null) {
- policy = writePolicyDefault;
- }
- TouchCommandProxy command = new TouchCommandProxy(executor, listener, policy, key);
- command.execute();
- }
-
- //-------------------------------------------------------
- // Existence-Check Operations
- //-------------------------------------------------------
-
- /**
- * Determine if a record key exists.
- *
- * @param policy generic configuration parameters, pass in null for defaults
- * @param key unique record identifier
- * @return whether record exists or not
- * @throws AerospikeException if command fails
- */
- @Override
- public boolean exists(Policy policy, Key key) {
- CompletableFuture future = new CompletableFuture<>();
- ExistsListener listener = prepareExistsListener(future);
- exists(null, listener, policy, key);
- return getFuture(future);
- }
-
- /**
- * Asynchronously determine if a record key exists.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results
- * @param policy generic configuration parameters, pass in null for defaults
- * @param key unique record identifier
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void exists(EventLoop eventLoop, ExistsListener listener, Policy policy, Key key) {
- if (policy == null) {
- policy = readPolicyDefault;
- }
- ExistsCommandProxy command = new ExistsCommandProxy(executor, listener, policy, key);
- command.execute();
- }
-
- /**
- * Check if multiple record keys exist in one batch call.
- * The returned boolean array is in positional order with the original key array order.
- *
- * @param policy batch configuration parameters, pass in null for defaults
- * @param keys array of unique record identifiers
- * @return array key/existence status pairs
- * @throws AerospikeException.BatchExists which contains results for keys that did complete
- */
- @Override
- public boolean[] exists(BatchPolicy policy, Key[] keys) {
- CompletableFuture future = new CompletableFuture<>();
- ExistsArrayListener listener = prepareExistsArrayListener(future);
- exists(null, listener, policy, keys);
- return getFuture(future);
- }
-
- /**
- * Asynchronously check if multiple record keys exist in one batch call.
- *
- * The returned boolean array is in positional order with the original key array order.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results
- * @param policy batch configuration parameters, pass in null for defaults
- * @param keys unique record identifiers
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void exists(EventLoop eventLoop, ExistsArrayListener listener, BatchPolicy policy, Key[] keys) {
- if (keys.length == 0) {
- listener.onSuccess(keys, new boolean[0]);
- return;
- }
-
- if (policy == null) {
- policy = batchPolicyDefault;
- }
-
- CommandProxy command = new BatchProxy.ExistsArrayCommand(executor, policy, listener, keys);
- command.execute();
- }
-
- /**
- * Asynchronously check if multiple record keys exist in one batch call.
- *
- * Each key's result is returned in separate onExists() calls.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results
- * @param policy batch configuration parameters, pass in null for defaults
- * @param keys unique record identifiers
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void exists(EventLoop eventLoop, ExistsSequenceListener listener, BatchPolicy policy, Key[] keys) {
- if (keys.length == 0) {
- listener.onSuccess();
- return;
- }
-
- if (policy == null) {
- policy = batchPolicyDefault;
- }
-
- CommandProxy command = new BatchProxy.ExistsSequenceCommand(executor, policy, listener, keys);
- command.execute();
- }
-
- //-------------------------------------------------------
- // Read Record Operations
- //-------------------------------------------------------
-
- /**
- * Read entire record for specified key.
- *
- * @param policy generic configuration parameters, pass in null for defaults
- * @param key unique record identifier
- * @return if found, return record instance. If not found, return null.
- * @throws AerospikeException if read fails
- */
- @Override
- public Record get(Policy policy, Key key) {
- return get(policy, key, (String[])null);
- }
-
- /**
- * Asynchronously read entire record for specified key.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results
- * @param policy generic configuration parameters, pass in null for defaults
- * @param key unique record identifier
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void get(EventLoop eventLoop, RecordListener listener, Policy policy, Key key) {
- get(eventLoop, listener, policy, key, (String[])null);
- }
-
- /**
- * Read record header and bins for specified key.
- *
- * @param policy generic configuration parameters, pass in null for defaults
- * @param key unique record identifier
- * @param binNames bins to retrieve
- * @return if found, return record instance. If not found, return null.
- * @throws AerospikeException if read fails
- */
- @Override
- public Record get(Policy policy, Key key, String... binNames) {
- CompletableFuture future = new CompletableFuture<>();
- RecordListener listener = prepareRecordListener(future);
- get(null, listener, policy, key, binNames);
- return getFuture(future);
- }
-
- /**
- * Asynchronously read record header and bins for specified key.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results
- * @param policy generic configuration parameters, pass in null for defaults
- * @param key unique record identifier
- * @param binNames bins to retrieve
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void get(EventLoop eventLoop, RecordListener listener, Policy policy, Key key, String... binNames) {
- if (policy == null) {
- policy = readPolicyDefault;
- }
- ReadCommandProxy command = new ReadCommandProxy(executor, listener, policy, key, binNames);
- command.execute();
- }
-
- /**
- * Read record generation and expiration only for specified key. Bins are not read.
- *
- * @param policy generic configuration parameters, pass in null for defaults
- * @param key unique record identifier
- * @return if found, return record instance. If not found, return null.
- * @throws AerospikeException if read fails
- */
- @Override
- public Record getHeader(Policy policy, Key key) {
- CompletableFuture future = new CompletableFuture<>();
- RecordListener listener = prepareRecordListener(future);
- getHeader(null, listener, policy, key);
- return getFuture(future);
- }
-
- /**
- * Asynchronously read record generation and expiration only for specified key. Bins are not read.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results
- * @param policy generic configuration parameters, pass in null for defaults
- * @param key unique record identifier
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void getHeader(EventLoop eventLoop, RecordListener listener, Policy policy, Key key) {
- if (policy == null) {
- policy = readPolicyDefault;
- }
- ReadHeaderCommandProxy command = new ReadHeaderCommandProxy(executor, listener, policy, key);
- command.execute();
- }
-
- //-------------------------------------------------------
- // Batch Read Operations
- //-------------------------------------------------------
-
- /**
- * Read multiple records for specified batch keys in one batch call.
- * This method allows different namespaces/bins to be requested for each key in the batch.
- * The returned records are located in the same list.
- * If the BatchRead key field is not found, the corresponding record field will be null.
- *
- * @param policy batch configuration parameters, pass in null for defaults
- * @param records list of unique record identifiers and the bins to retrieve.
- * The returned records are located in the same list.
- * @return true if all batch key requests succeeded
- * @throws AerospikeException if read fails
- */
- @Override
- public boolean get(BatchPolicy policy, List records) {
- if (records.size() == 0) {
- return true;
- }
-
- if (policy == null) {
- policy = batchPolicyDefault;
- }
-
- CompletableFuture future = new CompletableFuture<>();
- BatchListListenerSync listener = prepareBatchListListenerSync(future);
-
- CommandProxy command = new BatchProxy.ReadListCommandSync(executor, policy, listener, records);
- command.execute();
-
- return getFuture(future);
- }
-
- /**
- * Asynchronously read multiple records for specified batch keys in one batch call.
- *
- * This method allows different namespaces/bins to be requested for each key in the batch.
- * The returned records are located in the same list.
- * If the BatchRead key field is not found, the corresponding record field will be null.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results
- * @param policy batch configuration parameters, pass in null for defaults
- * @param records list of unique record identifiers and the bins to retrieve.
- * The returned records are located in the same list.
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void get(EventLoop eventLoop, BatchListListener listener, BatchPolicy policy, List records) {
- if (records.size() == 0) {
- listener.onSuccess(records);
- return;
- }
-
- if (policy == null) {
- policy = batchPolicyDefault;
- }
- CommandProxy command = new BatchProxy.ReadListCommand(executor, policy, listener, records);
- command.execute();
- }
-
- /**
- * Asynchronously read multiple records for specified batch keys in one batch call.
- *
- * This method allows different namespaces/bins to be requested for each key in the batch.
- * Each record result is returned in separate onRecord() calls.
- * If the BatchRead key field is not found, the corresponding record field will be null.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results
- * @param policy batch configuration parameters, pass in null for defaults
- * @param records list of unique record identifiers and the bins to retrieve.
- * The returned records are located in the same list.
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void get(EventLoop eventLoop, BatchSequenceListener listener, BatchPolicy policy, List records) {
- if (records.size() == 0) {
- listener.onSuccess();
- return;
- }
-
- if (policy == null) {
- policy = batchPolicyDefault;
- }
-
- CommandProxy command = new BatchProxy.ReadSequenceCommand(executor, policy, listener, records);
- command.execute();
- }
-
- /**
- * Read multiple records 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 null.
- *
- * @param policy batch configuration parameters, pass in null for defaults
- * @param keys array of unique record identifiers
- * @return array of records
- * @throws AerospikeException.BatchRecords which contains results for keys that did complete
- */
- @Override
- public Record[] get(BatchPolicy policy, Key[] keys) {
- CompletableFuture future = new CompletableFuture<>();
- RecordArrayListener listener = prepareRecordArrayListener(future);
- get(null, listener, policy, keys);
- return getFuture(future);
- }
-
- /**
- * Asynchronously read multiple records 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 null.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results
- * @param policy batch configuration parameters, pass in null for defaults
- * @param keys array of unique record identifiers
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void get(EventLoop eventLoop, RecordArrayListener listener, BatchPolicy policy, Key[] keys) {
- if (keys.length == 0) {
- listener.onSuccess(keys, new Record[0]);
- return;
- }
-
- if (policy == null) {
- policy = batchPolicyDefault;
- }
-
- CommandProxy command = new BatchProxy.GetArrayCommand(executor, policy, listener, keys, null, null, Command.INFO1_READ | Command.INFO1_GET_ALL, false);
- command.execute();
- }
-
- /**
- * Asynchronously read multiple records for specified keys in one batch call.
- *
- * Each record result is returned in separate onRecord() calls.
- * If a key is not found, the record will be null.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results
- * @param policy batch configuration parameters, pass in null for defaults
- * @param keys array of unique record identifiers
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void get(EventLoop eventLoop, RecordSequenceListener listener, BatchPolicy policy, Key[] keys) {
- if (keys.length == 0) {
- listener.onSuccess();
- return;
- }
-
- if (policy == null) {
- policy = batchPolicyDefault;
- }
-
- CommandProxy command = new BatchProxy.GetSequenceCommand(executor, policy, listener, keys, null, null, Command.INFO1_READ | Command.INFO1_GET_ALL, false);
- command.execute();
- }
-
- /**
- * 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 null.
- *
- * @param policy batch configuration parameters, pass in null for defaults
- * @param keys array of unique record identifiers
- * @param binNames array of bins to retrieve
- * @return array of records
- * @throws AerospikeException.BatchRecords which contains results for keys that did complete
- */
- @Override
- public Record[] get(BatchPolicy policy, Key[] keys, String... binNames) {
- CompletableFuture future = new CompletableFuture<>();
- RecordArrayListener listener = prepareRecordArrayListener(future);
- get(null, listener, policy, keys, binNames);
- return getFuture(future);
- }
-
- /**
- * Asynchronously 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 null.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results
- * @param policy batch configuration parameters, pass in null for defaults
- * @param keys array of unique record identifiers
- * @param binNames array of bins to retrieve
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void get(EventLoop eventLoop, RecordArrayListener listener, BatchPolicy policy, Key[] keys, String... binNames) {
- if (keys.length == 0) {
- listener.onSuccess(keys, new Record[0]);
- return;
- }
-
- if (policy == null) {
- policy = batchPolicyDefault;
- }
-
- int readAttr = (binNames == null || binNames.length == 0)? Command.INFO1_READ | Command.INFO1_GET_ALL : Command.INFO1_READ;
-
- CommandProxy command = new BatchProxy.GetArrayCommand(executor, policy, listener, keys, binNames, null, readAttr, false);
- command.execute();
- }
-
- /**
- * Asynchronously read multiple record headers and bins for specified keys in one batch call.
- *
- * Each record result is returned in separate onRecord() calls.
- * If a key is not found, the record will be null.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results
- * @param policy batch configuration parameters, pass in null for defaults
- * @param keys array of unique record identifiers
- * @param binNames array of bins to retrieve
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void get(EventLoop eventLoop, RecordSequenceListener listener, BatchPolicy policy, Key[] keys, String... binNames) {
- if (keys.length == 0) {
- listener.onSuccess();
- return;
- }
-
- if (policy == null) {
- policy = batchPolicyDefault;
- }
-
- int readAttr = (binNames == null || binNames.length == 0)? Command.INFO1_READ | Command.INFO1_GET_ALL : Command.INFO1_READ;
-
- CommandProxy command = new BatchProxy.GetSequenceCommand(executor, policy, listener, keys, binNames, null, readAttr, false);
- command.execute();
- }
-
- /**
- * Read multiple records for specified keys using read operations 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 null.
- *
- * @param policy batch configuration parameters, pass in null for defaults
- * @param keys array of unique record identifiers
- * @param ops array of read operations on record
- * @return array of records
- * @throws AerospikeException.BatchRecords which contains results for keys that did complete
- */
- @Override
- public Record[] get(BatchPolicy policy, Key[] keys, Operation... ops) {
- CompletableFuture future = new CompletableFuture<>();
- RecordArrayListener listener = prepareRecordArrayListener(future);
- get(null, listener, policy, keys, ops);
- return getFuture(future);
- }
-
- /**
- * Asynchronously read multiple records for specified keys using read operations 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 null.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results
- * @param policy batch configuration parameters, pass in null for defaults
- * @param keys array of unique record identifiers
- * @param ops array of read operations on record
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void get(EventLoop eventLoop, RecordArrayListener listener, BatchPolicy policy, Key[] keys, Operation... ops) {
- if (keys.length == 0) {
- listener.onSuccess(keys, new Record[0]);
- return;
- }
-
- if (policy == null) {
- policy = batchPolicyDefault;
- }
-
- CommandProxy command = new BatchProxy.GetArrayCommand(executor, policy, listener, keys, null, ops, Command.INFO1_READ, true);
- command.execute();
- }
-
- /**
- * Asynchronously read multiple records for specified keys using read operations in one batch call.
- *
- * Each record result is returned in separate onRecord() calls.
- * If a key is not found, the record will be null.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results
- * @param policy batch configuration parameters, pass in null for defaults
- * @param keys array of unique record identifiers
- * @param ops array of read operations on record
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void get(EventLoop eventLoop, RecordSequenceListener listener, BatchPolicy policy, Key[] keys, Operation... ops) {
- if (keys.length == 0) {
- listener.onSuccess();
- return;
- }
-
- if (policy == null) {
- policy = batchPolicyDefault;
- }
-
- CommandProxy command = new BatchProxy.GetSequenceCommand(executor, policy, listener, keys, null, ops, Command.INFO1_READ, true);
- command.execute();
- }
-
- /**
- * 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 null.
- *
- * @param policy batch configuration parameters, pass in null for defaults
- * @param keys array of unique record identifiers
- * @return array of records
- * @throws AerospikeException.BatchRecords which contains results for keys that did complete
- */
- @Override
- public Record[] getHeader(BatchPolicy policy, Key[] keys) {
- CompletableFuture future = new CompletableFuture<>();
- RecordArrayListener listener = prepareRecordArrayListener(future);
- getHeader(null, listener, policy, keys);
- return getFuture(future);
- }
-
- /**
- * Asynchronously 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 null.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results
- * @param policy batch configuration parameters, pass in null for defaults
- * @param keys array of unique record identifiers
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void getHeader(EventLoop eventLoop, RecordArrayListener listener, BatchPolicy policy, Key[] keys) {
- if (keys.length == 0) {
- listener.onSuccess(keys, new Record[0]);
- return;
- }
-
- if (policy == null) {
- policy = batchPolicyDefault;
- }
-
- CommandProxy command = new BatchProxy.GetArrayCommand(executor, policy, listener, keys, null, null, Command.INFO1_READ | Command.INFO1_NOBINDATA, false);
- command.execute();
- }
-
- /**
- * Asynchronously read multiple record header data for specified keys in one batch call.
- *
- * Each record result is returned in separate onRecord() calls.
- * If a key is not found, the record will be null.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results
- * @param policy batch configuration parameters, pass in null for defaults
- * @param keys array of unique record identifiers
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void getHeader(EventLoop eventLoop, RecordSequenceListener listener, BatchPolicy policy, Key[] keys) {
- if (keys.length == 0) {
- listener.onSuccess();
- return;
- }
-
- if (policy == null) {
- policy = batchPolicyDefault;
- }
-
- CommandProxy command = new BatchProxy.GetSequenceCommand(executor, policy, listener, keys, null, null, Command.INFO1_READ | Command.INFO1_NOBINDATA, false);
- command.execute();
- }
-
- //-------------------------------------------------------
- // Generic Database Operations
- //-------------------------------------------------------
-
- /**
- * 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.
- *
- * The server executes operations in the same order as the operations array.
- * Both scalar bin operations (Operation) and CDT bin operations (ListOperation,
- * MapOperation) can be performed in same call.
- *
- * @param policy write configuration parameters, pass in null for defaults
- * @param key unique record identifier
- * @param operations database operations to perform
- * @return record if there is a read in the operations list
- * @throws AerospikeException if command fails
- */
- @Override
- public Record operate(WritePolicy policy, Key key, Operation... operations) {
- CompletableFuture future = new CompletableFuture<>();
- RecordListener listener = prepareRecordListener(future);
- operate(null, listener, policy, key, operations);
- return getFuture(future);
- }
-
- /**
- * Asynchronously 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.
- *
- * The server executes operations in the same order as the operations array.
- * Both scalar bin operations (Operation) and CDT bin operations (ListOperation,
- * MapOperation) can be performed in same call.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results, pass in null for fire and forget
- * @param policy write configuration parameters, pass in null for defaults
- * @param key unique record identifier
- * @param operations database operations to perform
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void operate(EventLoop eventLoop, RecordListener listener, WritePolicy policy, Key key, Operation... operations) {
- OperateArgs args = new OperateArgs(policy, writePolicyDefault, operatePolicyReadDefault, operations);
- OperateCommandProxy command = new OperateCommandProxy(executor, listener, args.writePolicy, key, args);
- command.execute();
- }
-
- //-------------------------------------------------------
- // Batch Read/Write Operations
- //-------------------------------------------------------
-
- /**
- * Read/Write multiple records for specified batch keys in one batch call.
- * This method allows different namespaces/bins for each key in the batch.
- * The returned records are located in the same list.
- *
- * {@link BatchRecord} can be {@link BatchRead}, {@link BatchWrite}, {@link BatchDelete} or
- * {@link BatchUDF}.
- *
- * @param policy batch configuration parameters, pass in null for defaults
- * @param records list of unique record identifiers and read/write operations
- * @return true if all batch sub-commands succeeded
- * @throws AerospikeException if command fails
- */
- @Override
- public boolean operate(BatchPolicy policy, List records) {
- CompletableFuture future = new CompletableFuture<>();
- BatchOperateListListener listener = prepareBatchOperateListListener(future);
- operate(null, listener, policy, records);
- return getFuture(future);
- }
-
- /**
- * Asynchronously read/write multiple records for specified batch keys in one batch call.
- *
- * This method allows different namespaces/bins to be requested for each key in the batch.
- * The returned records are located in the same list.
- *
- * {@link BatchRecord} can be {@link BatchRead}, {@link BatchWrite}, {@link BatchDelete} or
- * {@link BatchUDF}.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results
- * @param policy batch configuration parameters, pass in null for defaults
- * @param records list of unique record identifiers and read/write operations
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void operate(
- EventLoop eventLoop,
- BatchOperateListListener listener,
- BatchPolicy policy,
- List records
- ) {
- if (records.size() == 0) {
- listener.onSuccess(records, true);
- return;
- }
-
- if (policy == null) {
- policy = batchParentPolicyWriteDefault;
- }
-
- CommandProxy command = new BatchProxy.OperateListCommand(this, executor, policy, listener, records);
- command.execute();
- }
-
- /**
- * Asynchronously read/write multiple records for specified batch keys in one batch call.
- *
- * This method allows different namespaces/bins to be requested for each key in the batch.
- * Each record result is returned in separate onRecord() calls.
- *
- * {@link BatchRecord} can be {@link BatchRead}, {@link BatchWrite}, {@link BatchDelete} or
- * {@link BatchUDF}.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results
- * @param policy batch configuration parameters, pass in null for defaults
- * @param records list of unique record identifiers and read/write operations
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void operate(
- EventLoop eventLoop,
- BatchRecordSequenceListener listener,
- BatchPolicy policy,
- List records
- ) {
- if (records.size() == 0) {
- listener.onSuccess();
- return;
- }
-
- if (policy == null) {
- policy = batchParentPolicyWriteDefault;
- }
-
- CommandProxy command = new BatchProxy.OperateSequenceCommand(this, executor, policy, listener, records);
- command.execute();
- }
-
- /**
- * Perform read/write operations on multiple keys. If a key is not found, the corresponding result
- * {@link BatchRecord#resultCode} will be {@link ResultCode#KEY_NOT_FOUND_ERROR}.
- *
- * @param batchPolicy batch configuration parameters, pass in null for defaults
- * @param writePolicy write configuration parameters, pass in null for defaults
- * @param keys array of unique record identifiers
- * @param ops
- * read/write operations to perform. {@link Operation#get()} is not allowed because it returns a
- * variable number of bins and makes it difficult (sometimes impossible) to lineup operations
- * with results. Instead, use {@link Operation#get(String)} for each bin name.
- * @throws AerospikeException.BatchRecordArray which contains results for keys that did complete
- */
- @Override
- public BatchResults operate(
- BatchPolicy batchPolicy,
- BatchWritePolicy writePolicy,
- Key[] keys,
- Operation... ops
- ) {
- CompletableFuture future = new CompletableFuture<>();
- BatchRecordArrayListener listener = prepareBatchRecordArrayListener(future);
- operate(null, listener, batchPolicy, writePolicy, keys, ops);
- return getFuture(future);
- }
-
- /**
- * Asynchronously perform read/write operations on multiple keys.
- *
- * If a key is not found, the corresponding result {@link BatchRecord#resultCode} will be
- * {@link ResultCode#KEY_NOT_FOUND_ERROR}.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results
- * @param batchPolicy batch configuration parameters, pass in null for defaults
- * @param writePolicy write configuration parameters, pass in null for defaults
- * @param keys array of unique record identifiers
- * @param ops
- * read/write operations to perform. {@link Operation#get()} is not allowed because it returns a
- * variable number of bins and makes it difficult (sometimes impossible) to lineup operations
- * with results. Instead, use {@link Operation#get(String)} for each bin name.
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void operate(
- EventLoop eventLoop,
- BatchRecordArrayListener listener,
- BatchPolicy batchPolicy,
- BatchWritePolicy writePolicy,
- Key[] keys,
- Operation... ops
- ) {
- if (keys.length == 0) {
- listener.onSuccess(new BatchRecord[0], true);
- return;
- }
-
- if (batchPolicy == null) {
- batchPolicy = batchParentPolicyWriteDefault;
- }
-
- if (writePolicy == null) {
- writePolicy = batchWritePolicyDefault;
- }
-
- BatchAttr attr = new BatchAttr(batchPolicy, writePolicy, ops);
-
- CommandProxy command = new BatchProxy.OperateRecordArrayCommand(executor,
- batchPolicy, keys, ops, listener, attr);
-
- command.execute();
- }
-
- /**
- * Asynchronously perform read/write operations on multiple keys.
- *
- * Each record result is returned in separate onRecord() calls.
- * If a key is not found, the corresponding result {@link BatchRecord#resultCode} will be
- * {@link ResultCode#KEY_NOT_FOUND_ERROR}.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results
- * @param batchPolicy batch configuration parameters, pass in null for defaults
- * @param writePolicy write configuration parameters, pass in null for defaults
- * @param keys array of unique record identifiers
- * @param ops
- * read/write operations to perform. {@link Operation#get()} is not allowed because it returns a
- * variable number of bins and makes it difficult (sometimes impossible) to lineup operations
- * with results. Instead, use {@link Operation#get(String)} for each bin name.
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void operate(
- EventLoop eventLoop,
- BatchRecordSequenceListener listener,
- BatchPolicy batchPolicy,
- BatchWritePolicy writePolicy,
- Key[] keys,
- Operation... ops
- ) {
- if (keys.length == 0) {
- listener.onSuccess();
- return;
- }
-
- if (batchPolicy == null) {
- batchPolicy = batchParentPolicyWriteDefault;
- }
-
- if (writePolicy == null) {
- writePolicy = batchWritePolicyDefault;
- }
-
- BatchAttr attr = new BatchAttr(batchPolicy, writePolicy, ops);
-
- CommandProxy command = new BatchProxy.OperateRecordSequenceCommand(executor,
- batchPolicy, keys, ops, listener, attr);
-
- command.execute();
- }
-
- //-------------------------------------------------------
- // Scan Operations
- //-------------------------------------------------------
-
- /**
- * Read all records in specified namespace and set.
- *
- * This call will block until the scan is complete - callbacks are made
- * within the scope of this call.
- *
- * @param policy scan configuration parameters, pass in null for defaults
- * @param namespace namespace - equivalent to database name
- * @param setName optional set name - equivalent to database table
- * @param callback read callback method - called with record data
- * @param binNames optional bin to retrieve. All bins will be returned if not specified.
- * @throws AerospikeException if scan fails
- */
- @Override
- public void scanAll(
- ScanPolicy policy,
- String namespace,
- String setName,
- ScanCallback callback,
- String... binNames
- ) {
- CompletableFuture future = new CompletableFuture<>();
- RecordSequenceListener listener = new RecordSequenceListenerToCallback(callback, future);
- scanPartitions(null, listener, policy, null, namespace, setName, binNames);
- getFuture(future);
- }
-
- /**
- * Asynchronously read all records in specified namespace and set.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results
- * @param policy scan configuration parameters, pass in null for defaults
- * @param namespace namespace - equivalent to database name
- * @param setName optional set name - equivalent to database table
- * @param binNames optional bin to retrieve. All bins will be returned if not specified.
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void scanAll(
- EventLoop eventLoop,
- RecordSequenceListener listener,
- ScanPolicy policy,
- String namespace,
- String setName,
- String... binNames
- ) {
- scanPartitions(eventLoop, listener, policy, null, namespace, setName, binNames);
- }
-
- /**
- * Not supported in proxy client.
- */
- @Override
- public void scanNode(
- ScanPolicy policy,
- String nodeName,
- String namespace,
- String setName,
- ScanCallback callback,
- String... binNames
- ) {
- throw new AerospikeException(NotSupported + "scanNode");
- }
-
- /**
- * Not supported in proxy client.
- */
- @Override
- public void scanNode(
- ScanPolicy policy,
- Node node,
- String namespace,
- String setName,
- ScanCallback callback,
- String... binNames
- ) {
- throw new AerospikeException(NotSupported + "scanNode");
- }
-
- /**
- * Read records in specified namespace, set and partition filter.
- *
- * This call will block until the scan is complete - callbacks are made
- * within the scope of this call.
- *
- * @param policy scan configuration parameters, pass in null for defaults
- * @param partitionFilter filter on a subset of data partitions
- * @param namespace namespace - equivalent to database name
- * @param setName optional set name - equivalent to database table
- * @param callback read callback method - called with record data
- * @param binNames optional bin to retrieve. All bins will be returned if not specified
- * @throws AerospikeException if scan fails
- */
- @Override
- public void scanPartitions(
- ScanPolicy policy,
- PartitionFilter partitionFilter,
- String namespace,
- String setName,
- ScanCallback callback,
- String... binNames
- ) {
- CompletableFuture future = new CompletableFuture<>();
- RecordSequenceListener listener = new RecordSequenceListenerToCallback(callback, future);
- scanPartitions(null, listener, policy, partitionFilter, namespace, setName, binNames);
- getFuture(future);
- }
-
- /**
- * Asynchronously read records in specified namespace, set and partition filter.
- *
- * @param eventLoop ignored, pass in null
- * @param listener where to send results
- * @param policy scan configuration parameters, pass in null for defaults
- * @param partitionFilter filter on a subset of data partitions
- * @param namespace namespace - equivalent to database name
- * @param setName optional set name - equivalent to database table
- * @param binNames optional bin to retrieve. All bins will be returned if not specified.
- * @throws AerospikeException if event loop registration fails
- */
- @Override
- public void scanPartitions(
- EventLoop eventLoop,
- RecordSequenceListener listener,
- ScanPolicy policy,
- PartitionFilter partitionFilter,
- String namespace,
- String setName,
- String... binNames
- ) {
- if (policy == null) {
- policy = scanPolicyDefault;
- }
-
- PartitionTracker tracker = null;
-
- if (partitionFilter != null) {
- tracker = new PartitionTracker(policy, 1, partitionFilter);
- }
-
- ScanCommandProxy command = new ScanCommandProxy(executor, policy, listener, namespace,
- setName, binNames, partitionFilter, tracker);
- command.execute();
- }
-
- //---------------------------------------------------------------
- // User defined functions
- //---------------------------------------------------------------
-
- /**
- * Not supported in proxy client.
- */
- @Override
- public RegisterTask register(Policy policy, String clientPath, String serverPath, Language language) {
- throw new AerospikeException(NotSupported + "register");
- }
-
- /**
- * Not supported in proxy client.
- */
- @Override
- public RegisterTask register(
- Policy policy,
- ClassLoader resourceLoader,
- String resourcePath,
- String serverPath,
- Language language
- ) {
- throw new AerospikeException(NotSupported + "register");
- }
-
- /**
- * Not supported in proxy client.
- */
- @Override
- public RegisterTask registerUdfString(Policy policy, String code, String serverPath, Language language) {
- throw new AerospikeException(NotSupported + "registerUdfString");
- }
-
- /**
- * Not supported in proxy client.
- */
- @Override
- public void removeUdf(InfoPolicy policy, String serverPath) {
- throw new AerospikeException(NotSupported + "removeUdf");
- }
-
- /**
- * 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:
- *
- * {@code udf file = /.lua}
- *
- * @param policy write configuration parameters, pass in null for defaults
- * @param key unique record identifier
- * @param packageName server package name where user defined function resides
- * @param functionName user defined function
- * @param functionArgs arguments passed in to user defined function
- * @return return value of user defined function
- * @throws AerospikeException if command fails
- */
- @Override
- public Object execute(WritePolicy policy, Key key, String packageName, String functionName, Value... functionArgs) {
- CompletableFuture