import com.microsoft.azure.documentdb.DocumentClient; //导入依赖的package包/类
public static Database GetDatabase(DocumentClient client, String databaseId) {
    BackoffExponentialRetryPolicy retryPolicy = new BackoffExponentialRetryPolicy();
    QueryIterable<Database> dbIterable = client.queryDatabases(
            new SqlQuerySpec("SELECT * FROM root r WHERE [email protected]",
                    new SqlParameterCollection(new SqlParameter("@id", databaseId))),
            null).getQueryIterable();
    
    List<Database> databases = null;
    while(retryPolicy.shouldRetry()){
        try {
            databases = dbIterable.toList();
            break;
        } catch (Exception e) {
            retryPolicy.errorOccured(e);
        }
    }
    
    if(databases.size() == 0) {
        return null;
    }
    
    return databases.get(0);
}
 

import com.microsoft.azure.documentdb.DocumentClient; //导入依赖的package包/类
public AsyncDocumentClient build() {
    
    ifThrowIllegalArgException(this.serviceEndpoint == null, "cannot build client without service endpoint");
    ifThrowIllegalArgException(this.masterKey == null, "cannot build client without masterKey");

    if (connectionPolicy != null && ConnectionMode.DirectHttps != connectionPolicy.getConnectionMode()) {
        return new RxDocumentClientImpl(serviceEndpoint, masterKey, connectionPolicy, desiredConsistencyLevel,
                eventLoopSize, separateComputationPoolSize);
    } else {
        
        ifThrowIllegalArgException(this.eventLoopSize != -1, "eventLoopSize is not applicable in direct mode");
        ifThrowIllegalArgException(this.separateComputationPoolSize != -1, "separateComputationPoolSize is not applicable in direct mode");
        
        // fall back to RX wrapper with blocking IO
        return new RxWrapperDocumentClientImpl(
                new DocumentClient(serviceEndpoint.toString(), masterKey, connectionPolicy, desiredConsistencyLevel));
    }
}
 

import com.microsoft.azure.documentdb.DocumentClient; //导入依赖的package包/类
@Before
public void setup() {
    mappingContext = new DocumentDbMappingContext();
    try {
        mappingContext.setInitialEntitySet(new EntityScanner(this.applicationContext)
                .scan(Persistent.class));
    } catch (ClassNotFoundException e) {
        throw new RuntimeException(e.getMessage());

    }
    dbConverter = new MappingDocumentDbConverter(mappingContext);
    documentClient = new DocumentClient(documentDbUri, documentDbKey,
            ConnectionPolicy.GetDefault(), ConsistencyLevel.Session);

    dbTemplate = new DocumentDbTemplate(documentClient, dbConverter, TEST_DB_NAME);

    dbTemplate.createCollectionIfNotExists(Person.class.getSimpleName(), null, null);
    dbTemplate.insert(Person.class.getSimpleName(), TEST_PERSON, null);
}
 

import com.microsoft.azure.documentdb.DocumentClient; //导入依赖的package包/类
private DocumentClient createDocumentClient() {
    LOG.debug("createDocumentClient");
    final ConnectionPolicy policy = connectionPolicy == null ? ConnectionPolicy.GetDefault() : connectionPolicy;

    String userAgent = (policy.getUserAgentSuffix() == null ? "" : ";" + policy.getUserAgentSuffix()) +
            ";" + USER_AGENT_SUFFIX;

    if (properties.isAllowTelemetry() && GetHashMac.getHashMac() != null) {
        userAgent += ";" + GetHashMac.getHashMac();
    }
    policy.setUserAgentSuffix(userAgent);

    return new DocumentClient(properties.getUri(), properties.getKey(), policy,
            properties.getConsistencyLevel() == null ?
                    ConsistencyLevel.Session : properties.getConsistencyLevel());
}
 

import com.microsoft.azure.documentdb.DocumentClient; //导入依赖的package包/类
@Test
public void canSetAllowTelemetryFalse() {
    PropertySettingUtil.setAllowTelemetryFalse();

    try (AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext()) {
        context.register(DocumentDBAutoConfiguration.class, ConnectionPolicyConfig.class);
        context.refresh();
        final DocumentClient documentClient = context.getBean(DocumentClient.class);

        final ConnectionPolicy connectionPolicy = documentClient.getConnectionPolicy();
        assertThat(connectionPolicy.getUserAgentSuffix()).contains(PropertySettingUtil.USER_AGENT_SUFFIX);
        assertThat(connectionPolicy.getUserAgentSuffix()).contains(
                PropertySettingUtil.DEFAULT_USER_AGENT_SUFFIX);
    }
    PropertySettingUtil.unsetAllowTelemetry();
}
 

