import com.microsoft.azure.documentdb.DocumentCollection; //导入依赖的package包/类
@Before
public void setUp() throws DocumentClientException {

    asyncClient = new AsyncDocumentClient.Builder()
            .withServiceEndpoint(TestConfigurations.HOST)
            .withMasterKey(TestConfigurations.MASTER_KEY)
            .withConnectionPolicy(ConnectionPolicy.GetDefault())
            .withConsistencyLevel(ConsistencyLevel.Session)
            .build();

    // Clean up before setting up
    this.cleanUpGeneratedDatabases();

    databaseDefinition = new Database();
    databaseDefinition.setId(DATABASE_ID);
    
    collectionDefinition = new DocumentCollection();
    collectionDefinition.setId(UUID.randomUUID().toString());
}
 

import com.microsoft.azure.documentdb.DocumentCollection; //导入依赖的package包/类
private DocumentCollection createMultiPartitionCollection(String databaseLink, String collectionId,
        String partitionKeyPath) throws DocumentClientException {
    PartitionKeyDefinition partitionKeyDef = new PartitionKeyDefinition();
    ArrayList<String> paths = new ArrayList<String>();
    paths.add(partitionKeyPath);
    partitionKeyDef.setPaths(paths);

    RequestOptions options = new RequestOptions();
    options.setOfferThroughput(10100);
    DocumentCollection collectionDefinition = new DocumentCollection();
    collectionDefinition.setId(collectionId);
    collectionDefinition.setPartitionKey(partitionKeyDef);
    DocumentCollection createdCollection = asyncClient
            .createCollection(databaseLink, collectionDefinition, options).toBlocking().single().getResource();

    return createdCollection;
}
 

import com.microsoft.azure.documentdb.DocumentCollection; //导入依赖的package包/类
@Override
public Observable<ResourceResponse<DocumentCollection>> replaceCollection(DocumentCollection collection,
        RequestOptions options) {
    return Observable.defer(() -> {
        try {
            if (collection == null) {
                throw new IllegalArgumentException("collection");
            }

            logger.debug("Replacing a Collection. id: [{}]", collection.getId());
            validateResource(collection);

            String path = Utils.joinPath(collection.getSelfLink(), null);
            Map<String, String> requestHeaders = this.getRequestHeaders(options);

            RxDocumentServiceRequest request = RxDocumentServiceRequest.create(OperationType.Replace,
                    ResourceType.DocumentCollection, path, collection, requestHeaders);

            return this.doReplace(request).map(response -> toResourceResponse(response, DocumentCollection.class));

        } catch (Exception e) {
            logger.debug("Failure in replacing a collection. due to [{}]", e.getMessage(), e);
            return Observable.error(e);
        }
    });
}
 

import com.microsoft.azure.documentdb.DocumentCollection; //导入依赖的package包/类
@Override
public Observable<ResourceResponse<DocumentCollection>> deleteCollection(String collectionLink,
        RequestOptions options) {
    return Observable.defer(() -> {
        try {
            if (StringUtils.isEmpty(collectionLink)) {
                throw new IllegalArgumentException("collectionLink");
            }

            logger.debug("Deleting a Collection. collectionLink: [{}]", collectionLink);
            String path = Utils.joinPath(collectionLink, null);
            Map<String, String> requestHeaders = this.getRequestHeaders(options);
            RxDocumentServiceRequest request = RxDocumentServiceRequest.create(OperationType.Delete,
                    ResourceType.DocumentCollection, path, requestHeaders);
            return this.doDelete(request).map(response -> toResourceResponse(response, DocumentCollection.class));

        } catch (Exception e) {
            logger.debug("Failure in deleting a collection, due to [{}]", e.getMessage(), e);
            return Observable.error(e);
        }
    });
}
 

import com.microsoft.azure.documentdb.DocumentCollection; //导入依赖的package包/类
private Observable<DocumentServiceResponse> doDelete(RxDocumentServiceRequest request)
        throws DocumentClientException {

    Observable<DocumentServiceResponse> responseObservable = Observable.defer(() -> {
        try {
            return this.gatewayProxy.doDelete(request).doOnNext(response -> {
                if (request.getResourceType() != ResourceType.DocumentCollection) {
                    this.captureSessionToken(request, response);
                } else {
                    this.clearToken(request, response);
                }
            });
        } catch (Exception e) {
            return Observable.error(e);
        }
    }).retryWhen(createExecuteRequestRetryHandler(request));
    
    return createPutMoreContentObservable(request, HttpConstants.HttpMethods.DELETE)
            .doOnNext(req -> this.applySessionToken(request))
            .flatMap(req -> responseObservable);
}
 

