import com.microsoft.azure.documentdb.Document; //导入依赖的package包/类
protected <R extends Object> R readInternal(final DocumentDbPersistentEntity<?> entity, Class<R> type,
                                            final Document sourceDocument) {
    final ObjectMapper objectMapper = new ObjectMapper();
    objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
    try {
        final DocumentDbPersistentProperty idProperty = entity.getIdProperty();
        final Object idValue = sourceDocument.getId();

        final JSONObject jsonObject = new JSONObject(sourceDocument.toJson());
        if (idProperty != null) {
            // Replace the key id to the actual id field name in domain
            jsonObject.remove("id");
            jsonObject.put(idProperty.getName(), idValue);
        }

        return objectMapper.readValue(jsonObject.toString(), type);
    } catch (IOException e) {
        throw  new IllegalStateException("Failed to read the source document " + sourceDocument.toJson()
                + "  to target type " + type, e);
    }
}
 

import com.microsoft.azure.documentdb.Document; //导入依赖的package包/类
@Override
public TodoItem updateTodoItem(String id, boolean isComplete) {
	// Retrieve the document from the database
	Document todoItemDocument = getDocumentById(id);

	// You can update the document as a JSON document directly.
	// For more complex operations - you could de-serialize the document in
	// to a POJO, update the POJO, and then re-serialize the POJO back in to
	// a document.
	todoItemDocument.set("complete", isComplete);

	try {
		// Persist/replace the updated document.
		todoItemDocument = documentClient.replaceDocument(todoItemDocument,
				null).getResource();
	} catch (DocumentClientException e) {
		e.printStackTrace();
		return null;
	}

	return gson.fromJson(todoItemDocument.toString(), TodoItem.class);
}
 

import com.microsoft.azure.documentdb.Document; //导入依赖的package包/类
@Test
public void testCreateDocument_Async() throws Exception {
   
    // create a document
    Document doc = new Document(String.format("{ 'id': 'doc%d', 'counter': '%d'}", 1, 1));
    Observable<ResourceResponse<Document>> createDocumentObservable = asyncClient.createDocument(createdCollection.getSelfLink(), doc, null, true);

    final CountDownLatch doneLatch = new CountDownLatch(1);

    // subscribe to events emitted by the observable
    createDocumentObservable
        .single()           // we know there will be one response
        .subscribe(
            
            documentResourceResponse -> {
                System.out.println(documentResourceResponse.getActivityId());
                doneLatch.countDown();
            },

            error -> {
                System.err.println("an error happened in document creation: actual cause: " + error.getMessage());
            });

    // wait till document creation completes
    doneLatch.await();
}
 

import com.microsoft.azure.documentdb.Document; //导入依赖的package包/类
@Test
public void testCreateDocument_toBlocking() throws DocumentClientException {
    
    // create a document
    // toBlocking() converts the observable to a blocking observable
    Document doc = new Document(String.format("{ 'id': 'doc%d', 'counter': '%d'}", 1, 1));
    Observable<ResourceResponse<Document>> createDocumentObservable =
            asyncClient.createDocument(createdCollection.getSelfLink(), doc, null, true);

    
    // toBlocking() converts to a blocking observable
    // single() gets the only result
    createDocumentObservable
        .toBlocking()           //converts the observable to a blocking observable
        .single();              //gets the single result
}
 

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

    // attempt to create a document which already exists
    // - first create a document
    // - Using the async api generate an async document creation observable
    // - Converts the Observable to blocking using Observable.toBlocking() api
    // - catch already exist failure (409)
    Document doc = new Document(String.format("{ 'id': 'doc%d', 'counter': '%d'}", 1, 1));
    asyncClient.createDocument(createdCollection.getSelfLink(), doc, null, false).toBlocking().single();

    // Create the document
    Observable<ResourceResponse<Document>> createDocumentObservable = asyncClient
            .createDocument(createdCollection.getSelfLink(), doc, null, false);

    try {
        createDocumentObservable
            .toBlocking()           //converts the observable to a blocking observable
            .single();              //gets the single result
        Assert.fail("Document Already Exists. Document Creation must fail");
    } catch (Exception e) {
        assertThat("Document already exists.", 
                ((DocumentClientException) e.getCause()).getStatusCode(), equalTo(409));
    }
}
 

