Merge pull request #14 from anfeldma-ms/master

Removed latch construct

Author: anfeldma-ms

pom.xml

@@ -58,8 +58,28 @@
         <dependency>
             <groupId>com.azure</groupId>
             <artifactId>azure-cosmos</artifactId>
-            <version>4.0.1</version>
+            <version>4.3.1</version>
         </dependency>
+
+        <dependency>
+            <groupId>org.apache.logging.log4j</groupId>
+            <artifactId>log4j-slf4j-impl</artifactId>
+            <version>2.13.0</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.logging.log4j</groupId>
+            <artifactId>log4j-api</artifactId>
+            <version>2.11.1</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-jdk14</artifactId>
+            <version>1.7.28</version>
+        </dependency>
+
         <dependency>
           <groupId>org.apache.commons</groupId>
           <artifactId>commons-lang3</artifactId>

src/main/java/com/azure/cosmos/sample/async/AsyncMain.java

@@ -20,12 +20,13 @@
 import com.azure.cosmos.sample.common.Families;
 import com.azure.cosmos.sample.common.Family;
 import com.azure.cosmos.util.CosmosPagedFlux;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.time.Duration;
 import java.util.Collections;
-import java.util.concurrent.CountDownLatch;
 import java.util.stream.Collectors;
 
 public class AsyncMain {
@@ -38,6 +39,8 @@ public class AsyncMain {
     private CosmosAsyncDatabase database;
     private CosmosAsyncContainer container;
 
+    protected static Logger logger = LoggerFactory.getLogger(AsyncMain.class.getSimpleName());
+
     public void close() {
         client.close();
     }
@@ -52,14 +55,14 @@ public static void main(String[] args) {
         AsyncMain p = new AsyncMain();
 
         try {
-            System.out.println("Starting ASYNC main");
+            logger.info("Starting ASYNC main");
             p.getStartedDemo();
-            System.out.println("Demo complete, please hold while resources are released");
+            logger.info("Demo complete, please hold while resources are released");
         } catch (Exception e) {
             e.printStackTrace();
-            System.err.println(String.format("Cosmos getStarted failed with %s", e));
+            logger.error("Cosmos getStarted failed with %s", e);
         } finally {
-            System.out.println("Closing the client");
+            logger.info("Closing the client");
             p.close();
         }
         System.exit(0);
@@ -68,7 +71,7 @@ public static void main(String[] args) {
     //  </Main>
 
     private void getStartedDemo() throws Exception {
-        System.out.println("Using Azure Cosmos DB endpoint: " + AccountSettings.HOST);
+        logger.info("Using Azure Cosmos DB endpoint: {}", AccountSettings.HOST);
 
         //  Create async client
         //  <CreateAsyncClient>
@@ -106,29 +109,29 @@ private void getStartedDemo() throws Exception {
                                 johnsonFamilyItem,
                                 smithFamilyItem);
 
-        System.out.println("Reading items.");
+        logger.info("Reading items.");
         readItems(familiesToCreate);
 
-        System.out.println("Querying items.");
+        logger.info("Querying items.");
         queryItems();
     }
 
     private void createDatabaseIfNotExists() throws Exception {
-        System.out.println("Create database " + databaseName + " if not exists.");
+        logger.info("Create database {} if not exists.", databaseName);
 
         //  Create database if not exists
         //  <CreateDatabaseIfNotExists>
         Mono<CosmosDatabaseResponse> databaseResponseMono = client.createDatabaseIfNotExists(databaseName);
         databaseResponseMono.flatMap(databaseResponse -> {
             database = client.getDatabase(databaseResponse.getProperties().getId());
-            System.out.println("Checking database " + database.getId() + " completed!\n");
+            logger.info("Checking database {} completed!\n", database.getId());
             return Mono.empty();
         }).block();
         //  </CreateDatabaseIfNotExists>
     }
 
     private void createContainerIfNotExists() throws Exception {
-        System.out.println("Create container " + containerName + " if not exists.");
+        logger.info("Create container {} if not exists.", containerName);
 
         //  Create container if not exists
         //  <CreateContainerIfNotExists>
@@ -139,7 +142,7 @@ private void createContainerIfNotExists() throws Exception {
         //  Create container with 400 RU/s
         containerResponseMono.flatMap(containerResponse -> {
             container = database.getContainer(containerResponse.getProperties().getId());
-            System.out.println("Checking container " + container.getId() + " completed!\n");
+            logger.info("Checking container {} completed!\n", container.getId());
             return Mono.empty();
         }).block();
 
@@ -150,46 +153,36 @@ private void createFamilies(Flux<Family> families) throws Exception {
 
         //  <CreateItem>
 
-        final CountDownLatch completionLatch = new CountDownLatch(1);
+        try {
 
-        //  Combine multiple item inserts, associated success println's, and a final aggregate stats println into one Reactive stream.
-        families.flatMap(family -> {
+            //  Combine multiple item inserts, associated success println's, and a final aggregate stats println into one Reactive stream.
+            double charge = families.flatMap(family -> {
                 return container.createItem(family);
             }) //Flux of item request responses
-            .flatMap(itemResponse -> {
-                System.out.println(String.format("Created item with request charge of %.2f within" +
-                    " duration %s",
-                    itemResponse.getRequestCharge(), itemResponse.getDuration()));
-                System.out.println(String.format("Item ID: %s\n", itemResponse.getItem().getId()));
-                return Mono.just(itemResponse.getRequestCharge());
-            }) //Flux of request charges
-            .reduce(0.0, 
-                (charge_n,charge_nplus1) -> charge_n + charge_nplus1
-            ) //Mono of total charge - there will be only one item in this stream            
-            .subscribe(charge -> { 
-                System.out.println(String.format("Created items with total request charge of %.2f\n",
-                charge));         
-            }, 
-                err -> {
-                    if (err instanceof CosmosException) {
-                        //Client-specific errors
-                        CosmosException cerr = (CosmosException)err;
-                        cerr.printStackTrace();
-                        System.err.println(String.format("Read Item failed with %s\n", cerr));
-                    } else {
-                        //General errors
-                        err.printStackTrace();
-                    }
-
-                    completionLatch.countDown();                    
-                }, 
-                () -> {completionLatch.countDown();}
-        ); //Preserve the total charge and print aggregate charge/item count stats.
-
-        try {
-            completionLatch.await();            
-        } catch (InterruptedException err) {
-            throw new AssertionError("Unexpected Interruption",err);
+                    .flatMap(itemResponse -> {
+                        logger.info("Created item with request charge of {}} within" +
+                                        " duration {}",
+                                itemResponse.getRequestCharge(), itemResponse.getDuration());
+                        logger.info("Item ID: {}\n", itemResponse.getItem().getId());
+                        return Mono.just(itemResponse.getRequestCharge());
+                    }) //Flux of request charges
+                    .reduce(0.0,
+                            (charge_n, charge_nplus1) -> charge_n + charge_nplus1
+                    ) //Mono of total charge - there will be only one item in this stream
+                    .block(); //Preserve the total charge and print aggregate charge/item count stats.
+
+            logger.info("Created items with total request charge of {}\n", charge);
+
+        } catch (Exception err) {
+            if (err instanceof CosmosException) {
+                //Client-specific errors
+                CosmosException cerr = (CosmosException) err;
+                cerr.printStackTrace();
+                logger.error("Read Item failed with %s\n", cerr);
+            } else {
+                //General errors
+                err.printStackTrace();
+            }
         }
 
         //  </CreateItem>            
@@ -200,39 +193,29 @@ private void readItems(Flux<Family> familiesToCreate) {
         //  This will help fast look up of items because of partition key
         //  <ReadItem>
 
-        final CountDownLatch completionLatch = new CountDownLatch(1);
-        
-        familiesToCreate.flatMap(family -> {
-                            Mono<CosmosItemResponse<Family>> asyncItemResponseMono = container.readItem(family.getId(), new PartitionKey(family.getLastName()), Family.class);
-                            return asyncItemResponseMono;
-                        })
-                        .subscribe(
-                            itemResponse -> {
-                                double requestCharge = itemResponse.getRequestCharge();
-                                Duration requestLatency = itemResponse.getDuration();
-                                System.out.println(String.format("Item successfully read with id %s with a charge of %.2f and within duration %s",
-                                    itemResponse.getItem().getId(), requestCharge, requestLatency));
-                            },
-                            err -> {
-                                if (err instanceof CosmosException) {
-                                    //Client-specific errors
-                                    CosmosException cerr = (CosmosException)err;
-                                    cerr.printStackTrace();
-                                    System.err.println(String.format("Read Item failed with %s\n", cerr));
-                                } else {
-                                    //General errors
-                                    err.printStackTrace();
-                                }
-
-                                completionLatch.countDown();
-                            },
-                            () -> {completionLatch.countDown();}
-        );
-
         try {
-            completionLatch.await();             
-        } catch (InterruptedException err) {
-            throw new AssertionError("Unexpected Interruption",err);
+
+            familiesToCreate.flatMap(family -> {
+                Mono<CosmosItemResponse<Family>> asyncItemResponseMono = container.readItem(family.getId(), new PartitionKey(family.getLastName()), Family.class);
+                return asyncItemResponseMono;
+            }).flatMap(itemResponse -> {
+                double requestCharge = itemResponse.getRequestCharge();
+                Duration requestLatency = itemResponse.getDuration();
+                logger.info("Item successfully read with id {} with a charge of {} and within duration {}",
+                        itemResponse.getItem().getId(), requestCharge, requestLatency);
+                return Flux.empty();
+            }).blockLast();
+
+        } catch (Exception err) {
+            if (err instanceof CosmosException) {
+                //Client-specific errors
+                CosmosException cerr = (CosmosException) err;
+                cerr.printStackTrace();
+                logger.error("Read Item failed with {}\n", cerr);
+            } else {
+                //General errors
+                err.printStackTrace();
+            }
         }
 
         //  </ReadItem>
@@ -242,47 +225,41 @@ private void queryItems() {
         //  <QueryItems>
         // Set some common query options
 
+        int preferredPageSize = 10; // We'll use this later
+
         CosmosQueryRequestOptions queryOptions = new CosmosQueryRequestOptions();
-        //  Set query metrics enabled to get metrics around query executions
+
+        //  Set populate query metrics to get metrics around query executions
         queryOptions.setQueryMetricsEnabled(true);
 
         CosmosPagedFlux<Family> pagedFluxResponse = container.queryItems(
-            "SELECT * FROM Family WHERE Family.lastName IN ('Andersen', 'Wakefield', 'Johnson')", queryOptions, Family.class);
-
-        final CountDownLatch completionLatch = new CountDownLatch(1);
-
-        pagedFluxResponse.byPage(10).subscribe(
-            fluxResponse -> {
-                System.out.println("Got a page of query result with " +
-                    fluxResponse.getResults().size() + " items(s)"
-                    + " and request charge of " + fluxResponse.getRequestCharge());
-
-                System.out.println("Item Ids " + fluxResponse
-                    .getResults()
-                    .stream()
-                    .map(Family::getId)
-                    .collect(Collectors.toList()));
-            },
-            err -> {
-                if (err instanceof CosmosException) {
-                    //Client-specific errors
-                    CosmosException cerr = (CosmosException)err;
-                    cerr.printStackTrace();
-                    System.err.println(String.format("Read Item failed with %s\n", cerr));
-                } else {
-                    //General errors
-                    err.printStackTrace();
-                }
-
-                completionLatch.countDown();
-            },
-            () -> {completionLatch.countDown();}
-        );
+                "SELECT * FROM Family WHERE Family.lastName IN ('Andersen', 'Wakefield', 'Johnson')", queryOptions, Family.class);
 
         try {
-            completionLatch.await();             
-        } catch (InterruptedException err) {
-            throw new AssertionError("Unexpected Interruption",err);
+
+            pagedFluxResponse.byPage(preferredPageSize).flatMap(fluxResponse -> {
+                logger.info("Got a page of query result with {} items(s)"
+                        + " and request charge of {}", fluxResponse.getResults().size(), fluxResponse.getRequestCharge());
+
+                logger.info("Item Ids {}" , fluxResponse
+                        .getResults()
+                        .stream()
+                        .map(Family::getId)
+                        .collect(Collectors.toList()));
+
+                return Flux.empty();
+            }).blockLast();
+
+        } catch(Exception err) {
+            if (err instanceof CosmosException) {
+                //Client-specific errors
+                CosmosException cerr = (CosmosException) err;
+                cerr.printStackTrace();
+                logger.error("Read Item failed with %s\n", cerr);
+            } else {
+                //General errors
+                err.printStackTrace();
+            }
         }
 
         // </QueryItems>