import com.aerospike.client.policy.Policy; //导入依赖的package包/类
/**
 * Reads and processes Read {@link Policy} from the configuration file.
 * 
 * @return
 */
private Policy readPolicy(String clazzName) {
	String className = clazzName.isEmpty() ? "" : clazzName + ".";
	Policy policy = new Policy();
	// Read max retries
	int maxRetries = readIntegerProperty(className + READ_POLICY + "maxRetries");
	policy.maxRetries = maxRetries != ERROR_INT_VALUE ? maxRetries : policy.maxRetries;

	// Read priority
	Enum priority = readEnumProperty(className + READ_POLICY + "priority", Priority.DEFAULT);
	policy.priority = priority != ERROR_ENUM_VALUE ? (Priority) priority : policy.priority;

	// Read timeout
	int timeout = readIntegerProperty(className + READ_POLICY + "timeout");
	policy.timeout = timeout != ERROR_INT_VALUE ? timeout : policy.timeout;

	// Read sleep between retires
	int sleepBetweenRetries = readIntegerProperty(className + READ_POLICY + "sleepBetweenRetries");
	policy.sleepBetweenRetries = sleepBetweenRetries != ERROR_INT_VALUE ? sleepBetweenRetries : policy.sleepBetweenRetries;
	return policy;
}
 

import com.aerospike.client.policy.Policy; //导入依赖的package包/类
private String sendInfoCommand(Policy policy, String command) throws AerospikeException {		
	Node node = cluster.getRandomNode();
	int timeout = (policy == null)? 0 : policy.timeout;
	Connection conn = node.getConnection(timeout);
	Info info;
	
	try {
		info = new Info(conn, command);
		node.putConnection(conn);
	}
	catch (AerospikeException ae) {
		conn.close();
		throw ae;
	}
	catch (RuntimeException re) {
		conn.close();
		throw new AerospikeException(re);
	}
	return info.getValue();
}
 

import com.aerospike.client.policy.Policy; //导入依赖的package包/类
public BatchCommandGet(
	Node node,
	BatchNode.BatchNamespace batchNamespace,
	Policy policy,		
	HashMap<Key,BatchItem> keyMap,
	HashSet<String> binNames,
	Record[] records,
	int readAttr
) {
	super(node);
	this.batchNamespace = batchNamespace;
	this.policy = policy;
	this.keyMap = keyMap;
	this.binNames = binNames;
	this.records = records;
	this.readAttr = readAttr;
}
 

import com.aerospike.client.policy.Policy; //导入依赖的package包/类
public AsyncBatchGetArray(
	AsyncMultiExecutor parent,
	AsyncCluster cluster,
	AsyncNode node,
	BatchNode.BatchNamespace batchNamespace,
	Policy policy,
	HashMap<Key,BatchItem> keyMap,
	HashSet<String> binNames,
	Record[] records,
	int readAttr
) {
	super(parent, cluster, node, false, binNames);
	this.batchNamespace = batchNamespace;
	this.policy = policy;
	this.keyMap = keyMap;
	this.records = records;
	this.readAttr = readAttr;
}
 

import com.aerospike.client.policy.Policy; //导入依赖的package包/类
public AsyncBatchGetSequence(
	AsyncMultiExecutor parent,
	AsyncCluster cluster,
	AsyncNode node,
	BatchNode.BatchNamespace batchNamespace,
	Policy policy,
	HashSet<String> binNames,
	RecordSequenceListener listener,
	int readAttr
) {
	super(parent, cluster, node, false, binNames);
	this.batchNamespace = batchNamespace;
	this.policy = policy;
	this.listener = listener;
	this.readAttr = readAttr;
}
 

import com.aerospike.client.policy.Policy; //导入依赖的package包/类
public AerospikeDB(String namespace) {
	AerospikeDB.NAMESPACE = namespace;
	writePolicy = new WritePolicy();
	readPolicy = new Policy();
	batchPolicy = new BatchPolicy();
	client = new AerospikeClient(Constants.IP, Constants.PORT);
}
 

