feat: mongdb read preferences by query

quarkusio · Dec 11, 2020 · b2c6b4c · b2c6b4c
1 parent de1b9c6
commit b2c6b4c
Show file tree

Hide file tree

Showing 20 changed files with 121 additions and 6 deletions.
diff --git a/...odb-client/runtime/src/main/java/io/quarkus/mongodb/impl/ReactiveMongoCollectionImpl.java b/...odb-client/runtime/src/main/java/io/quarkus/mongodb/impl/ReactiveMongoCollectionImpl.java
@@ -7,6 +7,7 @@
 import org.bson.conversions.Bson;
 
 import com.mongodb.MongoNamespace;
+import com.mongodb.ReadPreference;
 import com.mongodb.bulk.BulkWriteResult;
 import com.mongodb.client.model.BulkWriteOptions;
 import com.mongodb.client.model.CountOptions;
@@ -825,4 +826,9 @@ public CodecRegistry getCodecRegistry() {
     public <NewTDocument> ReactiveMongoCollection<NewTDocument> withDocumentClass(Class<NewTDocument> clazz) {
         return new ReactiveMongoCollectionImpl<>(this.collection.withDocumentClass(clazz));
     }
+
+    @Override
+    public ReactiveMongoCollectionImpl<T> withReadPreference(ReadPreference readPreference) {
+        return new ReactiveMongoCollectionImpl<>(this.collection.withReadPreference(readPreference));
+    }
 }
diff --git a/...odb-client/runtime/src/main/java/io/quarkus/mongodb/reactive/ReactiveMongoCollection.java b/...odb-client/runtime/src/main/java/io/quarkus/mongodb/reactive/ReactiveMongoCollection.java
@@ -7,8 +7,8 @@
 import org.bson.conversions.Bson;
 
 import com.mongodb.MongoNamespace;
+import com.mongodb.ReadPreference;
 import com.mongodb.bulk.BulkWriteResult;
-import com.mongodb.client.MongoCollection;
 import com.mongodb.client.model.BulkWriteOptions;
 import com.mongodb.client.model.CountOptions;
 import com.mongodb.client.model.CreateIndexOptions;
@@ -1497,7 +1497,15 @@ Uni<Void> renameCollection(ClientSession clientSession, MongoNamespace newCollec
      *
      * @param clazz the default class to cast any documents returned from the database into.
      * @param <NewTDocument> The type that the new collection will encode documents from and decode documents to
-     * @return a new MongoCollection instance with the different default class
+     * @return a new ReactiveMongoCollection instance with the different default class
      */
     <NewTDocument> ReactiveMongoCollection<NewTDocument> withDocumentClass(Class<NewTDocument> clazz);
+
+    /**
+     * Create a new ReactiveMongoCollection instance with a different read preference.
+     *
+     * @param readPreference the new {@link com.mongodb.ReadPreference} for the collection
+     * @return a new ReactiveMongoCollection instance with the different readPreference
+     */
+    ReactiveMongoCollection<T> withReadPreference(ReadPreference readPreference);
 }
diff --git a/...main/java/io/quarkus/mongodb/panache/reactive/runtime/CommonReactivePanacheQueryImpl.java b/...main/java/io/quarkus/mongodb/panache/reactive/runtime/CommonReactivePanacheQueryImpl.java
@@ -7,6 +7,7 @@
 import org.bson.Document;
 import org.bson.conversions.Bson;
 
+import com.mongodb.ReadPreference;
 import com.mongodb.client.model.Collation;
 
 import io.quarkus.mongodb.FindOptions;
@@ -143,6 +144,11 @@ public <T extends Entity> CommonReactivePanacheQueryImpl<T> withCollation(Collat
         return (CommonReactivePanacheQueryImpl<T>) this;
     }
 
+    public <T extends Entity> CommonReactivePanacheQueryImpl<T> withReadPreference(ReadPreference readPreference) {
+        this.collection = this.collection.withReadPreference(readPreference);
+        return (CommonReactivePanacheQueryImpl<T>) this;
+    }
+
     // Results
 
     @SuppressWarnings("unchecked")

