import com.aerospike.client.Host; //导入依赖的package包/类
/**
 * initialize and validate configuration options
 *
 * @param context
 * @param issues
 */
public void init(Target.Context context, List<Target.ConfigIssue> issues) {
  List<Host> hosts = getAerospikeHosts(issues, connectionString, Groups.AEROSPIKE.getLabel(), "aerospikeBeanConfig.connectionString", context);
  ClientPolicy cp = new ClientPolicy();
  try {
    client = new AerospikeClient(cp, hosts.toArray(new Host[hosts.size()]));
    int retries = 0;
    while (!client.isConnected() && retries <= maxRetries) {
      if (retries > maxRetries) {
        issues.add(context.createConfigIssue(Groups.AEROSPIKE.getLabel(), "aerospikeBeanConfig.connectionString", AerospikeErrors.AEROSPIKE_03, connectionString));
        return;
      }
      retries++;
      try {
        Thread.sleep(100);
      } catch (InterruptedException ignored) {
      }
    }

  } catch (AerospikeException ex) {
    issues.add(context.createConfigIssue(Groups.AEROSPIKE.getLabel(), "aerospikeBeanConfig.connectionString", AerospikeErrors.AEROSPIKE_03, connectionString));
  }
}
 

import com.aerospike.client.Host; //导入依赖的package包/类
/**
 * Request current status from server node.
 *  
 * @param friends		other nodes in the cluster, populated by this method
 * @throws Exception	if status request fails
 */
public final void refresh(List<Host> friends) throws Exception {
	Connection conn = getConnection(1000);
	
	try {
		HashMap<String,String> infoMap = Info.request(conn, "node", "partition-generation", "services");
		verifyNodeName(infoMap);			
		restoreHealth();
		responded = true;
		addFriends(infoMap, friends);
		updatePartitions(conn, infoMap);
		putConnection(conn);
	}
	catch (Exception e) {
		conn.close();
		decreaseHealth();
		throw e;
	}
}
 

import com.aerospike.client.Host; //导入依赖的package包/类
public final void addSeeds(Host[] hosts) {
	// Use copy on write semantics.
	Host[] seedArray = new Host[seeds.length + hosts.length];
	int count = 0;
	
	// Add existing seeds.
	for (Host seed : seeds) {
		seedArray[count++] = seed;
	}
	
	// Add new seeds
	for (Host host : hosts) {
		if (Log.debugEnabled()) {
			Log.debug("Add seed " + host);
		}
		seedArray[count++] = host;
	}
	
	// Replace nodes with copy.
	seeds = seedArray;
}
 

import com.aerospike.client.Host; //导入依赖的package包/类
private final void removeNodes(List<Node> nodesToRemove) {
	// There is no need to delete nodes from partitionWriteMap because the nodes 
	// have already been set to inactive. Further connection requests will result 
	// in an exception and a different node will be tried.
	
	// Cleanup node resources.
	for (Node node : nodesToRemove) {			
		// Remove node's aliases from cluster alias set.
		// Aliases are only used in tend thread, so synchronization is not necessary.
		for (Host alias : node.getAliases()) {
			// Log.debug("Remove alias " + alias);
			aliases.remove(alias);
		}
		node.close();
	}

	// Remove all nodes at once to avoid copying entire array multiple times.
	removeNodesCopy(nodesToRemove);
}
 

import com.aerospike.client.Host; //导入依赖的package包/类
public AsyncCluster(AsyncClientPolicy policy, Host[] hosts) throws AerospikeException {
	super(policy, hosts);
	maxCommands = policy.asyncMaxCommands;
	
	switch (policy.asyncMaxCommandAction) {
	case ACCEPT:
		bufferQueue = new AcceptBufferQueue();
		break;
		
	case REJECT:
		bufferQueue = new RejectBufferQueue(maxCommands);
		break;
		
	case BLOCK:
	default:
		bufferQueue = new BlockBufferQueue(maxCommands);
		break;
	}
	
	selectorManagers = new SelectorManagers(policy);
	initTendThread();
}
 

import com.aerospike.client.Host; //导入依赖的package包/类
private List<Host> getAliases(Host host) {
	InetAddress[] addresses;
	
	try {
		addresses = InetAddress.getAllByName(host.name);
	}
	catch (UnknownHostException uhe) {
		throw new AerospikeException.Connection("Invalid host: " + host);
	}
		
	if (addresses.length == 0) {
		throw new AerospikeException.Connection("Failed to find addresses for " + host);
	}
	
	// Add capacity for current address aliases plus IPV6 address and hostname.
	List<Host> aliases = new ArrayList<Host>(addresses.length + 2);
	
	for (InetAddress address : addresses) {
		aliases.add(new Host(address.getHostAddress(), host.tlsName, host.port));
	}
	
	return aliases;
}
 