import com.microsoft.azure.documentdb.DocumentCollection; //导入依赖的package包/类
@Test(groups = { "simple" }, timeOut = TIMEOUT)
public void deleteCollection() throws Exception {
    DocumentCollection collectionDefinition = getCollectionDefinition();
    
    Observable<ResourceResponse<DocumentCollection>> createObservable = client.createCollection(database.getSelfLink(), collectionDefinition, null);
    DocumentCollection collection = createObservable.toBlocking().single().getResource();

    Observable<ResourceResponse<DocumentCollection>> deleteObservable = client.deleteCollection(collection.getSelfLink(),
            null);

    ResourceResponseValidator<DocumentCollection> validator = new ResourceResponseValidator.Builder<DocumentCollection>()
            .nullResource().build();
    validateSuccess(deleteObservable, validator);
    
    //TODO validate after deletion the resource is actually deleted (not found)
}
 

import com.microsoft.azure.documentdb.DocumentCollection; //导入依赖的package包/类
@Test(groups = { "simple" }, timeOut = TIMEOUT)
public void replaceCollection() throws Exception {
    // create a collection
    DocumentCollection collectionDefinition = getCollectionDefinition();
    Observable<ResourceResponse<DocumentCollection>> createObservable = client.createCollection(database.getSelfLink(), collectionDefinition, null);
    DocumentCollection collection = createObservable.toBlocking().single().getResource();
    // sanity check
    assertThat(collection.getIndexingPolicy().getIndexingMode()).isEqualTo(IndexingMode.Consistent);
    
    // replace indexing mode
    IndexingPolicy indexingMode = new IndexingPolicy();
    indexingMode.setIndexingMode(IndexingMode.Lazy);
    collection.setIndexingPolicy(indexingMode);
    Observable<ResourceResponse<DocumentCollection>> readObservable = client.replaceCollection(collection, null);
    
    // validate
    ResourceResponseValidator<DocumentCollection> validator = new ResourceResponseValidator.Builder<DocumentCollection>()
            .indexingMode(IndexingMode.Lazy).build();
    validateSuccess(readObservable, validator);
}
 

import com.microsoft.azure.documentdb.DocumentCollection; //导入依赖的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.DocumentCollection; //导入依赖的package包/类
private void replaceOffer(DocumentCollection collection, int desiredOfferThroughput) throws DocumentClientException {
    FeedResponse<Offer> offers = client.queryOffers(String.format("SELECT * FROM c where c.offerResourceId = '%s'", collection.getResourceId()), null);

    Iterator<Offer> offerIterator = offers.getQueryIterator();
    if (!offerIterator.hasNext()) {
        throw new IllegalStateException("Cannot find Collection's corresponding offer");
    }

    Offer offer = offerIterator.next();
    int collectionThroughput = offer.getContent().getInt("offerThroughput");
    assertThat(collectionThroughput, equalTo(400));
    
    offer.getContent().put("offerThroughput", desiredOfferThroughput);
    Offer newOffer = client.replaceOffer(offer).getResource();
    assertThat(newOffer.getContent().getInt("offerThroughput"), equalTo(desiredOfferThroughput));
}
 

import com.microsoft.azure.documentdb.DocumentCollection; //导入依赖的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.DocumentCollection; //导入依赖的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.DocumentCollection; //导入依赖的package包/类
/**
 * Writes data to DocumentDB if the cached documents reach the maximum cache size.
 */
public void write(Writable key, DocumentDBWritable value) throws IOException {
    Document doc = value.getDoc();
    DocumentCollection targetCollection = this.collections[this.currentStoredProcedureIndex];
    currentStoredProcedureIndex = (this.currentStoredProcedureIndex + 1) % this.collections.length;
    this.documentsProcessed++;
    if(targetCollection.getPartitionKey() != null) {
        DocumentDBConnectorUtil.createDocument(this.client, targetCollection.getSelfLink(), doc, this.enableUpsert);
        if (documentsProcessed % MAX_DOC_SIZE == 0) {
        	LOG.info(String.format("wrote %d documents", this.documentsProcessed));
        }
    } else {
        DocumentDBConnectorUtil.addIdIfMissing(doc);
        this.cachedDocs.add(doc);
        if (this.documentsProcessed % MAX_DOC_SIZE == 0) {
            this.writeCurrentBatch();
            LOG.info(String.format("wrote %d documents", this.documentsProcessed));
        }
    }
}
 

import com.microsoft.azure.documentdb.DocumentCollection; //导入依赖的package包/类
/**
 * Writes data to DocumentDB if the cached documents reach the maximum cache size.
 */