diff --git a/...mmon/runtime/src/main/java/io/quarkus/mongodb/panache/runtime/CommonPanacheQueryImpl.java b/...mmon/runtime/src/main/java/io/quarkus/mongodb/panache/runtime/CommonPanacheQueryImpl.java
@@ -9,6 +9,7 @@
 import org.bson.Document;
 import org.bson.conversions.Bson;
 
+import com.mongodb.ReadPreference;
 import com.mongodb.client.FindIterable;
 import com.mongodb.client.MongoCollection;
 import com.mongodb.client.MongoCursor;
@@ -139,6 +140,11 @@ public <T extends Entity> CommonPanacheQueryImpl<T> withCollation(Collation coll
         return (CommonPanacheQueryImpl<T>) this;
     }
 
+    public <T extends Entity> CommonPanacheQueryImpl<T> withReadPreference(ReadPreference readPreference) {
+        this.collection = this.collection.withReadPreference(readPreference);
+        return (CommonPanacheQueryImpl<T>) this;
+    }
+
     // Results
 
     @SuppressWarnings("unchecked")

diff --git a/...ain/java/io/quarkus/mongodb/panache/kotlin/reactive/runtime/ReactivePanacheQueryImpl.java b/...ain/java/io/quarkus/mongodb/panache/kotlin/reactive/runtime/ReactivePanacheQueryImpl.java
@@ -4,6 +4,7 @@
 
 import org.bson.conversions.Bson;
 
+import com.mongodb.ReadPreference;
 import com.mongodb.client.model.Collation;
 
 import io.quarkus.mongodb.panache.kotlin.reactive.ReactivePanacheQuery;
@@ -98,6 +99,12 @@ public ReactivePanacheQuery<Entity> withCollation(Collation collation) {
         return this;
     }
 