import com.aerospike.client.Host; //导入依赖的package包/类
/**
	 * This method brings Spikefy service up to speed
	 * ... can be called multiple times
	 * ... but will execute only once (and once it should be enough)
	 */
	public Spikeify getSfy() {

		if (SpikeifyService.getClient() == null) {
			log.info("Starting Aerospike initialization...");
			Map<String, Integer> hosts;

			log.info("--== LOCALHOST ==--");
			hosts = new HashMap<>();
			hosts.put(DEFAULT_HOST, DEFAULT_PORT);

			String namespace = DEFAULT_NAMESPACE;
			log.info("Aerospike default namespace: " + namespace);

			List<Host> hostsData = hosts.entrySet().stream().map(stringIntegerEntry -> new Host(stringIntegerEntry.getKey(), stringIntegerEntry.getValue())).collect(Collectors.toList());
			SpikeifyService.globalConfig(namespace, hostsData.toArray(new Host[hostsData.size()]));

			log.info("Aerospike configured.");

			log.info("Creating indexes...");

			SpikeifyService.register(FileSegment.class);

			/*SpikeifyService.getClient().createIndex(new Policy(), namespace, File.class.getSimpleName(), "fileName", "fileName", IndexType.STRING);
			SpikeifyService.getClient().createIndex(new Policy(), namespace, FileLock.class.getSimpleName(), "lockName", "lockName", IndexType.STRING);
*/
			log.info("End of creating indexes.");

		}

		return SpikeifyService.sfy();
	}
 

import com.aerospike.client.Host; //导入依赖的package包/类
public List<Host> validateConnectionString(
    List<Stage.ConfigIssue> issues,
    String connectionString,
    String configGroupName,
    String configName,
    Stage.Context context
) {
  List<Host> clusterNodesList = new ArrayList<>();
  if (connectionString == null || connectionString.isEmpty()) {
    issues.add(context.createConfigIssue(configGroupName, configName,
        AerospikeErrors.AEROSPIKE_01, configName));
  } else {
    String[] nodes = connectionString.split(",");
    for (String node : nodes) {
      try {
        HostAndPort hostAndPort = HostAndPort.fromString(node);
        if(!hostAndPort.hasPort() || hostAndPort.getPort() < 0) {
          issues.add(context.createConfigIssue(configGroupName, configName, AerospikeErrors.AEROSPIKE_02, connectionString));
        } else {
          clusterNodesList.add(new Host(hostAndPort.getHostText(), hostAndPort.getPort()));
        }
      } catch (IllegalArgumentException e) {
        issues.add(context.createConfigIssue(configGroupName, configName, AerospikeErrors.AEROSPIKE_02, connectionString));
      }
    }
  }
  return clusterNodesList;
}
 

import com.aerospike.client.Host; //导入依赖的package包/类
public static List<Host> getAerospikeHosts(
    List<Target.ConfigIssue> issues,
    List<String> connectionString,
    String configGroupName,
    String configName,
    Target.Context context
) {
  List<Host> clusterNodesList = new ArrayList<>();
  if (connectionString == null || connectionString.isEmpty()) {
    issues.add(context.createConfigIssue(configGroupName, configName,
        AerospikeErrors.AEROSPIKE_01, configName));
  } else {
    for (String node : connectionString) {
      try {
        HostAndPort hostAndPort = HostAndPort.fromString(node);
        if (!hostAndPort.hasPort() || hostAndPort.getPort() < 0) {
          issues.add(context.createConfigIssue(configGroupName, configName, AerospikeErrors.AEROSPIKE_02, connectionString));
        } else {
          clusterNodesList.add(new Host(hostAndPort.getHostText(), hostAndPort.getPort()));
        }
      } catch (IllegalArgumentException e) {
        issues.add(context.createConfigIssue(configGroupName, configName, AerospikeErrors.AEROSPIKE_02, connectionString));
      }
    }
  }
  return clusterNodesList;
}
 

import com.aerospike.client.Host; //导入依赖的package包/类
public Host[] getAerospikeHosts() {
	String[] hostsCluster = readTextProperty(AEROSPIKE_HOSTS).split(",");
	Host[] hosts = new Host[hostsCluster.length];
	for (int i = 0; i < hostsCluster.length; i++) {
		String[] nodeConfig = hostsCluster[i].split(":");
		hosts[i] = new Host(nodeConfig[0], Integer.parseInt(nodeConfig[1]));
	}
	return hosts;
}
 

