import com.microsoft.azure.documentdb.DocumentClientException; //导入依赖的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.DocumentClientException; //导入依赖的package包/类
@Test
public void testCreateDatabase_toBlocking_DatabaseAlreadyExists_Fails() throws DocumentClientException {

    // attempt to create a database which already exists
    // - first create a database
    // - Using the async api generate an async database creation observable
    // - Converts the Observable to blocking using Observable.toBlocking() api
    // - catch already exist failure (409)
    
    asyncClient.createDatabase(databaseDefinition, null).toBlocking().single();

    // Create the database for test.
    Observable<ResourceResponse<Database>> databaseForTestObservable = asyncClient
            .createDatabase(databaseDefinition, null);

    try {
        databaseForTestObservable
            .toBlocking()           //blocks
            .single();              //gets the single result
        assertThat("Should not reach here", false);
    } catch (Exception e) {
        assertThat("Database already exists.", 
                ((DocumentClientException) e.getCause()).getStatusCode(), equalTo(409));
    }
}
 

import com.microsoft.azure.documentdb.DocumentClientException; //导入依赖的package包/类
private void cleanUpGeneratedDatabases() throws DocumentClientException {
    LOGGER.info("cleanup databases invoked");

    String[] allDatabaseIds = { DATABASE_ID };

    for (String id : allDatabaseIds) {
        try {
            List<FeedResponsePage<Database>> feedResponsePages = asyncClient
                    .queryDatabases(new SqlQuerySpec("SELECT * FROM root r WHERE [email protected]",
                            new SqlParameterCollection(new SqlParameter("@id", id))), null).toList().toBlocking().single();
            
            
            if (!feedResponsePages.get(0).getResults().isEmpty()) {
                Database res = feedResponsePages.get(0).getResults().get(0);
                LOGGER.info("deleting a database " + feedResponsePages.get(0));
                asyncClient.deleteDatabase(res.getSelfLink(), null).toBlocking().single();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
 

import com.microsoft.azure.documentdb.DocumentClientException; //导入依赖的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.DocumentClientException; //导入依赖的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.DocumentClientException; //导入依赖的package包/类
private void cleanUpGeneratedDatabases() throws DocumentClientException {
    LOGGER.info("cleanup databases invoked");

    String[] allDatabaseIds = { DATABASE_ID };

    for (String id : allDatabaseIds) {
        try {
            List<FeedResponsePage<Database>> feedResponsePages = asyncClient
                    .queryDatabases(new SqlQuerySpec("SELECT * FROM root r WHERE [email protected]",
                            new SqlParameterCollection(new SqlParameter("@id", id))), null)
                    .toList().toBlocking().single();

            if (!feedResponsePages.get(0).getResults().isEmpty()) {
                Database res = feedResponsePages.get(0).getResults().get(0);
                LOGGER.info("deleting a database " + feedResponsePages.get(0));
                asyncClient.deleteDatabase(res.getSelfLink(), null).toBlocking().single();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
 

import com.microsoft.azure.documentdb.DocumentClientException; //导入依赖的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.DocumentClientException; //导入依赖的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.DocumentClientException; //导入依赖的package包/类
private Observable<DocumentServiceResponse> doRead(RxDocumentServiceRequest request)
        throws DocumentClientException {

    Observable<DocumentServiceResponse> responseObservable = Observable.defer(() -> {
        try {
            return this.gatewayProxy.processMessage(request).doOnNext(response -> {
                this.captureSessionToken(request, response);
            });
        } catch (Exception e) {
            return Observable.error(e);
        }
    }).retryWhen(createExecuteRequestRetryHandler(request));
    
    return createPutMoreContentObservable(request, HttpConstants.HttpMethods.GET)
        .doOnNext(req -> this.applySessionToken(request))
        .flatMap(req -> responseObservable);
}
 

import com.microsoft.azure.documentdb.DocumentClientException; //导入依赖的package包/类
private Observable<Long> handleRetryAttemptInternal(DocumentClientException e, int attemptNumber) throws DocumentClientException {

        RetryPolicy retryPolicy = null;
        if (e.getStatusCode() == HttpConstants.StatusCodes.BADREQUEST && e.getSubStatusCode() != null
                && e.getSubStatusCode() == HttpConstants.SubStatusCodes.PARTITION_KEY_MISMATCH) {
            // If HttpStatusCode is 404 (NotFound) and SubStatusCode is
            // 1001 (PartitionKeyMismatch), invoke the partition key mismatch retry policy
            retryPolicy = keyMismatchRetryPolicy;
        }

        if (retryPolicy == null || !retryPolicy.shouldRetry(e)) {
            LOGGER.trace("Execution encontured exception: {}, status code {} sub status code {}. Won't retry!", 
                    e.getMessage(), e.getStatusCode(), e.getSubStatusCode());
            return Observable.error(e);
        }
        LOGGER.trace("Execution encontured exception: {}, status code {} sub status code {}. Will retry in {}ms", 
                e.getMessage(), e.getStatusCode(), e.getSubStatusCode(), retryPolicy.getRetryAfterInMilliseconds());
       
        long waitTime = retryPolicy.getRetryAfterInMilliseconds();
        return Observable.just(waitTime);
    }
 

import com.microsoft.azure.documentdb.DocumentClientException; //导入依赖的package包/类
private static Observable<Long> handleRetryAttempt(Throwable t, int attemptNumber, RxRetryHandler retryPolicy) {
    Throwable cause = extractDocumentClientCause(t, attemptNumber);
    
    if (LOGGER.isDebugEnabled()) {
        if (cause instanceof DocumentClientException) {
            DocumentClientException ex = (DocumentClientException) cause;
            LOGGER.debug("Handling Failure Attempt [{}], StatusCode [{}], SubStatusCode,"
                    + " Error: [{}] ", attemptNumber, ex.getStatusCode(), ex.getSubStatusCode(), ex.getError(), ex);
        } else {
            LOGGER.debug("Handling Failure Attempt [{}], req [{}]", attemptNumber, cause);
        }
    }
    
    try {
        return retryPolicy.handleRetryAttempt(cause, attemptNumber);
    } catch (Exception e) {
        return Observable.error(e);
    }
}
 

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

import com.microsoft.azure.documentdb.DocumentClientException; //导入依赖的package包/类
private void setupSpySession(final List<String> capturedRequestSessionTokenList, final List<String> capturedResponseSessionTokenList, 
        RxDocumentClientImpl spyClient, final RxDocumentClientImpl origClient) throws DocumentClientException {

    Mockito.reset(spyClient);
    doAnswer(new Answer<Void>() {
        public Void answer(InvocationOnMock invocation) throws Throwable {
            Object[] args = invocation.getArguments();
            RxDocumentServiceRequest req = (RxDocumentServiceRequest) args[0];
            DocumentServiceResponse resp = (DocumentServiceResponse) args[1];

            capturedRequestSessionTokenList.add(req.getHeaders().get(HttpConstants.HttpHeaders.SESSION_TOKEN));
            capturedResponseSessionTokenList.add(resp.getResponseHeaders().get(HttpConstants.HttpHeaders.SESSION_TOKEN));

            origClient.captureSessionToken(req, resp);

            return null;
        }})
    .when(spyClient).captureSessionToken(Mockito.any(RxDocumentServiceRequest.class), Mockito.any(DocumentServiceResponse.class));
}
 

import com.microsoft.azure.documentdb.DocumentClientException; //导入依赖的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.DocumentClientException; //导入依赖的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.DocumentClientException; //导入依赖的package包/类
@Override
public boolean deleteTodoItem(String id) {
	// DocumentDB refers to documents by self link rather than id.

	// Query for the document to retrieve the self link.
	Document todoItemDocument = getDocumentById(id);

	try {
		// Delete the document by self link.
		documentClient.deleteDocument(todoItemDocument.getSelfLink(), null);
	} catch (DocumentClientException e) {
		e.printStackTrace();
		return false;
	}

	return true;
}
 

import com.microsoft.azure.documentdb.DocumentClientException; //导入依赖的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.DocumentClientException; //导入依赖的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.DocumentClientException; //导入依赖的package包/类
@Override
public boolean deleteTodoItem(String id) {
    // DocumentDB refers to documents by self link rather than id.

    // Query for the document to retrieve the self link.
    Document todoItemDocument = getDocumentById(id);

    try {
        // Delete the document by self link.
        documentClient.deleteDocument(todoItemDocument.getSelfLink(), null);
    } catch (DocumentClientException e) {
        e.printStackTrace();
        return false;
    }

    return true;
}