+    @Override
+    public ReactivePanacheQuery<Entity> withReadPreference(ReadPreference readPreference) {
+        delegate.withReadPreference(readPreference);
+        return this;
+    }
+
     @Override
     public Uni<Long> count() {
         return delegate.count();

diff --git a/...lin/runtime/src/main/java/io/quarkus/mongodb/panache/kotlin/runtime/PanacheQueryImpl.java b/...lin/runtime/src/main/java/io/quarkus/mongodb/panache/kotlin/runtime/PanacheQueryImpl.java
@@ -5,6 +5,7 @@
 
 import org.bson.conversions.Bson;
 
+import com.mongodb.ReadPreference;
 import com.mongodb.client.MongoCollection;
 import com.mongodb.client.model.Collation;
 
@@ -95,6 +96,12 @@ public PanacheQuery<Entity> withCollation(Collation collation) {
         return this;
     }
 
+    @Override
+    public PanacheQuery<Entity> withReadPreference(ReadPreference readPreference) {
+        delegate.withReadPreference(readPreference);
+        return this;
+    }
+
     // Results
 
     @Override

diff --git a/...-panache-kotlin/runtime/src/main/kotlin/io/quarkus/mongodb/panache/kotlin/PanacheQuery.kt b/...-panache-kotlin/runtime/src/main/kotlin/io/quarkus/mongodb/panache/kotlin/PanacheQuery.kt
@@ -1,5 +1,6 @@
 package io.quarkus.mongodb.panache.kotlin
 
+import com.mongodb.ReadPreference
 import com.mongodb.client.model.Collation
 import io.quarkus.panache.common.Page
 import io.quarkus.panache.common.exception.PanacheQueryException
@@ -137,6 +138,15 @@ interface PanacheQuery<Entity: Any> {
      * @return this query, modified
      */
     fun withCollation(collation: Collation): PanacheQuery<Entity>
+
+    /**
+     * Define the read preference used for this query.
+     *
+     * @param readPreference the read preference to be used for this query.
+     * @return this query, modified
+     */
+    fun withReadPreference(readPreference: ReadPreference?): PanacheQuery<Entity>
+
     // Results
     /**
      * Reads and caches the total number of entities this query operates on. This causes a database

diff --git a/...untime/src/main/kotlin/io/quarkus/mongodb/panache/kotlin/reactive/ReactivePanacheQuery.kt b/...untime/src/main/kotlin/io/quarkus/mongodb/panache/kotlin/reactive/ReactivePanacheQuery.kt
@@ -1,5 +1,6 @@
 package io.quarkus.mongodb.panache.kotlin.reactive
 
+import com.mongodb.ReadPreference
 import com.mongodb.client.model.Collation
 import io.quarkus.panache.common.Page
 import io.smallrye.mutiny.Multi
@@ -135,6 +136,14 @@ interface ReactivePanacheQuery<Entity> {
      */
     fun withCollation(collation: Collation): ReactivePanacheQuery<Entity>
 
+    /**
+     * Define the read preference used for this query.
+     *
+     * @param readPreference the read preference to be used for this query.
+     * @return this query, modified
+     */
+    fun withReadPreference(readPreference: ReadPreference): ReactivePanacheQuery<Entity>
+
     /**
      * Reads and caches the total number of entities this query operates on. This causes a database
      * query with `SELECT COUNT(*)` and a query equivalent to the current query, minus

diff --git a/...anache/mongodb-panache/runtime/src/main/java/io/quarkus/mongodb/panache/PanacheQuery.java b/...anache/mongodb-panache/runtime/src/main/java/io/quarkus/mongodb/panache/PanacheQuery.java
@@ -4,6 +4,7 @@
 import java.util.Optional;
 import java.util.stream.Stream;
 
+import com.mongodb.ReadPreference;
 import com.mongodb.client.model.Collation;
 
 import io.quarkus.panache.common.Page;
@@ -144,6 +145,14 @@ public interface PanacheQuery<Entity> {
      */
     public <T extends Entity> PanacheQuery<T> withCollation(Collation collation);
 
+    /**
+     * Define the read preference used for this query.
+     *
+     * @param readPreference the read preference to be used for this query.
+     * @return this query, modified
+     */
+    public <T extends Entity> PanacheQuery<T> withReadPreference(ReadPreference readPreference);
+
     // Results
 
     /**

diff --git a/...nache/runtime/src/main/java/io/quarkus/mongodb/panache/reactive/ReactivePanacheQuery.java b/...nache/runtime/src/main/java/io/quarkus/mongodb/panache/reactive/ReactivePanacheQuery.java
@@ -4,6 +4,7 @@
 import java.util.Optional;
 import java.util.stream.Stream;
 
+import com.mongodb.ReadPreference;
 import com.mongodb.client.model.Collation;
 
 import io.quarkus.panache.common.Page;
@@ -146,6 +147,14 @@ public interface ReactivePanacheQuery<Entity> {
      */
     public <T extends Entity> ReactivePanacheQuery<T> withCollation(Collation collation);
 
+    /**
+     * Define the read preference used for this query.
+     *
+     * @param readPreference the read preference to be used for this query.
+     * @return this query, modified
+     */
+    public <T extends Entity> ReactivePanacheQuery<T> withReadPreference(ReadPreference readPreference);
+
     // Results
 
     /**

diff --git a/...e/src/main/java/io/quarkus/mongodb/panache/reactive/runtime/ReactivePanacheQueryImpl.java b/...e/src/main/java/io/quarkus/mongodb/panache/reactive/runtime/ReactivePanacheQueryImpl.java
@@ -5,6 +5,7 @@
 
 import org.bson.conversions.Bson;
 
+import com.mongodb.ReadPreference;
 import com.mongodb.client.model.Collation;
 
 import io.quarkus.mongodb.panache.reactive.ReactivePanacheQuery;
@@ -98,6 +99,12 @@ public <T extends Entity> ReactivePanacheQuery<T> withCollation(Collation collat
         return (ReactivePanacheQuery<T>) this;
     }
 
+    @Override
+    public <T extends Entity> ReactivePanacheQuery<T> withReadPreference(ReadPreference readPreference) {
+        delegate.withReadPreference(readPreference);
+        return (ReactivePanacheQuery<T>) this;
+    }
+
     @Override
     public Uni<Long> count() {
         return delegate.count();

diff --git a/...db-panache/runtime/src/main/java/io/quarkus/mongodb/panache/runtime/PanacheQueryImpl.java b/...db-panache/runtime/src/main/java/io/quarkus/mongodb/panache/runtime/PanacheQueryImpl.java
@@ -6,6 +6,7 @@
 
 import org.bson.conversions.Bson;
 
+import com.mongodb.ReadPreference;
 import com.mongodb.client.MongoCollection;
 import com.mongodb.client.model.Collation;
 
@@ -97,6 +98,12 @@ public <T extends Entity> PanacheQuery<T> withCollation(Collation collation) {
         return (PanacheQuery<T>) this;
     }
 
+    @Override
+    public <T extends Entity> PanacheQuery<T> withReadPreference(ReadPreference readPreference) {
+        delegate.withReadPreference(readPreference);
+        return (PanacheQuery<T>) this;
+    }
+
     // Results
 
     @Override

diff --git a/...in/src/main/kotlin/io/quarkus/it/mongodb/panache/person/resources/PersonEntityResource.kt b/...in/src/main/kotlin/io/quarkus/it/mongodb/panache/person/resources/PersonEntityResource.kt
@@ -1,5 +1,6 @@
 package io.quarkus.it.mongodb.panache.person.resources
 
+import com.mongodb.ReadPreference
 import io.quarkus.it.mongodb.panache.person.PersonEntity
 import io.quarkus.it.mongodb.panache.person.PersonName
 import io.quarkus.it.mongodb.panache.person.Status
@@ -27,6 +28,7 @@ class PersonEntityResource {
     fun searchPersons(@PathParam("name") name: String): Set<PersonName> {
         return PersonEntity.find("lastname = ?1 and status = ?2", name, Status.ALIVE)
                 .project(PersonName::class.java)
+                .withReadPreference(ReadPreference.primaryPreferred())
                 .list()
                 .toSet()
     }

diff --git a/...rc/main/kotlin/io/quarkus/it/mongodb/panache/person/resources/PersonRepositoryResource.kt b/...rc/main/kotlin/io/quarkus/it/mongodb/panache/person/resources/PersonRepositoryResource.kt
@@ -1,5 +1,6 @@
 package io.quarkus.it.mongodb.panache.person.resources
 
+import com.mongodb.ReadPreference
 import io.quarkus.it.mongodb.panache.person.MockablePersonRepository
 import io.quarkus.it.mongodb.panache.person.Person
 import io.quarkus.it.mongodb.panache.person.PersonName
@@ -39,6 +40,7 @@ class PersonRepositoryResource {
     fun searchPersons(@PathParam("name") name: String): Set<PersonName> {
         return personRepository.find("lastname = ?1 and status = ?2", name, Status.ALIVE)
                 .project(PersonName::class.java)
+                .withReadPreference(ReadPreference.primaryPreferred())
                 .list()
                 .toSet()
     }

diff --git a/...n/io/quarkus/it/mongodb/panache/reactive/person/resources/ReactivePersonEntityResource.kt b/...n/io/quarkus/it/mongodb/panache/reactive/person/resources/ReactivePersonEntityResource.kt
@@ -1,6 +1,7 @@
 package io.quarkus.it.mongodb.panache.reactive.person.resources
 
 
+import com.mongodb.ReadPreference
 import io.quarkus.it.mongodb.panache.person.PersonName
 import io.quarkus.it.mongodb.panache.reactive.person.ReactivePersonEntity
 import io.quarkus.panache.common.Sort
@@ -31,6 +32,7 @@ class ReactivePersonEntityResource {
         val uniqueNames = mutableSetOf<PersonName>()
         val lastnames: List<PersonName> = ReactivePersonEntity.find("lastname", name)
                 .project(PersonName::class.java)
+                .withReadPreference(ReadPreference.primaryPreferred())
                 .list()
                 .await()
                 .indefinitely()

diff --git a/.../quarkus/it/mongodb/panache/reactive/person/resources/ReactivePersonRepositoryResource.kt b/.../quarkus/it/mongodb/panache/reactive/person/resources/ReactivePersonRepositoryResource.kt
@@ -1,6 +1,7 @@
 package io.quarkus.it.mongodb.panache.reactive.person.resources
 
 
+import com.mongodb.ReadPreference
 import io.quarkus.it.mongodb.panache.person.Person
 import io.quarkus.it.mongodb.panache.person.PersonName
 import io.quarkus.it.mongodb.panache.reactive.person.ReactivePersonRepository
@@ -36,6 +37,7 @@ class ReactivePersonRepositoryResource {
         val uniqueNames = mutableSetOf<PersonName>()
         val lastnames: List<PersonName> = reactivePersonRepository.find("lastname", name)
                 .project(PersonName::class.java)
+                .withReadPreference(ReadPreference.primaryPreferred())
                 .list()
                 .await()
                 .indefinitely()

diff --git a/...he/src/main/java/io/quarkus/it/mongodb/panache/person/resources/PersonEntityResource.java b/...he/src/main/java/io/quarkus/it/mongodb/panache/person/resources/PersonEntityResource.java
@@ -8,6 +8,8 @@
 import javax.ws.rs.*;
 import javax.ws.rs.core.Response;
 
+import com.mongodb.ReadPreference;
+
 import io.quarkus.it.mongodb.panache.person.PersonEntity;
 import io.quarkus.it.mongodb.panache.person.PersonName;
 import io.quarkus.it.mongodb.panache.person.Status;
@@ -29,6 +31,7 @@ public Set<PersonName> searchPersons(@PathParam("name") String name) {
         Set<PersonName> uniqueNames = new HashSet<>();
         List<PersonName> lastnames = PersonEntity.find("lastname = ?1 and status = ?2", name, Status.ALIVE)
                 .project(PersonName.class)
+                .withReadPreference(ReadPreference.primaryPreferred())
                 .list();
         lastnames.forEach(p -> uniqueNames.add(p));// this will throw if it's not the right type
         return uniqueNames;

diff --git a/...rc/main/java/io/quarkus/it/mongodb/panache/person/resources/PersonRepositoryResource.java b/...rc/main/java/io/quarkus/it/mongodb/panache/person/resources/PersonRepositoryResource.java
@@ -9,6 +9,8 @@
 import javax.ws.rs.*;
 import javax.ws.rs.core.Response;
 
+import com.mongodb.ReadPreference;
+
 import io.quarkus.it.mongodb.panache.person.MockablePersonRepository;
 import io.quarkus.it.mongodb.panache.person.Person;
 import io.quarkus.it.mongodb.panache.person.PersonName;
@@ -40,6 +42,7 @@ public Set<PersonName> searchPersons(@PathParam("name") String name) {
         Set<PersonName> uniqueNames = new HashSet<>();
         List<PersonName> lastnames = personRepository.find("lastname = ?1 and status = ?2", name, Status.ALIVE)
                 .project(PersonName.class)
+                .withReadPreference(ReadPreference.primaryPreferred())
                 .list();
         lastnames.forEach(p -> uniqueNames.add(p));// this will throw if it's not the right type
         return uniqueNames;

diff --git a/...io/quarkus/it/mongodb/panache/reactive/person/resources/ReactivePersonEntityResource.java b/...io/quarkus/it/mongodb/panache/reactive/person/resources/ReactivePersonEntityResource.java
@@ -8,6 +8,8 @@
 import javax.ws.rs.*;
 import javax.ws.rs.core.Response;
 
+import com.mongodb.ReadPreference;
+
 import io.quarkus.it.mongodb.panache.person.PersonName;
 import io.quarkus.it.mongodb.panache.reactive.person.ReactivePersonEntity;
 import io.quarkus.panache.common.Sort;
@@ -27,8 +29,11 @@ public Uni<List<ReactivePersonEntity>> getPersons(@QueryParam("sort") String sor
     @Path("/search/{name}")
     public Set<PersonName> searchPersons(@PathParam("name") String name) {
         Set<PersonName> uniqueNames = new HashSet<>();
-        List<PersonName> lastnames = ReactivePersonEntity.find("lastname", name).project(PersonName.class).list().await()
-                .indefinitely();
+        List<PersonName> lastnames = ReactivePersonEntity.find("lastname", name)
+                .project(PersonName.class)
+                .withReadPreference(ReadPreference.primaryPreferred())
+                .list()
+                .await().indefinitely();
         lastnames.forEach(p -> uniqueNames.add(p));// this will throw if it's not the right type
         return uniqueNames;
     }