import com.microsoft.azure.documentdb.Document; //导入依赖的package包/类
@Override
public ArrayList<AlarmServiceModel> getList(DateTime from, DateTime to, String order, int skip,
                                            int limit, String[] devices) throws Exception {
    String sqlQuery = QueryBuilder.getDocumentsSQL(
        ALARM_SCHEMA_KEY,
        null, null,
        from, MESSAGE_RECEIVED_KEY,
        to, MESSAGE_RECEIVED_KEY,
        order, MESSAGE_RECEIVED_KEY,
        skip,
        limit,
        devices, DEVICE_ID_KEY);
    ArrayList<Document> docs = this.storageClient.queryDocuments(
        this.databaseName,
        this.collectionId,
        null,
        sqlQuery,
        skip);

    ArrayList<AlarmServiceModel> alarms = new ArrayList<>();
    for (Document doc : docs) {
        alarms.add(new AlarmServiceModel(doc));
    }

    return alarms;
}
 

import com.microsoft.azure.documentdb.Document; //导入依赖的package包/类
@Test
public void testTransformObservableToGoogleGuavaListenableFuture() throws Exception {
    
    // You can convert an Observable to a ListenableFuture.
    // ListenableFuture (part of google guava library) is a popular extension
    // of Java's Future which allows registering listener callbacks:
    // https://github.com/google/guava/wiki/ListenableFutureExplained
    Document doc = new Document(String.format("{ 'id': 'doc%d', 'counter': '%d'}", 1, 1));
    Observable<ResourceResponse<Document>> createDocumentObservable = asyncClient
            .createDocument(createdCollection.getSelfLink(), doc, null, false);
    ListenableFuture<ResourceResponse<Document>> listenableFuture = ListenableFutureObservable.to(createDocumentObservable);

    ResourceResponse<Document> rrd = listenableFuture.get();
    
    assertThat(rrd.getRequestCharge(), greaterThan((double) 0));
    System.out.print(rrd.getRequestCharge());
}
 

import com.microsoft.azure.documentdb.Document; //导入依赖的package包/类
@Override
public TodoItem createTodoItem(TodoItem todoItem) {
	// Serialize the TodoItem as a JSON Document.
	Document todoItemDocument = new Document(gson.toJson(todoItem));

	// Annotate the document as a TodoItem for retrieval (so that we can
	// store multiple entity types in the collection).
	todoItemDocument.set("entityType", "todoItem");

	try {
		// Persist the document using the DocumentClient.
		todoItemDocument = documentClient.createDocument(
				getTodoCollection().getSelfLink(), todoItemDocument, null,
				false).getResource();
	} catch (DocumentClientException e) {
		e.printStackTrace();
		return null;
	}

	return gson.fromJson(todoItemDocument.toString(), TodoItem.class);
}
 

import com.microsoft.azure.documentdb.Document; //导入依赖的package包/类
@Test
public void testConvertFromDocumentToEntity() {
    final JSONObject json = new JSONObject();
    json.put(idPropertyName, id);
    json.put(firstNamePropertyName, firstName);
    json.put(lastNamePropertyName, lastName);
    json.put(hobbiesPropertyName, hobbies);
    json.put(shippingAddressesPropertyName, addresses);

    final Document document = new Document(JSONObject.valueToString(json));
    final Person person = dbConverter.convertFromDocument(document, Person.class);
    assertThat(person.getId()).isEqualTo(id);
    assertThat(person.getFirstName()).isEqualTo(firstName);
    assertThat(person.getLastName()).isEqualTo(lastName);
    assertThat(person.getHobbies().equals(hobbies));
    assertThat(person.getShippingAddresses().equals(addresses));
}
 

import com.microsoft.azure.documentdb.Document; //导入依赖的package包/类
/**
 * This does the same thing as {@link #groupByInMemory_MoreDetail()} but with pedagogical details
 * @throws Exception
 */
@Test
public void groupByInMemory_MoreDetail() throws Exception {

    int requestPageSize = 3;
    FeedOptions options = new FeedOptions();
    options.setPageSize(requestPageSize);


    Observable<Document> documentsObservable = asyncClient
            .queryDocuments(createdCollection.getSelfLink(),
                    new SqlQuerySpec("SELECT * FROM root r WHERE [email protected]_id",
                            new SqlParameterCollection(new SqlParameter("@site_id", "ABC"))),
                    options)
            .flatMap(page -> Observable.from(page.getResults()));

    final LocalDateTime now = LocalDateTime.now();

    Observable<GroupedObservable<Integer, Document>> groupedByPayerIdObservable = documentsObservable
            .filter(doc -> Math.abs(now.getSecond() - doc.getInt("created_time")) <= 90)
            .groupBy(doc -> doc.getInt("payer_id"));

    Observable<List<Document>> docsGroupedAsList = groupedByPayerIdObservable.flatMap(grouped -> {
        Observable<List<Document>> list = grouped.toList();
        return list;
    });

    List<List<Document>> resultsGroupedAsLists = docsGroupedAsList.toList().toBlocking().single();

    for(List<Document> resultsForEachPayer : resultsGroupedAsLists) {
        System.out.println("documents with payer_id : " + resultsForEachPayer.get(0).getInt("payer_id") + " are " + resultsForEachPayer);
    }
}
 