public void write(Writable key, DocumentDBWritable value) throws IOException {
    Document doc = value.getDoc();
    DocumentCollection targetCollection = this.collections[this.currentStoredProcedureIndex];
    currentStoredProcedureIndex = (this.currentStoredProcedureIndex + 1) % this.collections.length;
    this.documentsProcessed++;
    if(targetCollection.getPartitionKey() != null) {
       DocumentDBConnectorUtil.createDocument(this.client, this.collections[this.currentStoredProcedureIndex].getSelfLink(), doc, this.enableUpsert);
       if (documentsProcessed % MAX_DOC_SIZE == 0) {
    	   LOG.info(String.format("wrote %d documents", this.documentsProcessed));
       }
    } else {
        DocumentDBConnectorUtil.addIdIfMissing(doc);
        this.cachedDocs.add(doc);
        if (documentsProcessed % MAX_DOC_SIZE == 0) {
            this.writeCurrentBatch();
            LOG.info(String.format("wrote %d documents", this.documentsProcessed));
        }
    }
}
 

import com.microsoft.azure.documentdb.DocumentCollection; //导入依赖的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.DocumentCollection; //导入依赖的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.DocumentCollection; //导入依赖的package包/类
public AbstractBulkInsertBenchmark(Configuration cfg, Database database, DocumentCollection collection) throws DocumentClientException {
    this.cfg = cfg;
    this.database = database;
    this.collection = collection;
    this.reporter = ConsoleReporter.forRegistry(metricsRegistry)
            .convertRatesTo(TimeUnit.SECONDS)
            .convertDurationsTo(TimeUnit.MILLISECONDS)
            .build();

    this.logger = LoggerFactory.getLogger(this.getClass());
}
 

import com.microsoft.azure.documentdb.DocumentCollection; //导入依赖的package包/类
RxAsyncBulkInsertBenchmark(Configuration cfg, Database database, DocumentCollection collection,
        DatabaseAccount databaseAccount) throws DocumentClientException {
    super(cfg, database, collection);

    // the throttle semaphore is used to throttle concurrent async creates
    // if we don't throttle when the pool exhausted the subsequent async
    // create will immediately fail
    this.throttle = new Semaphore(cfg.getConnectionPolicy().getMaxPoolSize());
}
 

import com.microsoft.azure.documentdb.DocumentCollection; //导入依赖的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);
    }
}
 

import com.microsoft.azure.documentdb.DocumentCollection; //导入依赖的package包/类
@Before
public void setUp() throws DocumentClientException {

    // sets up the requirements for each test
    
    asyncClient = new AsyncDocumentClient.Builder()
            .withServiceEndpoint(TestConfigurations.HOST)
            .withMasterKey(TestConfigurations.MASTER_KEY)
            .withConnectionPolicy(ConnectionPolicy.GetDefault())
            .withConsistencyLevel(ConsistencyLevel.Session)
            .build();
    // Clean up the database.
    this.cleanUpGeneratedDatabases();

    Database databaseDefinition = new Database();
    databaseDefinition.setId(DATABASE_ID);

    DocumentCollection collectionDefinition = new DocumentCollection();
    collectionDefinition.setId(UUID.randomUUID().toString());

    // create database
    ResourceResponse<Database> databaseCreationResponse = asyncClient.createDatabase(databaseDefinition, null)
            .toBlocking().single();
    
    // create collection
    createdCollection = asyncClient
            .createCollection(databaseCreationResponse.getResource().getSelfLink(), collectionDefinition, null)
            .toBlocking().single().getResource();
}
 

import com.microsoft.azure.documentdb.DocumentCollection; //导入依赖的package包/类
@Before
public void setUp() throws DocumentClientException {

    asyncClient = new AsyncDocumentClient.Builder()
            .withServiceEndpoint(TestConfigurations.HOST)
            .withMasterKey(TestConfigurations.MASTER_KEY)
            .withConnectionPolicy(ConnectionPolicy.GetDefault())
            .withConsistencyLevel(ConsistencyLevel.Session)
            .build();
    
    // Clean up the database.
    this.cleanUpGeneratedDatabases();

    Database databaseDefinition = new Database();
    databaseDefinition.setId(DATABASE_ID);

    DocumentCollection collectionDefinition = new DocumentCollection();
    collectionDefinition.setId(UUID.randomUUID().toString());

    // create database
    ResourceResponse<Database> databaseCreationResponse = asyncClient.createDatabase(databaseDefinition, null)
            .toBlocking().single();

    createdDatabase = databaseCreationResponse.getResource();
    
    // create collection
    createdCollection = asyncClient
            .createCollection(databaseCreationResponse.getResource().getSelfLink(), collectionDefinition, null)
            .toBlocking().single().getResource();

    numberOfDocuments = 20;
    // add documents
    for (int i = 0; i < numberOfDocuments; i++) {
        Document doc = new Document(String.format("{ 'id': 'loc%d', 'counter': %d}", i, i));
        asyncClient.createDocument(createdCollection.getSelfLink(), doc, null, true).toBlocking().single();
    }
}