import com.microsoft.azure.documentdb.DocumentClient; //导入依赖的package包/类
/**
 * Initializes a new instance of {@link DocumentBulkImporter}
 *
 * @param client {@link DocumentClient} instance to use
 * @param collectionLink inserts documents to the specified collection
 * @param maxMiniBatchSize
 * @throws DocumentClientException if any failure
 */
private DocumentBulkImporter(DocumentClient client, 
        String collectionLink,
        PartitionKeyDefinition partitionKeyDefinition,
        int collectionOfferThroughput) {
    Preconditions.checkNotNull(client, "client cannot be null");
    Preconditions.checkNotNull(partitionKeyDefinition, "partitionKeyDefinition cannot be null");
    Preconditions.checkNotNull(collectionLink, "collectionLink cannot be null");
    Preconditions.checkArgument(collectionOfferThroughput > 0, "collection throughput is less than 10,000");

    this.client = client;
    this.collectionLink = collectionLink;
    this.collectionThroughput =  collectionOfferThroughput;
    this.partitionKeyDefinition = partitionKeyDefinition;
    this.listeningExecutorService = MoreExecutors.listeningDecorator(Executors.newCachedThreadPool());
}
 

import com.microsoft.azure.documentdb.DocumentClient; //导入依赖的package包/类
private CollectionRoutingMap getCollectionRoutingMap(DocumentClient client) {
    List<ImmutablePair<PartitionKeyRange, Boolean>> ranges = new ArrayList<>();

    for (PartitionKeyRange range : client.readPartitionKeyRanges(this.collectionLink, (FeedOptions) null).getQueryIterable().toList()) {
        ranges.add(new ImmutablePair<>(range, true));
    }

    CollectionRoutingMap routingMap = InMemoryCollectionRoutingMap.tryCreateCompleteRoutingMap(ranges,
            StringUtils.EMPTY);

    if (routingMap == null) {
        throw new IllegalStateException("Cannot create complete routing map");
    }

    return routingMap;
}
 

import com.microsoft.azure.documentdb.DocumentClient; //导入依赖的package包/类
@Test
public void callbackCount() {
    BulkImportStoredProcedureOptions options = null;
    String bulkImportSproc = null;
    DocumentClient client = Mockito.mock(DocumentClient.class);
    List<List<String>> batchesToInsert = new ArrayList<>();
    batchesToInsert.add(new ArrayList<>());
    batchesToInsert.add(new ArrayList<>());
    batchesToInsert.add(new ArrayList<>());

    String partitionIndex = "0";
    BatchInserter bi = new BatchInserter(partitionIndex, batchesToInsert, client, bulkImportSproc, options);

    Iterator<Callable<InsertMetrics>> callbackIterator = bi.miniBatchInsertExecutionCallableIterator();

    List<Callable<InsertMetrics>> list = new ArrayList<>();
    Iterators.addAll(list, callbackIterator);

    assertThat(list.size(), equalTo(3));
}
 

import com.microsoft.azure.documentdb.DocumentClient; //导入依赖的package包/类
public void setup(DocumentClient client) throws Exception {
    logger.debug("setting up ...");
    cleanUpDatabase(client);

    Thread.sleep(5000);
    Database d = new Database();
    d.setId(databaseId);
    Database database = client.createDatabase(d, null).getResource();


    PartitionKeyDefinition partitionKeyDef = new PartitionKeyDefinition();
    ArrayList<String> paths = new ArrayList<String>();
    paths.add("/mypk");
    partitionKeyDef.setPaths(paths);

    pCollection = new DocumentCollection();
    pCollection.setId(collectionId);
    pCollection.setPartitionKey(partitionKeyDef);

    RequestOptions options = new RequestOptions();
    options.setOfferThroughput(10100);
    pCollection = client.createCollection(database.getSelfLink(), pCollection, options).getResource();
    logger.debug("setting up done.");
}
 

import com.microsoft.azure.documentdb.DocumentClient; //导入依赖的package包/类
@Before
public void setUp() throws DocumentClientException {
    // create client
    client = new DocumentClient(AccountCredentials.HOST, AccountCredentials.MASTER_KEY, null, null);

    // removes the database "exampleDB" (if exists)
    deleteDatabase();
    // create database exampleDB;
    createDatabase();

    // create collection
    createMultiPartitionCollection();

    // populate documents
    populateDocuments();
}
 