import com.microsoft.azure.documentdb.Document; //导入依赖的package包/类
@Override
public Observable<ResourceResponse<Document>> replaceDocument(String documentLink, Object document,
        RequestOptions options) {
    return Observable.defer(() -> {

        try {
            if (StringUtils.isEmpty(documentLink)) {
                throw new IllegalArgumentException("documentLink");
            }

            if (document == null) {
                throw new IllegalArgumentException("document");
            }

            Document typedDocument = documentFromObject(document);

            return this.replaceDocumentInternal(documentLink, typedDocument, options);

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

import com.microsoft.azure.documentdb.Document; //导入依赖的package包/类
@Override
public List<TodoItem> readTodoItems() {
	List<TodoItem> todoItems = new ArrayList<TodoItem>();
	// Retrieve the TodoItem documents
	List<Document> documentList = documentClient
			.queryDocuments(getTodoCollection().getSelfLink(),
					"SELECT * FROM root r WHERE r.entityType = 'todoItem'",
					null).getQueryIterable().toList();

	// De-serialize the documents in to TodoItems.
	for (Document todoItemDocument : documentList) {
		todoItems.add(gson.fromJson(todoItemDocument.toString(),
				TodoItem.class));
	}
	return todoItems;
}
 

import com.microsoft.azure.documentdb.Document; //导入依赖的package包/类
@Test
public void testConvertFromEntityToDocument() {
    final Person person = new Person(id, firstName, lastName, hobbies, addresses);
    final Document document = dbConverter.convertToDocument(person);

    assertTrue(document.has(idPropertyName));
    assertTrue(document.has(firstNamePropertyName));
    assertTrue(document.has(lastNamePropertyName));
    assertTrue(document.has(hobbiesPropertyName));
    assertTrue(document.has(shippingAddressesPropertyName));
    assertThat(document.getId()).isEqualTo(id);
    assertThat(document.getString(firstNamePropertyName)).isEqualTo(firstName);
    assertThat(document.getString(lastNamePropertyName)).isEqualTo(lastName);

    final Collection<String> convertedHobbies = document.getCollection(hobbiesPropertyName, String.class);
    assertTrue(hobbies.equals(convertedHobbies));

    final Collection<Address> convertedAddresses =
            document.getCollection(shippingAddressesPropertyName, Address.class);
    assertTrue(addresses.equals(convertedAddresses));
}
 

import com.microsoft.azure.documentdb.Document; //导入依赖的package包/类
@Test(groups = { "simple" }, timeOut = TIMEOUT)
public void queryDocuments() throws Exception {

    String query = "SELECT * from root";
    FeedOptions options = new FeedOptions();
    options.setEnableCrossPartitionQuery(true);
    Observable<FeedResponsePage<Document>> queryObservable = client
            .queryDocuments(getCollectionLink(), query, options);

    FeedResponsePageListValidator<Document> validator = new FeedResponsePageListValidator
            .Builder<Document>()
            .containsExactly(createdDocuments
                    .stream()
                    .map(d -> d.getResourceId())
                    .collect(Collectors.toList()))
            .allPagesSatisfy(new FeedResponsePageValidator.Builder<Document>()
                    .requestChargeGreaterThanOrEqualTo(1.0).build())
            .build();
    validateQuerySuccess(queryObservable, validator);
}
 

import com.microsoft.azure.documentdb.Document; //导入依赖的package包/类
@Test(groups = { "simple" }, timeOut = TIMEOUT)
public void queryDocuments_NoResults() throws Exception {

    String query = "SELECT * from root r where r.id = '2'";
    FeedOptions options = new FeedOptions();
    options.setEnableCrossPartitionQuery(true);
    Observable<FeedResponsePage<Document>> queryObservable = client
            .queryDocuments(getCollectionLink(), query, options);

    FeedResponsePageListValidator<Document> validator = new FeedResponsePageListValidator.Builder<Document>()
            .containsExactly(new ArrayList<>())
            .numberOfPages(1)
            .pageSatisfy(0, new FeedResponsePageValidator.Builder<Document>()
                    .requestChargeGreaterThanOrEqualTo(1.0).build())
            .build();
    validateQuerySuccess(queryObservable, validator);
}
 

import com.microsoft.azure.documentdb.Document; //导入依赖的package包/类
@Test(groups = { "simple" }, timeOut = TIMEOUT)
public void queryDocumentsWithPageSize() throws Exception {

    String query = "SELECT * from root";
    FeedOptions options = new FeedOptions();
    options.setPageSize(3);
    options.setEnableCrossPartitionQuery(true);
    Observable<FeedResponsePage<Document>> queryObservable = client
            .queryDocuments(getCollectionLink(), query, options);

    FeedResponsePageListValidator<Document> validator = new FeedResponsePageListValidator
            .Builder<Document>()
            .containsExactly(createdDocuments
                    .stream()
                    .map(d -> d.getResourceId())
                    .collect(Collectors.toList()))
            .numberOfPages((createdDocuments.size() + 1) / 3)
            .allPagesSatisfy(new FeedResponsePageValidator.Builder<Document>()
                    .requestChargeGreaterThanOrEqualTo(1.0).build())
            .build();
    validateQuerySuccess(queryObservable, validator);
}
 

import com.microsoft.azure.documentdb.Document; //导入依赖的package包/类
@Test(groups = { "simple" }, timeOut = TIMEOUT)
public void queryOrderBy() throws Exception {

    String query = "SELECT * FROM r ORDER BY r.prop ASC";
    FeedOptions options = new FeedOptions();
    options.setEnableCrossPartitionQuery(true);
    options.setPageSize(3);
    Observable<FeedResponsePage<Document>> queryObservable = client
            .queryDocuments(getCollectionLink(), query, options);

    FeedResponsePageListValidator<Document> validator = new FeedResponsePageListValidator.Builder<Document>()
            .containsExactly(createdDocuments.stream()
                    .sorted((e1, e2) -> Integer.compare(e1.getInt("prop"), e2.getInt("prop")))
                    .map(d -> d.getResourceId()).collect(Collectors.toList()))
            .numberOfPages((createdDocuments.size() + 1) / 3)
            .allPagesSatisfy(new FeedResponsePageValidator.Builder<Document>()
                    .requestChargeGreaterThanOrEqualTo(1.0).build())
            .build();
    validateQuerySuccess(queryObservable, validator);
}
 

import com.microsoft.azure.documentdb.Document; //导入依赖的package包/类
@Test(groups = { "simple" }, timeOut = TIMEOUT, enabled = false)
public void invalidQuerySytax() throws Exception {

    // NOTE: this test passes on Linux but not on Windows in the presence of dlls
    
    // ServiceJNIWrapper in DocumentClient throws IllegalArgumentException instead of DocumentClientException
    // after the behavior is fixed enable this test
    String query = "I am an invalid query";
    FeedOptions options = new FeedOptions();
    options.setEnableCrossPartitionQuery(true);
    Observable<FeedResponsePage<Document>> queryObservable = client
            .queryDocuments(getCollectionLink(), query, options);

    FailureValidator validator = new FailureValidator.Builder()
            .instanceOf(DocumentClientException.class)
            .statusCode(400)
            .notNullActivityId()
            .build();
    validateQueryFailure(queryObservable, validator);
}
 

import com.microsoft.azure.documentdb.Document; //导入依赖的package包/类
@Test(groups = { "simple" }, timeOut = TIMEOUT)
public void readDocument() throws Exception {
    Document docDefinition = getDocumentDefinition();

    Document document = client
            .createDocument(getCollectionLink(), docDefinition, null, false).toBlocking().single().getResource();


    RequestOptions options = new RequestOptions();
    options.setPartitionKey(new PartitionKey(document.get("mypk")));
    Observable<ResourceResponse<Document>> readObservable = client.readDocument(document.getSelfLink(), options);

    ResourceResponseValidator<Document> validator = new ResourceResponseValidator.Builder<Document>()
            .withId(document.getId())
            .build();
    validateSuccess(readObservable, validator);
}
 

import com.microsoft.azure.documentdb.Document; //导入依赖的package包/类
@Test(groups = { "simple" }, timeOut = TIMEOUT)
public void readDocument_DoesntExist() throws Exception {
    Document docDefinition = getDocumentDefinition();

    Document document = client
            .createDocument(getCollectionLink(), docDefinition, null, false).toBlocking().single().getResource();

    RequestOptions options = new RequestOptions();
    options.setPartitionKey(new PartitionKey(document.get("mypk")));
    client.deleteDocument(document.getSelfLink(), options).toBlocking().first();

    options.setPartitionKey(new PartitionKey("looloo"));
    Observable<ResourceResponse<Document>> readObservable = client.readDocument(document.getSelfLink(), options);

    FailureValidator validator = new FailureValidator.Builder().instanceOf(DocumentClientException.class)
            .statusCode(404).build();
    validateFailure(readObservable, validator);
}