import com.aerospike.client.cluster.Node; //导入依赖的package包/类
public static String info(Node node, String name) {
  try {
    String result = Info.request(node.getHost().name, node.getHost().port, name);
    return result;
    // The result looks like this (broken here into 5 lines). Node the ; in the middle.
    //   reads:20:07:45-GMT,ops/sec,>1ms,>8ms,>64ms;20:07:55,0.0,0.00,0.00,0.00;
    //   writes_master:20:07:45-GMT,ops/sec,>1ms,>8ms,>64ms;20:07:55,0.0,0.00,0.00,0.00;
    //   proxy:20:07:45-GMT,ops/sec,>1ms,>8ms,>64ms;20:07:55,0.0,0.00,0.00,0.00;
    //   udf:20:07:45-GMT,ops/sec,>1ms,>8ms,>64ms;20:07:55,0.0,0.00,0.00,0.00;
    //   query:20:07:45-GMT,ops/sec,>1ms,>8ms,>64ms;20:07:55,0.0,0.00,0.00,0.00;
  } catch (AerospikeException e) {
    int resultCode = e.getResultCode();
    System.err.format("info request %s %s %s\n", name, ResultCode.getResultString(resultCode), e);
  }
  return null;
}
 

import com.aerospike.client.cluster.Node; //导入依赖的package包/类
private KeyRecordIterator queryByLua(Statement stmt, Boolean metaOnly, Node node, Qualifier[] qualifiers){
	Map<String, Object> originArgs = new HashMap<String, Object>();
	originArgs.put("includeAllFields", 1);
	ResultSet resultSet = null;

	String filterFuncStr = buildFilterFunction(qualifiers);
	originArgs.put("filterFuncStr", filterFuncStr);

	if (metaOnly)
		stmt.setAggregateFunction(this.getClass().getClassLoader(), AS_UTILITY_PATH, QUERY_MODULE, "query_meta", Value.get(originArgs));
	else
		stmt.setAggregateFunction(this.getClass().getClassLoader(), AS_UTILITY_PATH, QUERY_MODULE, "select_records", Value.get(originArgs));
	if (node != null) {
		resultSet = this.client.queryAggregateNode(queryPolicy, stmt, node);
	} else {
		resultSet = this.client.queryAggregate(queryPolicy, stmt);
	}
	return new KeyRecordIterator(stmt.getNamespace(), resultSet);

}
 

import com.aerospike.client.cluster.Node; //导入依赖的package包/类
public void refreshNamespaceData(Node node, Namespace namespace) {
	/*
	 * refresh namespace data
	 */
	try {
		String nameSpaceString = Info.request(infoPolicy, node, "namespace/" + namespace);
		namespace.mergeNamespaceInfo(nameSpaceString);
		String setsString = Info.request(infoPolicy, node, "sets/" + namespace);
		if (!setsString.isEmpty()) {
			String[] sets = setsString.split(";");
			for (String setData : sets) {
				namespace.mergeSet(setData);
			}
		}
	} catch (AerospikeException e) {
		log.error("Error geting Namespace details", e);
	}
}
 

import com.aerospike.client.cluster.Node; //导入依赖的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.cluster.Node; //导入依赖的package包/类
/**
 * Get one info value by name from the specified database server node.
 * 
 * @param node					server node
 * @param name					name of value to retrieve
 * @return						info value
 */
public static String request(Node node, String name) 
	throws AerospikeException {
	Connection conn = node.getConnection(DEFAULT_TIMEOUT);
	
	try {
		String response = Info.request(conn, name);
		node.putConnection(conn);
		return response;
	}
	catch (AerospikeException ae) {
		conn.close();
		throw ae;
	}
	catch (RuntimeException re) {
		conn.close();
		throw re;
	}
}
 

import com.aerospike.client.cluster.Node; //导入依赖的package包/类
/**
 * Query all nodes for task completion status.
 */
