import com.couchbase.client.CouchbaseConnectionFactoryBuilder; //导入依赖的package包/类
@Override
public void connect() throws IOException
{
  String[] tokens = uriString.split(",");
  URI uri = null;
  for (String url : tokens) {
    try {
      uri = new URI("http", url, "/pools", null, null);
    } catch (URISyntaxException ex) {
      DTThrowable.rethrow(ex);
    }
    baseURIs.add(uri);
  }
  try {
    CouchbaseConnectionFactoryBuilder cfb = new CouchbaseConnectionFactoryBuilder();
    cfb.setOpTimeout(timeout);  // wait up to 10 seconds for an operation to succeed
    cfb.setOpQueueMaxBlockTime(blockTime); // wait up to 10 second when trying to enqueue an operation
    client = new CouchbaseClient(cfb.buildCouchbaseConnection(baseURIs, bucket, password));
    //client = new CouchbaseClient(baseURIs, "default", "");
  } catch (IOException e) {
    logger.error("Error connecting to Couchbase:", e);
    DTThrowable.rethrow(e);
  }
}
 

import com.couchbase.client.CouchbaseConnectionFactoryBuilder; //导入依赖的package包/类
/**
     * @param nodes
     * @param bucketName
     * @param password
     * @param timeoutMillis
     */
    public CouchbaseClientFactory(String nodes, String bucketName, String password, Integer timeoutMillis) {
        info = "CouchbaseClientFactory{" + nodes + ", " + bucketName + ", " + password + "}";
        log.info(info);
        ArrayList<URI> nodesList = new ArrayList<URI>();
        for (String node : nodes.split(",")) {
            nodesList.add(URI.create("http://" + node + ":8091/pools"));
        }

        CouchbaseConnectionFactoryBuilder builder = new CouchbaseConnectionFactoryBuilder();
        builder.setOpTimeout(timeoutMillis);//2.5 by default
//        builder.setObsTimeout(timeoutMillis);
        try {
            client = new CouchbaseClient(builder.buildCouchbaseConnection(nodesList, bucketName, password));
        } catch (Exception e) {
            log.error("Error connecting to Couchbase: ", e);
            throw new RuntimeException(e);
        }
    }
 

import com.couchbase.client.CouchbaseConnectionFactoryBuilder; //导入依赖的package包/类
/**
 * Method to instance the couchbase client that will interact with the server
 */
protected void createClient() {
  try {
    CouchbaseConnectionFactoryBuilder builder = new CouchbaseConnectionFactoryBuilder();
    builder.setOpTimeout(operationTimeout);
    builder.setViewTimeout(viewTimeout);
    builder.setObsTimeout(observerTimeout);
    builder.setViewConnsPerNode(viewConnsPerNode);
    // normally the worker size should be equal to the number of processors
    // org.apache.http.impl.nio.reactor.IOReactorConfig.AVAIL_PROCS
    builder.setViewWorkerSize(Runtime.getRuntime().availableProcessors());

    client = new CouchbaseClient(builder.buildCouchbaseConnection(bootstrapUris(), bucketName, bucketPassword));
  } catch (Exception e) {
    logger.error("Failed to connect to couchbase server", e);
    throw new RuntimeException(e);
  }
}
 

import com.couchbase.client.CouchbaseConnectionFactoryBuilder; //导入依赖的package包/类
@Override
public Object getObject() throws IOException {
    List<URI> addresses = getAddresses(connectionURI);
    long timeout = Math.max(readTimeout, writeTimeout);
    //TODO make all the below properties configurable via properties file
    ConnectionFactoryBuilder builder = new CouchbaseConnectionFactoryBuilder()
            .setOpTimeout(timeout)                      // wait up to timeout seconds for an operation to succeed
            .setOpQueueMaxBlockTime(enqueueTimeout)     // wait up to 'enqueueTimeout' seconds when trying to enqueue an operation
            .setDaemon(true)
            .setProtocol(Protocol.BINARY)
            .setHashAlg(DefaultHashAlgorithm.KETAMA_HASH)
            .setFailureMode(FailureMode.Redistribute)
            .setInitialObservers(Collections.singleton((ConnectionObserver) new CouchbaseAlerter()));
    if(transcoder!=null) {
        builder.setTranscoder(transcoder);
    }

    //assuming there isn't any password set for Couchbase
    CouchbaseConnectionFactory connectionFactory
        = ((CouchbaseConnectionFactoryBuilder)builder).buildCouchbaseConnection(addresses, "default", "", "");

    return new CouchbaseClient(connectionFactory);
}
 

