MongoDbIO { public static Read read() { return new AutoValue_MongoDbIO_Read.Builder() .setMaxConnectionIdleTime(60000) .setNumSplits(0) .setBucketAuto(false) .setSslEnabled(false) .setIgnoreSSLCertificate(false) .setSslInvalidHostNameAllowed(false) .setQueryFn(FindQuery.create()) .build(); } private  MongoDbIO(); static Read read(); static Write write();  }
@Test public void testBuildAutoBuckets() { List<BsonDocument> aggregates = new ArrayList<BsonDocument>(); aggregates.add( new BsonDocument( "$match", new BsonDocument("country", new BsonDocument("$eq", new BsonString("England"))))); MongoDbIO.Read spec = MongoDbIO.read() .withUri("mongodb: .withDatabase(DATABASE) .withCollection(COLLECTION) .withQueryFn(AggregationQuery.create().withMongoDbPipeline(aggregates)); MongoDatabase database = client.getDatabase(DATABASE); List<Document> buckets = MongoDbIO.BoundedMongoDbSource.buildAutoBuckets(database, spec); assertEquals(10, buckets.size()); }
@Test public void testFullRead() { PCollection<Document> output = pipeline.apply( MongoDbIO.read() .withUri("mongodb: .withDatabase(DATABASE) .withCollection(COLLECTION)); PAssert.thatSingleton(output.apply("Count All", Count.globally())).isEqualTo(1000L); PAssert.that( output .apply("Map Scientist", MapElements.via(new DocumentToKVFn())) .apply("Count Scientist", Count.perKey())) .satisfies( input -> { for (KV<String, Long> element : input) { assertEquals(100L, element.getValue().longValue()); } return null; }); pipeline.run(); }
@Test public void testReadWithCustomConnectionOptions() { MongoDbIO.Read read = MongoDbIO.read() .withUri("mongodb: .withMaxConnectionIdleTime(10) .withDatabase(DATABASE) .withCollection(COLLECTION); assertEquals(10, read.maxConnectionIdleTime()); PCollection<Document> documents = pipeline.apply(read); PAssert.thatSingleton(documents.apply("Count All", Count.globally())).isEqualTo(1000L); PAssert.that( documents .apply("Map Scientist", MapElements.via(new DocumentToKVFn())) .apply("Count Scientist", Count.perKey())) .satisfies( input -> { for (KV<String, Long> element : input) { assertEquals(100L, element.getValue().longValue()); } return null; }); pipeline.run(); }
@Test public void testReadWithFilter() { PCollection<Document> output = pipeline.apply( MongoDbIO.read() .withUri("mongodb: .withDatabase(DATABASE) .withCollection(COLLECTION) .withQueryFn(FindQuery.create().withFilters(Filters.eq("scientist", "Einstein")))); PAssert.thatSingleton(output.apply("Count", Count.globally())).isEqualTo(100L); pipeline.run(); }
@Test public void testReadWithFilterAndLimit() throws Exception { PCollection<Document> output = pipeline.apply( MongoDbIO.read() .withUri("mongodb: .withDatabase(DATABASE) .withCollection(COLLECTION) .withNumSplits(10) .withQueryFn( FindQuery.create() .withFilters(Filters.eq("scientist", "Einstein")) .withLimit(5))); PAssert.thatSingleton(output.apply("Count", Count.globally())).isEqualTo(5L); pipeline.run(); }
@Test public void testReadWithAggregate() throws Exception { List<BsonDocument> aggregates = new ArrayList<BsonDocument>(); aggregates.add( new BsonDocument( "$match", new BsonDocument("country", new BsonDocument("$eq", new BsonString("England"))))); PCollection<Document> output = pipeline.apply( MongoDbIO.read() .withUri("mongodb: .withDatabase(DATABASE) .withCollection(COLLECTION) .withQueryFn(AggregationQuery.create().withMongoDbPipeline(aggregates))); PAssert.thatSingleton(output.apply("Count", Count.globally())).isEqualTo(300L); pipeline.run(); }
@Test public void testReadWithAggregateWithLimit() throws Exception { List<BsonDocument> aggregates = new ArrayList<BsonDocument>(); aggregates.add( new BsonDocument( "$match", new BsonDocument("country", new BsonDocument("$eq", new BsonString("England"))))); aggregates.add(new BsonDocument("$limit", new BsonInt32(10))); PCollection<Document> output = pipeline.apply( MongoDbIO.read() .withUri("mongodb: .withDatabase(DATABASE) .withCollection(COLLECTION) .withQueryFn(AggregationQuery.create().withMongoDbPipeline(aggregates))); PAssert.thatSingleton(output.apply("Count", Count.globally())).isEqualTo(10L); pipeline.run(); }