public boolean isDone() throws AerospikeException {
	String command = "udf-list";
	Node[] nodes = cluster.getNodes();
	boolean done = false;

	for (Node node : nodes) {
		String response = Info.request(node, command);
		String find = "filename=" + packageName;
		int index = response.indexOf(find);

		if (index < 0) {
			return false;
		}
		done = true;
	}
	return done;
}
 

import com.aerospike.client.cluster.Node; //导入依赖的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.cluster.Node; //导入依赖的package包/类
public AsyncScanExecutor(
	AsyncCluster cluster,
	ScanPolicy policy,
	RecordSequenceListener listener,
	String namespace,
	String setName,
	String[] binNames
) throws AerospikeException {
	this.listener = listener;

	Node[] nodes = cluster.getNodes();
	if (nodes.length == 0) {
		throw new AerospikeException(ResultCode.SERVER_NOT_AVAILABLE, "Scan failed because cluster is empty.");
	}

	completedSize = nodes.length;

	for (Node node : nodes) {			
		AsyncScan async = new AsyncScan(this, cluster, (AsyncNode)node, policy, listener, namespace, setName, binNames);
		async.execute();
	}
}
 

import com.aerospike.client.cluster.Node; //导入依赖的package包/类
/**
 * Retrieves and sets the server specific parameters
 * Validates the existence of user provided namespace and validates for single binned
 * namespaces
 *
 * @param client aerospike client used to connect with the server
 */
public void setServerSpecificParameters(AerospikeClient client) {

  String namespaceTokens = null;
  for (Node node : client.getNodes()) {
    String namespaceFilter = "namespace/" + aerospikeMapping.getNamespace();
    namespaceTokens = Info.request(null, node, namespaceFilter);

    if (namespaceTokens != null) {
      isSingleBinEnabled = parseBoolean(namespaceTokens, "single-bin");
      break;
    }
  }
  if (namespaceTokens == null) {
    LOG.error("Failed to get namespace info from Aerospike");
    throw new RuntimeException("Failed to get namespace info from Aerospike");
  }
}
 

import com.aerospike.client.cluster.Node; //导入依赖的package包/类
public static void printNodes(AerospikeClient client) {
    if (client == null) {
        LOG.warn("null client has been sent returning");
        return;
    }
    for (Node node : client.getNodes()) {
        LOG.info("node name is " + node.getAddress().getHostString());
    }
}
 

import com.aerospike.client.cluster.Node; //导入依赖的package包/类
public static String[] infoAll(AerospikeClient client, String name) {
  String[] result = new String[client.getNodes().length];
  int i = 0;
  for (Node node : client.getNodes()) {
    result[i++] = info(node, name);
  }
  return result;
}
 

import com.aerospike.client.cluster.Node; //导入依赖的package包/类
/**
 * refreshes the cached Namespace information
 */
public synchronized void refreshNamespaces() {
	/*
	 * cache namespaces
	 */
	if (this.namespaceCache == null) {
		this.namespaceCache = new TreeMap<String, Namespace>();
		Node[] nodes = client.getNodes();
		for (Node node : nodes) {
			try {
				String namespaceString = Info.request(getInfoPolicy(), node, "namespaces");
				if (!namespaceString.isEmpty()) {
					String[] namespaceList = namespaceString.split(";");
					for (String namespace : namespaceList) {
						Namespace ns = this.namespaceCache.get(namespace);
						if (ns == null) {
							ns = new Namespace(namespace);
							this.namespaceCache.put(namespace, ns);
						}
						refreshNamespaceData(node, ns);
					}
				}
			} catch (AerospikeException e) {
				log.error("Error geting Namespaces ", e);
			}

		}
	}
}
 

import com.aerospike.client.cluster.Node; //导入依赖的package包/类
/**
 * refreshes the Index cache from the Cluster
 */