import com.microsoft.azure.documentdb.DocumentClient; //导入依赖的package包/类
/**
 * Creates a document and replaces it if it already exists when isUpsert is true. The function also retries on throttling 
 * @param client The DocumentClient instance.
 * @param collectionSelfLink The self link of the passed collection.
 * @param isUpsert Specify if the document should be upserted.
 */
public static Document createDocument(DocumentClient client, String collectionSelfLink, Document doc, boolean isUpsert) {
    BackoffExponentialRetryPolicy retryPolicy = new BackoffExponentialRetryPolicy();
    while(retryPolicy.shouldRetry()){
    	try {
     	if(isUpsert) {
     		return client.upsertDocument(collectionSelfLink, doc, null, false).getResource();
     	} else {
     		return client.createDocument(collectionSelfLink, doc, null, false).getResource();	
     	}
    	} catch(DocumentClientException e){
    		retryPolicy.errorOccured(e);
    	}            
    }
    
    return null;
}
 

import com.microsoft.azure.documentdb.DocumentClient; //导入依赖的package包/类
/**
 * Gets an output collection with the passed name ( if the collection already exists return it, otherwise create new one
 * @param client The DocumentClient instance.
 * @param databaseSelfLink the self link of the passed database.
 * @param collectionId The id of the output collection.
 */
public static DocumentCollection GetDocumentCollection(DocumentClient client, String databaseSelfLink, String collectionId) {
    BackoffExponentialRetryPolicy retryPolicy = new BackoffExponentialRetryPolicy();
    QueryIterable<DocumentCollection> collIterable = client.queryCollections(
            databaseSelfLink,
            new SqlQuerySpec("SELECT * FROM root r WHERE [email protected]",
                    new SqlParameterCollection(new SqlParameter("@id", collectionId))),
            null).getQueryIterable();
    
    List<DocumentCollection> collections = null;
    while(retryPolicy.shouldRetry()){
        try {
            collections = collIterable.toList();
            break;
        } catch (Exception e) {
            retryPolicy.errorOccured(e);
        }
    }
    
    if(collections.size() == 0) {
        return null;
    }
    
    return collections.get(0);
}
 

import com.microsoft.azure.documentdb.DocumentClient; //导入依赖的package包/类
/**
 * Executes the bulk import stored procedure for a list of documents.
 * The execution takes into consideration throttling and blacklisting of the stored procedure.
 * @param client The DocumentClient instance for DocumentDB
 * @param collectionSelfLink the self-link for the collection to write to.
 * @param sproc The stored procedure to execute
 * @param allDocs The list of documents to write
 * @param upsert  Specifies whether to replace the document if exists or not. By default it's true.
 */
public static void executeWriteStoredProcedure(final DocumentClient client, String collectionSelfLink, final StoredProcedure sproc,
        List<Document> allDocs, final boolean upsert) {
    
    int currentCount = 0;
        
    while (currentCount < allDocs.size())
        {
        String []jsonArrayString = CreateBulkInsertScriptArguments(allDocs, currentCount, MAX_SCRIPT_SIZE);
        BackoffExponentialRetryPolicy retryPolicy = new BackoffExponentialRetryPolicy();
        String response = null;
        while(retryPolicy.shouldRetry()){
            try {
                response = client.executeStoredProcedure(sproc.getSelfLink(), new Object[] { jsonArrayString, upsert })
                    .getResponseAsString();
                break;
            } catch(Exception e){
                retryPolicy.errorOccured(e);  
            }
        }

        int createdCount = Integer.parseInt(response);
        currentCount += createdCount;
    }
}
 

import com.microsoft.azure.documentdb.DocumentClient; //导入依赖的package包/类
/**
 * Reads the bulk import script body from the file.
 * @param client the DocumentClient instance.
 * @return a string that contains the stored procedure body.
 */
private static String getBulkImportBody(DocumentClient client) {
    try {
        InputStream stream = DocumentDBConnectorUtil.class.getResourceAsStream(BULK_IMPORT_PATH);
        List<String> scriptLines = IOUtils.readLines(stream);
        StringBuilder scriptBody = new StringBuilder();
        for (Iterator<String> iterator = scriptLines.iterator(); iterator.hasNext();) {
            String line = (String) iterator.next();
            scriptBody.append(line + "\n");
        }

        return scriptBody.toString();
    } catch (IOException e) {
        throw new IllegalStateException(e);
    }
}
 