import com.aerospike.client.Host; //导入依赖的package包/类
private void setAliases(Host host) throws AerospikeException {	
	try {
		InetAddress[] addresses = InetAddress.getAllByName(host.name);
		int count = 0;
		aliases = new Host[addresses.length];
		
		for (InetAddress address : addresses) {
			aliases[count++] = new Host(address.getHostAddress(), host.port);
		}
	}
	catch (UnknownHostException uhe) {
		throw new AerospikeException.Connection("Invalid host: " + host);
	}
}
 

import com.aerospike.client.Host; //导入依赖的package包/类
private final void addFriends(HashMap <String,String> infoMap, List<Host> friends) throws AerospikeException {
	// Parse the service addresses and add the friends to the list.
	String friendString = infoMap.get("services");
	
	if (friendString == null || friendString.length() == 0) {
		return;
	}

	String friendNames[] = friendString.split(";");

	for (String friend : friendNames) {
		String friendInfo[] = friend.split(":");
		String host = friendInfo[0];
		int port = Integer.parseInt(friendInfo[1]);
		Host alias = new Host(host, port);
		Node node = cluster.findAlias(alias);
		
		if (node != null) {
			node.referenceCount++;
		}
		else {
			if (! findAlias(friends, alias)) {
				friends.add(alias);					
			}
		}
	}
}
 

import com.aerospike.client.Host; //导入依赖的package包/类
private final static boolean findAlias(List<Host> friends, Host alias) {
	for (Host host : friends) {
		if (host.equals(alias)) {
			return true;
		}
	}
	return false;
}
 

import com.aerospike.client.Host; //导入依赖的package包/类
/**
 * Add node alias to list.
 */
public final void addAlias(Host aliasToAdd) {
	// Aliases are only referenced in the cluster tend thread,
	// so synchronization is not necessary.
	Host[] tmpAliases = new Host[aliases.length + 1];
	int count = 0;
	
	for (Host host : aliases) {
		tmpAliases[count++] = host;
	}
	tmpAliases[count] = aliasToAdd;
	aliases = tmpAliases;
}
 

import com.aerospike.client.Host; //导入依赖的package包/类
public Cluster(ClientPolicy policy, Host[] hosts) throws AerospikeException {
	this.seeds = hosts;
	connectionQueueSize = policy.maxThreads + 1;  // Add one connection for tend thread.
	connectionTimeout = policy.timeout;
	maxSocketIdle = policy.maxSocketIdle;
	
	if (policy.threadPool == null) {
		// Create cached thread pool with daemon threads.
		// Daemon threads automatically terminate when the program terminates.
		threadPool = Executors.newCachedThreadPool(new ThreadFactory() {
			public final Thread newThread(Runnable runnable) {
				Thread thread = new Thread(runnable);
				thread.setDaemon(true);
				return thread;
			}
		});
	}
	else {
		threadPool = policy.threadPool;
	}
	sharedThreadPool = policy.sharedThreadPool;
	
	aliases = new HashMap<Host,Node>();
	nodes = new Node[0];	
	partitionWriteMap = new HashMap<String,AtomicReferenceArray<Node>>();		
	nodeIndex = new AtomicInteger();
}
 

import com.aerospike.client.Host; //导入依赖的package包/类
private final boolean findSeed(Host search) {
	for (Host seed : seeds) {
		if (seed.equals(search)) {
			return true;
		}
	}
	return false;
}
 

import com.aerospike.client.Host; //导入依赖的package包/类
private final ArrayList<Node> findNodesToAdd(List<Host> hosts) {
	ArrayList<Node> list = new ArrayList<Node>(hosts.size());
	
	for (Host host : hosts) {
		try {
			NodeValidator nv = new NodeValidator(host, connectionTimeout);
			Node node = findNode(nv.name);
			
			if (node != null) {
				// Duplicate node name found.  This usually occurs when the server 
				// services list contains both internal and external IP addresses 
				// for the same node.  Add new host to list of alias filters
				// and do not add new node.
				node.referenceCount++;
				node.addAlias(host);
				aliases.put(host, node);
				continue;
			}
			node = createNode(nv);		
			list.add(node);
		}
		catch (Exception e) {
			if (Log.warnEnabled()) {
				Log.warn("Add node " + host + " failed: " + Util.getErrorMessage(e));
			}
		}
	}
	return list;
}
 

import com.aerospike.client.Host; //导入依赖的package包/类
private final void addAliases(Node node) {		
	// Add node's aliases to global alias set.
	// Aliases are only used in tend thread, so synchronization is not necessary.
	for (Host alias : node.getAliases()) {
		aliases.put(alias, node);
	}
}
 

import com.aerospike.client.Host; //导入依赖的package包/类
public NodeValidator(Host host, int timeoutMillis) throws AerospikeException {
	setAliases(host);
	setAddress(timeoutMillis);
}
 

import com.aerospike.client.Host; //导入依赖的package包/类
/**
 * Return server node IP address and port.
 */
public final Host getHost() {
	return host;
}