public synchronized void refreshIndexes() {
	/*
	 * cache index by Bin name
	 */
	if (this.indexCache == null)
		this.indexCache = new TreeMap<String, Index>();

	Node[] nodes = client.getNodes();
	for (Node node : nodes) {
		if (node.isActive()) {
			try {
				String indexString = Info.request(getInfoPolicy(), node, "sindex");
				if (!indexString.isEmpty()) {
					String[] indexList = indexString.split(";");
					for (String oneIndexString : indexList) {
						Index index = new Index(oneIndexString);
						this.indexCache.put(index.toKeyString(), index);
					}
				}
				break;
			} catch (AerospikeException e) {
				log.error("Error geting Index informaton", e);
			}
		}
	}
}
 

import com.aerospike.client.cluster.Node; //导入依赖的package包/类
/**
 * refreshes the Module cache from the cluster. The Module cache contains a list of register UDF modules.
 */
public synchronized void refreshModules() {
	if (this.moduleCache == null)
		this.moduleCache = new TreeMap<String, Module>();
	boolean loadedModules = false;
	Node[] nodes = client.getNodes();
	for (Node node : nodes) {
		try {

			String packagesString = Info.request(infoPolicy, node, "udf-list");
			if (!packagesString.isEmpty()) {
				String[] packagesList = packagesString.split(";");
				for (String pkgString : packagesList) {
					Module module = new Module(pkgString);
					String udfString = Info.request(infoPolicy, node, "udf-get:filename=" + module.getName());
					module.setDetailInfo(udfString);//gen=qgmyp0d8hQNvJdnR42X3BXgUGPE=;type=LUA;recordContent=bG9jYWwgZnVuY3Rpb24gcHV0QmluKHIsbmFtZSx2YWx1ZSkKICAgIGlmIG5vdCBhZXJvc3Bpa2U6ZXhpc3RzKHIpIHRoZW4gYWVyb3NwaWtlOmNyZWF0ZShyKSBlbmQKICAgIHJbbmFtZV0gPSB2YWx1ZQogICAgYWVyb3NwaWtlOnVwZGF0ZShyKQplbmQKCi0tIFNldCBhIHBhcnRpY3VsYXIgYmluCmZ1bmN0aW9uIHdyaXRlQmluKHIsbmFtZSx2YWx1ZSkKICAgIHB1dEJpbihyLG5hbWUsdmFsdWUpCmVuZAoKLS0gR2V0IGEgcGFydGljdWxhciBiaW4KZnVuY3Rpb24gcmVhZEJpbihyLG5hbWUpCiAgICByZXR1cm4gcltuYW1lXQplbmQKCi0tIFJldHVybiBnZW5lcmF0aW9uIGNvdW50IG9mIHJlY29yZApmdW5jdGlvbiBnZXRHZW5lcmF0aW9uKHIpCiAgICByZXR1cm4gcmVjb3JkLmdlbihyKQplbmQKCi0tIFVwZGF0ZSByZWNvcmQgb25seSBpZiBnZW4gaGFzbid0IGNoYW5nZWQKZnVuY3Rpb24gd3JpdGVJZkdlbmVyYXRpb25Ob3RDaGFuZ2VkKHIsbmFtZSx2YWx1ZSxnZW4pCiAgICBpZiByZWNvcmQuZ2VuKHIpID09IGdlbiB0aGVuCiAgICAgICAgcltuYW1lXSA9IHZhbHVlCiAgICAgICAgYWVyb3NwaWtlOnVwZGF0ZShyKQogICAgZW5kCmVuZAoKLS0gU2V0IGEgcGFydGljdWxhciBiaW4gb25seSBpZiByZWNvcmQgZG9lcyBub3QgYWxyZWFkeSBleGlzdC4KZnVuY3Rpb24gd3JpdGVVbmlxdWUocixuYW1lLHZhbHVlKQogICAgaWYgbm90IGFlcm9zcGlrZTpleGlzdHMocikgdGhlbiAKICAgICAgICBhZXJvc3Bpa2U6Y3JlYXRlKHIpIAogICAgICAgIHJbbmFtZV0gPSB2YWx1ZQogICAgICAgIGFlcm9zcGlrZTp1cGRhdGUocikKICAgIGVuZAplbmQKCi0tIFZhbGlkYXRlIHZhbHVlIGJlZm9yZSB3cml0aW5nLgpmdW5jdGlvbiB3cml0ZVdpdGhWYWxpZGF0aW9uKHIsbmFtZSx2YWx1ZSkKICAgIGlmICh2YWx1ZSA+PSAxIGFuZCB2YWx1ZSA8PSAxMCkgdGhlbgogICAgICAgIHB1dEJpbihyLG5hbWUsdmFsdWUpCiAgICBlbHNlCiAgICAgICAgZXJyb3IoIjEwMDA6SW52YWxpZCB2YWx1ZSIpIAogICAgZW5kCmVuZAoKLS0gUmVjb3JkIGNvbnRhaW5zIHR3byBpbnRlZ2VyIGJpbnMsIG5hbWUxIGFuZCBuYW1lMi4KLS0gRm9yIG5hbWUxIGV2ZW4gaW50ZWdlcnMsIGFkZCB2YWx1ZSB0byBleGlzdGluZyBuYW1lMSBiaW4uCi0tIEZvciBuYW1lMSBpbnRlZ2VycyB3aXRoIGEgbXVsdGlwbGUgb2YgNSwgZGVsZXRlIG5hbWUyIGJpbi4KLS0gRm9yIG5hbWUxIGludGVnZXJzIHdpdGggYSBtdWx0aXBsZSBvZiA5LCBkZWxldGUgcmVjb3JkLiAKZnVuY3Rpb24gcHJvY2Vzc1JlY29yZChyLG5hbWUxLG5hbWUyLGFkZFZhbHVlKQogICAgbG9jYWwgdiA9IHJbbmFtZTFdCgogICAgaWYgKHYgJSA5ID09IDApIHRoZW4KICAgICAgICBhZXJvc3Bpa2U6cmVtb3ZlKHIpCiAgICAgICAgcmV0dXJuCiAgICBlbmQKCiAgICBpZiAodiAlIDUgPT0gMCkgdGhlbgogICAgICAgIHJbbmFtZTJdID0gbmlsCiAgICAgICAgYWVyb3NwaWtlOnVwZGF0ZShyKQogICAgICAgIHJldHVybgogICAgZW5kCgogICAgaWYgKHYgJSAyID09IDApIHRoZW4KICAgICAgICByW25hbWUxXSA9IHYgKyBhZGRWYWx1ZQogICAgICAgIGFlcm9zcGlrZTp1cGRhdGUocikKICAgIGVuZAplbmQKCi0tIFNldCBleHBpcmF0aW9uIG9mIHJlY29yZAotLSBmdW5jdGlvbiBleHBpcmUocix0dGwpCi0tICAgIGlmIHJlY29yZC50dGwocikgPT0gZ2VuIHRoZW4KLS0gICAgICAgIHJbbmFtZV0gPSB2YWx1ZQotLSAgICAgICAgYWVyb3NwaWtlOnVwZGF0ZShyKQotLSAgICBlbmQKLS0gZW5kCg==;
					this.moduleCache.put(module.getName(), module);
				}
			}
			loadedModules = true;
			break;
		} catch (AerospikeException e) {

		}
	}
	if (!loadedModules) {
		throw new ClusterRefreshError("Cannot find UDF modules");
	}
}
 