import com.microsoft.azure.documentdb.DocumentClient; //导入依赖的package包/类
public DocumentDBRecordWriter(Configuration conf, String host, String key, String dbName, String[] collNames,
        int outputStringPrecision, boolean upsert, String offerType) throws IOException {
    try {
        ConnectionPolicy policy = ConnectionPolicy.GetDefault();
        policy.setUserAgentSuffix(DocumentDBConnectorUtil.UserAgentSuffix);
        DocumentClient client = new DocumentClient(host, key, policy,
                ConsistencyLevel.Session);

        Database db = DocumentDBConnectorUtil.GetDatabase(client, dbName);
        this.collections = new DocumentCollection[collNames.length];
        this.sprocs = new StoredProcedure[collNames.length];
        for (int i = 0; i < collNames.length; i++) {
            this.collections[i] =  DocumentDBConnectorUtil.getOrCreateOutputCollection(client, db.getSelfLink(), collNames[i],
                    outputStringPrecision, offerType);
            this.sprocs[i] = DocumentDBConnectorUtil.CreateBulkImportStoredProcedure(client, this.collections[i].getSelfLink());
        }
        
        this.client = client;
        this.enableUpsert = upsert;
        this.cachedDocs = new LinkedList<Document>();
    } catch (Exception e) {
        e.printStackTrace();
        throw new IOException(e);
    }
}
 

import com.microsoft.azure.documentdb.DocumentClient; //导入依赖的package包/类
@Override
protected void onInit() throws Exception {
    RxNetty.useMetricListenersFactory(new ServoEventsListenerFactory());
    documentClient = new DocumentClient(cfg.getServiceEndpoint(), cfg.getMasterKey(),
            cfg.getConnectionPolicy(), cfg.getConsistencyLevel());
    
    ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(cfg.getConcurrency(), cfg.getConcurrency(),
            10, TimeUnit.MINUTES, new ArrayBlockingQueue<Runnable>(cfg.getConcurrency(), true),
            new ThreadPoolExecutor.CallerRunsPolicy());

    this.executor = MoreExecutors.listeningDecorator(threadPoolExecutor);
}
 

import com.microsoft.azure.documentdb.DocumentClient; //导入依赖的package包/类
public static SyncBulkInsertBenchmark fromConfiguration(Configuration cfg) throws DocumentClientException {
    try (DocumentClient client = new DocumentClient(cfg.getServiceEndpoint(), cfg.getMasterKey(),
            cfg.getConnectionPolicy(), cfg.getConsistencyLevel())) {

        Database database = DocDBUtils.getDatabase(client, cfg.getDatabaseId());
        DocumentCollection collection = DocDBUtils.getCollection(client, database.getSelfLink(),
                cfg.getCollectionId());
        DatabaseAccount databaseAccount = client.getDatabaseAccount();

        return new SyncBulkInsertBenchmark(cfg, database, collection, databaseAccount);
    }
}
 

import com.microsoft.azure.documentdb.DocumentClient; //导入依赖的package包/类
public static Database getDatabase(DocumentClient client, String databaseId) {
    FeedResponse<Database> feedResponsePages = client
            .queryDatabases(new SqlQuerySpec("SELECT * FROM root r WHERE [email protected]",
                    new SqlParameterCollection(new SqlParameter("@id", databaseId))), null);
    Iterator<Database> it = feedResponsePages.getQueryIterator();
    if (it == null || !it.hasNext()) {
        throw new RuntimeException("cannot find database " + databaseId);
    }
    return it.next();
}
 

import com.microsoft.azure.documentdb.DocumentClient; //导入依赖的package包/类
public static DocumentCollection getCollection(DocumentClient client, String databaseLink, String collectionId) {
    FeedResponse<DocumentCollection> feedResponsePages = client
            .queryCollections(databaseLink, new SqlQuerySpec("SELECT * FROM root r WHERE [email protected]",
                    new SqlParameterCollection(new SqlParameter("@id", collectionId))), null);
    Iterator<DocumentCollection> it = feedResponsePages.getQueryIterator();
    if (it == null || !it.hasNext()) {
        throw new RuntimeException("cannot find collection " + collectionId);
    }
    return it.next();
}
 

import com.microsoft.azure.documentdb.DocumentClient; //导入依赖的package包/类
public static RxAsyncBulkInsertBenchmark fromConfiguration(Configuration cfg) throws DocumentClientException {
    try (DocumentClient client = new DocumentClient(cfg.getServiceEndpoint(), cfg.getMasterKey(),
            cfg.getConnectionPolicy(), cfg.getConsistencyLevel())) {

        Database database = DocDBUtils.getDatabase(client, cfg.getDatabaseId());
        DocumentCollection collection = DocDBUtils.getCollection(client, database.getSelfLink(),
                cfg.getCollectionId());
        DatabaseAccount databaseAccount = client.getDatabaseAccount();

        return new RxAsyncBulkInsertBenchmark(cfg, database, collection, databaseAccount);
    }
}