import com.couchbase.client.CouchbaseConnectionFactoryBuilder; //导入依赖的package包/类
@Override
public void connect() throws IOException
{
  super.connect();
  logger.debug("connection established");
  try {
    CouchbaseConnectionFactoryBuilder cfb = new CouchbaseConnectionFactoryBuilder();
    cfb.setOpTimeout(timeout);  // wait up to 10 seconds for an operation to succeed
    cfb.setOpQueueMaxBlockTime(blockTime); // wait up to 10 second when trying to enqueue an operation
    clientMeta = new CouchbaseClient(cfb.buildCouchbaseConnection(baseURIs, bucketMeta, passwordMeta));
  } catch (IOException e) {
    logger.error("Error connecting to Couchbase: ", e);
    DTThrowable.rethrow(e);
  }
}
 

import com.couchbase.client.CouchbaseConnectionFactoryBuilder; //导入依赖的package包/类
@Bean(name = "couchbaseClient")
public CouchbaseClient couchbaseClient() throws IOException, URISyntaxException {
    if (couchbaseClientEnabled) {
        logger.info("Creating CouchbaseClient for servers: {}", couchbaseServers);
        CouchbaseConnectionFactoryBuilder builder = new CouchbaseConnectionFactoryBuilder();
        builder.setOpTimeout(opTimeout);
        CouchbaseConnectionFactory cf = builder.buildCouchbaseConnection(getServersList(), couchbaseBucket,
                                                                         StringUtils.trimAllWhitespace(Optional.ofNullable(couchbasePassword)
                                                                                                               .orElse("")));
        return new CouchbaseClient(
                                   cf);
    }
    return null;

}
 

import com.couchbase.client.CouchbaseConnectionFactoryBuilder; //导入依赖的package包/类
public CouchbaseClientFactory(String bucket, List<URI> nodes, long opTimeout, String instanceName) {
    this.builder = new CouchbaseConnectionFactoryBuilder();
    this.registeredClients = new ArrayList<CouchbaseClient>();
    this.bucket = bucket;
    this.nodes = nodes;
    this.opTimeout = opTimeout;
    this.instanceName = instanceName;
}
 

import com.couchbase.client.CouchbaseConnectionFactoryBuilder; //导入依赖的package包/类
public static CouchbaseClient createClient(String bucketName, Transcoder transcoder) throws Exception {
    if (clients == null) {
        clients = new HashMap<>();
    }
    CouchbaseConnectionFactoryBuilder cfb = new CouchbaseConnectionFactoryBuilder();
    if(transcoder!=null) {
        cfb.setTranscoder(transcoder);
    }
    List<URI> uriList = new ArrayList<>();
    uriList.add(new URI("http", null, "localhost", couchbasePort, "/pools", "", ""));
    CouchbaseConnectionFactory connectionFactory = cfb.buildCouchbaseConnection(uriList, bucketName, PASSWORD);
    clients.put(bucketName, new CouchbaseClient(connectionFactory));

    return clients.get(bucketName);
}
 

import com.couchbase.client.CouchbaseConnectionFactoryBuilder; //导入依赖的package包/类
public MemcachedClientIF getObject() throws Exception {
try {
    CouchbaseConnectionFactoryBuilder connectionFactoryBuilder = new CouchbaseConnectionFactoryBuilder();
    connectionFactoryBuilder
	    .setFailureMode(CouchbaseConnectionFactory.DEFAULT_FAILURE_MODE);
    connectionFactoryBuilder
	    .setHashAlg(CouchbaseConnectionFactory.DEFAULT_HASH);

    connectionFactoryBuilder.setOpQueueFactory(new CustomQueueFactory(
	    queueSize));

    if (opQueueMaxBlockTime > 0) {
	connectionFactoryBuilder
		.setOpQueueMaxBlockTime(opQueueMaxBlockTime);
    }
    if (operationTimeout > 0) {
	connectionFactoryBuilder.setOpTimeout(operationTimeout);
    }
    if (baseList.size() > 0) {
	return getCouchBaseConnection(connectionFactoryBuilder);
    } else if (StringUtils.isNotBlank(servers)) {
	return getPlainMemcachedConnection(connectionFactoryBuilder);
    } else {
	return createEmptyConnection();
    }
} catch (NullPointerException npe) {
    log.error("Can't connect to membase baseList:" + baseList
	    + ", bucketName:" + bucketName);

    throw new RuntimeException(
	    "Can't create memcache client, check your properties", npe);
}

   }
 