import com.aerospike.client.cluster.Node; //导入依赖的package包/类
/**
 * Sends an "Info" command to all nodes in the cluster
 *
 * @param client AerospikeClient instance
 * @param cmd	Info command to be sent to the cluster
 * @return A string containing the results from all nodes in the cluster
 */
public static String infoAll(AerospikeClient client, String cmd) {
	Node[] nodes = client.getNodes();
	StringBuilder results = new StringBuilder();
	for (Node node : nodes) {
		results.append(Info.request(node.getHost().name, node.getHost().port, cmd)).append("\n");
	}
	return results.toString();
}
 

import com.aerospike.client.cluster.Node; //导入依赖的package包/类
/**
 * Return list of active server node names in the cluster.
 * 
 * @return	list of active node names
 */
public final List<String> getNodeNames() {
	Node[] nodes = cluster.getNodes();		
	ArrayList<String> names = new ArrayList<String>(nodes.length);
	
	for (Node node : nodes) {
		names.add(node.getName());
	}
	return names;
}
 

import com.aerospike.client.cluster.Node; //导入依赖的package包/类
/**
 * Query all nodes for task completion status.
 */
@Override
public boolean isDone() throws AerospikeException {
   if(cluster == null) return true;
	String command = "sindex/" + namespace + '/' + indexName;
	Node[] nodes = cluster.getNodes();
	boolean complete = false;

	for (Node node : nodes) {
		try {
			String response = Info.request(node, command);
			String find = "load_pct=";
			int index = response.indexOf(find);

			if (index < 0) {
				complete = true;
				continue;
			}

			int begin = index + find.length();
			int end = response.indexOf(';', begin);
			String str = response.substring(begin, end);
			int pct = Integer.parseInt(str);
			
			if (pct >= 0 && pct < 100) {
				return false;
			}
			complete = true;
		}
		catch (Exception e) {
			complete = true;
		}
	}
	return complete;
}
 