import com.aerospike.client.policy.Policy; //导入依赖的package包/类
private static void createIndexes(Parameters params,
								  AerospikeClient client) throws Exception {
	Policy policy = new Policy();
	policy.timeout = 0; // Do not timeout on index create.
	IndexTask task =
		client.createIndex(policy, params.namespace, params.set,
						   params.locndx, params.locbin,
						   IndexType.GEO2DSPHERE);
	task.waitTillComplete();
}
 

import com.aerospike.client.policy.Policy; //导入依赖的package包/类
@Override
public Drone getDroneWhereIdIs(int id) {
  Key key = new Key(database.namespace, setName, id);
  Policy readPolicy = new Policy();
  ++Metering.droneGets;
  Record record = database.client.get(readPolicy, key);
  return record == null ? null : get(key, record);
}
 

import com.aerospike.client.policy.Policy; //导入依赖的package包/类
/**
 * 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:
 * <p>
 * udf file = <server udf dir>/<package name>.lua
 * <p>
 * This method is only supported by Aerospike 3 servers.
 * 
 * @param policy				generic 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 args					arguments passed in to user defined function
 * @return						return value of user defined function
 * @throws AerospikeException	if transaction fails
 */
public final Object execute(Policy policy, Key key, String packageName, String functionName, Value... args) 
	throws AerospikeException {
	ExecuteCommand command = new ExecuteCommand(cluster, policy, key, packageName, functionName, args);
	command.execute();
	
	Record record = command.getRecord();
	
	if (record == null || record.bins == null) {
		return null;
	}
	
	Map<String,Object> map = record.bins;

	Object obj = map.get("SUCCESS");
	
	if (obj != null) {
		return obj;
	}
	
	// User defined functions don't have to return a value.
	if (map.containsKey("SUCCESS")) {
		return null;
	}
	
	obj = map.get("FAILURE");
	
	if (obj != null) {
		throw new AerospikeException(obj.toString());
	}
	throw new AerospikeException("Invalid UDF return value");
}
 

import com.aerospike.client.policy.Policy; //导入依赖的package包/类
/**
 * Delete secondary index.
 * This method is only supported by Aerospike 3 servers.
 * 
 * @param policy				generic configuration parameters, pass in null for defaults
 * @param namespace				namespace - equivalent to database name
 * @param setName				optional set name - equivalent to database table
 * @param indexName				name of secondary index
 * @throws AerospikeException	if index create fails
 */
public final void dropIndex(
	Policy policy, 
	String namespace, 
	String setName, 
	String indexName
) throws AerospikeException {
					
	StringBuilder sb = new StringBuilder(500);
	sb.append("sindex-delete:ns=");
	sb.append(namespace);
	
	if (setName != null && setName.length() > 0) {
		sb.append(";set=");
		sb.append(setName);
	}		
	sb.append(";indexname=");
	sb.append(indexName);
	
	// Send index command to one node. That node will distribute the command to other nodes.
	String response = sendInfoCommand(policy, sb.toString());

	if (response.equalsIgnoreCase("OK")) {
		return;
	}
	
	if (response.startsWith("FAIL:201")) {
		// Index did not previously exist. Return without error.
		return;
	}
		
	throw new AerospikeException("Drop index failed: " + response);
}
 

import com.aerospike.client.policy.Policy; //导入依赖的package包/类
public ExecuteCommand(
	Cluster cluster, 
	Policy policy,
	Key key,
	String packageName,
	String functionName,
	Value[] args
) {
	super(cluster, policy, key, null);
	this.packageName = packageName;
	this.functionName = functionName;
	this.args = args;
}
 

import com.aerospike.client.policy.Policy; //导入依赖的package包/类
public BatchCommandExists(
	Node node,
	BatchNode.BatchNamespace batchNamespace,
	Policy policy,
	HashMap<Key,BatchItem> keyMap,
	boolean[] existsArray
) {
	super(node);
	this.batchNamespace = batchNamespace;
	this.policy = policy;
	this.keyMap = keyMap;
	this.existsArray = existsArray;
}
 

import com.aerospike.client.policy.Policy; //导入依赖的package包/类
public AsyncBatchExistsSequence(
	AsyncMultiExecutor parent,
	AsyncCluster cluster,
	AsyncNode node,
	BatchNode.BatchNamespace batchNamespace,
	Policy policy,
	ExistsSequenceListener listener
) {
	super(parent, cluster, node, false);
	this.batchNamespace = batchNamespace;
	this.policy = policy;
	this.listener = listener;
}
 