import com.couchbase.client.CouchbaseConnectionFactoryBuilder; //导入依赖的package包/类
@Override
   protected CouchbaseClient getCouchBaseConnection(
    CouchbaseConnectionFactoryBuilder connectionFactoryBuilder)
    throws IOException {

ConnectionFactory factory = connectionFactoryBuilder.build();
Assert.assertEquals(factory.getOperationTimeout(), 10223);
assertEquals(factory.getOpQueueMaxBlockTime(), 10223);

return Mockito.mock(CouchbaseClient.class);
   }
 

import com.couchbase.client.CouchbaseConnectionFactoryBuilder; //导入依赖的package包/类
/**
 * Creates and returns {@link CouchbaseClient} based on the parameters provided.
 * @param urls
 * @param bucketName
 * @param password
 * @return {@link CouchbaseClient} instance.
 * @throws IOException
 */
public static CouchbaseClient getCouchBaseClient(String[] urls, String bucketName, String password) throws IOException {

    List<URI> uris = new ArrayList<URI>();
    for (String url: urls){
        uris.add(URI.create(url));
    }

    CouchbaseConnectionFactoryBuilder builder = new CouchbaseConnectionFactoryBuilder();
    CouchbaseClient couchbaseClient = new CouchbaseClient(builder.buildCouchbaseConnection(uris,bucketName,password));
    return couchbaseClient;
}
 

import com.couchbase.client.CouchbaseConnectionFactoryBuilder; //导入依赖的package包/类
protected void init() throws Exception {

        if (BucketConfig.instance == null){
            new BucketConfig().afterPropertiesSet();
        }
        this.config = BucketConfig.instance;

		initMetrics();
		
		this.buckets = new HashMap<String, CouchbaseClient>();
		this.entityBuckets = new HashMap<String, String>();
		
		List<Map> items = (List<Map>) this.config.getBuckets();

		CouchbaseConnectionFactoryBuilder builder = new CouchbaseConnectionFactoryBuilder();
		builder.setDaemon(true);
		builder.setEnableMetrics(MetricType.PERFORMANCE);
		builder.setFailureMode(FailureMode.Redistribute);
		
		for (Map item : items) {
			String name = Objects.toString(item.get("bucket"), "").trim();
			String user = Objects.toString(item.get("user"), "");
			String pwd = Objects.toString(item.get("pwd"), "");
			String urls = Objects.toString(item.get("urls"), "");
			String[] ents = Objects.toString(item.get("ents"), "").split(",");
            logger.info("@Couchbase connecting to bucket: {}, urls: {}", name, urls);
			CouchbaseConnectionFactory ccf = null;
            if(StringUtils.isNotBlank(user)){
                ccf = builder.buildCouchbaseConnection(this.wrapBaseList(urls), name, user, pwd);
            }else{
                ccf = builder.buildCouchbaseConnection(this.wrapBaseList(urls), name, pwd);
            }
			this.buckets.put(name, new CouchbaseClient(ccf));
			this.defaultBucket = name;
			for(String str : ents){
				this.entityBuckets.put(str.toLowerCase().trim(), name);
			}
			logger.info("@Couchbase init, bucket:{}, urls:{}", name, urls);
		}
	}
 

import com.couchbase.client.CouchbaseConnectionFactoryBuilder; //导入依赖的package包/类
protected void setBuilder(CouchbaseConnectionFactoryBuilder builder) {
    this.builder = builder;
}
 

import com.couchbase.client.CouchbaseConnectionFactoryBuilder; //导入依赖的package包/类
protected MemcachedClient getPlainMemcachedConnection(
    CouchbaseConnectionFactoryBuilder connectionFactoryBuilder)
    throws IOException {
return new MemcachedClient(connectionFactoryBuilder.setOpFact(
	getOperationFactory()).build(), AddrUtil.getAddresses(servers));
   }
 

import com.couchbase.client.CouchbaseConnectionFactoryBuilder; //导入依赖的package包/类
@Override
   protected MemcachedClient getPlainMemcachedConnection(
    CouchbaseConnectionFactoryBuilder connectionFactoryBuilder)
    throws IOException {
return Mockito.mock(MemcachedClient.class);
   }