import com.aerospike.client.cluster.Node; //导入依赖的package包/类
public ScanCommand(
	Node node,
	ScanPolicy policy,
	String namespace,
	String setName,
	ScanCallback callback,
	String[] binNames
) {
	super(node);
	this.policy = policy;
	this.namespace = namespace;
	this.setName = setName;
	this.callback = callback;
	this.binNames = binNames;
}
 

import com.aerospike.client.cluster.Node; //导入依赖的package包/类
public ScanExecutor(Cluster cluster, Node[] nodes, ScanPolicy policy, String namespace, String setName, ScanCallback callback, String[] binNames) {
	this.completedCount = new AtomicInteger();
	this.threadPool = cluster.getThreadPool();
	
	// Initialize threads.		
	threads = new ScanThread[nodes.length];
	
	for (int i = 0; i < nodes.length; i++) {
		ScanCommand command = new ScanCommand(nodes[i], policy, namespace, setName, callback, binNames);
		threads[i] = new ScanThread(command);
	}
	
	// Initialize maximum number of nodes to query in parallel.
	maxConcurrentNodes = (policy.maxConcurrentNodes == 0 || policy.maxConcurrentNodes >= threads.length)? threads.length : policy.maxConcurrentNodes;
}
 

import com.aerospike.client.cluster.Node; //导入依赖的package包/类
public static List<BatchNode> generateList(Cluster cluster, Key[] keys) throws AerospikeException {
	Node[] nodes = cluster.getNodes();
	
	if (nodes.length == 0) {
		throw new AerospikeException(ResultCode.SERVER_NOT_AVAILABLE, "Command failed because cluster is empty.");
	}
	
	int nodeCount = nodes.length;
	int keysPerNode = keys.length / nodeCount + 10;

	// Split keys by server node.
	List<BatchNode> batchNodes = new ArrayList<BatchNode>(nodeCount+1);
			
	for (int i = 0; i < keys.length; i++) {
		Key key = keys[i];
		Partition partition = new Partition(key);			
		BatchNode batchNode;
		
		Node node = cluster.getNode(partition);
		batchNode = findBatchNode(batchNodes, node);
		
		if (batchNode == null) {
			batchNodes.add(new BatchNode(node, keysPerNode, key));
		}
		else {
			batchNode.addKey(key);
		}
	}
	return batchNodes;
}