import com.aerospike.client.policy.Policy; //导入依赖的package包/类
public void execute() throws AerospikeException {
	Policy policy = getPolicy();
	timeout = policy.timeout;
	
	if (timeout > 0) {	
		limit = System.currentTimeMillis() + timeout;
	}
	
	byteBuffer = cluster.getByteBuffer();
	executeCommand();
}
 

import com.aerospike.client.policy.Policy; //导入依赖的package包/类
private boolean retryOnInit() throws AerospikeException {
	if (complete.get()) {
		return true;
	}

	Policy policy = getPolicy();
	
	if (++iteration > policy.maxRetries) {
		return failOnNetworkInit();
	}

	if (limit > 0 && System.currentTimeMillis() + policy.sleepBetweenRetries > limit) {
		// Might as well stop here because the transaction will
		// timeout after sleep completed.
		return failOnNetworkInit();
	}

	// Prepare for retry.
	resetConnection();

	if (policy.sleepBetweenRetries > 0) {
		Util.sleep(policy.sleepBetweenRetries);
	}

	// Retry command recursively.
	executeCommand();
	return true;
}
 

import com.aerospike.client.policy.Policy; //导入依赖的package包/类
protected final void retryAfterInit(AerospikeException ae) {
	if (complete.get()) {
		return;
	}

	Policy policy = getPolicy();

	if (++iteration > policy.maxRetries) {
		failOnNetworkError(ae);
		return;
	}

	if (limit > 0 && System.currentTimeMillis() + policy.sleepBetweenRetries > limit) {
		// Might as well stop here because the transaction will
		// timeout after sleep completed.
		failOnNetworkError(ae);
		return;
	}

	// Prepare for retry.
	resetConnection();

	if (policy.sleepBetweenRetries > 0) {
		Util.sleep(policy.sleepBetweenRetries);
	}

	try {
		// Retry command recursively.
		executeCommand();
	}
	catch (Exception e) {
		// Command has already been cleaned up.
		// Notify user of original exception.
		onFailure(ae);
	}
}
 

import com.aerospike.client.policy.Policy; //导入依赖的package包/类
public AsyncBatchExistsArray(
	AsyncMultiExecutor parent,
	AsyncCluster cluster,
	AsyncNode node,
	BatchNode.BatchNamespace batchNamespace,
	Policy policy,
	HashMap<Key,BatchItem> keyMap,
	boolean[] existsArray
) {
	super(parent, cluster, node, false);
	this.batchNamespace = batchNamespace;
	this.policy = policy;
	this.keyMap = keyMap;
	this.existsArray = existsArray;
}
 

import com.aerospike.client.policy.Policy; //导入依赖的package包/类
@Override
public void configure(JobConf job) {
    String host = AerospikeConfigUtil.getInputHost(job);
    int port = AerospikeConfigUtil.getInputPort(job);

    policy = new Policy();
    policy.timeout = 10000;
    client = new AerospikeClient(host, port);

    namespace = AerospikeConfigUtil.getInputNamespace(job);
    setName = AerospikeConfigUtil.getInputSetName(job);
}
 

import com.aerospike.client.policy.Policy; //导入依赖的package包/类
protected DbParameters(String host, int port, 
		String namespace, String baseNamespace, String cacheNamespace) {
	this.host = host;
	this.port = port;
	this.namespace = namespace;
	this.baseNamespace = baseNamespace;
	this.cacheNamespace = cacheNamespace;
	this.writePolicy = new WritePolicy();
	this.writePolicy.timeout = 1000;
	this.writePolicy.maxRetries = 0;
	this.policy = new Policy();
}
 

import com.aerospike.client.policy.Policy; //导入依赖的package包/类
/**
 * Constructor for LMAP OPERATION class.
 * @param client
 * @param console
 * @throws AerospikeException
 */
public LMapOperations(AerospikeClient client, Console console) 
		throws AerospikeException 
{

	this.client = client;	
	this.ldtPolicy = new Policy();
	this.console = console;
}