diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveFindOperation.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveFindOperation.java index 8dbc357d3..ff9d0c57c 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveFindOperation.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveFindOperation.java @@ -85,6 +85,21 @@ public interface ReactiveFindOperation { */ Flux all(); + /** + * Get all matching elements using a tailable cursor. + *

+ * The stream may become dead, or invalid, if either the query returns no match or the cursor returns the document + * at the "end" of the collection and then the application deletes that document. + *

+ * A stream that is no longer in use must be {@link reactor.core.Disposable#dispose()} disposed} otherwise the + * streams will linger and exhaust resources.
+ * NOTE: Requires a capped collection. + * + * @return never {@literal null}. + * @since 2.1 + */ + Flux tail(); + /** * Get the number of matching elements. * diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveFindOperationSupport.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveFindOperationSupport.java index 90aa9f2de..dc540c2c2 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveFindOperationSupport.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveFindOperationSupport.java @@ -169,6 +169,15 @@ class ReactiveFindOperationSupport implements ReactiveFindOperation { return doFind(null); } + /* + * (non-Javadoc) + * @see org.springframework.data.mongodb.core.ReactiveFindOperation.TerminatingFind#tail() + */ + @Override + public Flux tail() { + return doFind(template.new TailingQueryFindPublisherPreparer(query, domainType)); + } + /* * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveFindOperation.FindWithQuery#near(org.springframework.data.mongodb.core.query.NearQuery) diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveFindOperationSupportTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveFindOperationSupportTests.java index a3218392b..dcd475a58 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveFindOperationSupportTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveFindOperationSupportTests.java @@ -22,9 +22,14 @@ import static org.springframework.data.mongodb.core.query.Query.*; import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; +import reactor.core.Disposable; +import reactor.core.publisher.Flux; import reactor.test.StepVerifier; import java.util.Date; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; import java.util.function.Consumer; import org.bson.BsonString; @@ -41,6 +46,7 @@ import org.springframework.data.mongodb.core.index.GeoSpatialIndexType; import org.springframework.data.mongodb.core.index.GeospatialIndex; import org.springframework.data.mongodb.core.mapping.Field; import org.springframework.data.mongodb.core.query.BasicQuery; +import org.springframework.data.mongodb.core.query.Criteria; import org.springframework.data.mongodb.core.query.NearQuery; import com.mongodb.MongoClient; @@ -67,6 +73,12 @@ public class ReactiveFindOperationSupportTests { blocking = new MongoTemplate(new SimpleMongoDbFactory(new MongoClient(), "ExecutableFindOperationSupportTests")); blocking.dropCollection(STAR_WARS); + insertObjects(); + + template = new ReactiveMongoTemplate(MongoClients.create(), "ExecutableFindOperationSupportTests"); + } + + void insertObjects() { han = new Person(); han.firstname = "han"; han.lastname = "solo"; @@ -79,8 +91,6 @@ public class ReactiveFindOperationSupportTests { blocking.save(han); blocking.save(luke); - - template = new ReactiveMongoTemplate(MongoClients.create(), "ExecutableFindOperationSupportTests"); } @Test(expected = IllegalArgumentException.class) // DATAMONGO-1719 @@ -291,6 +301,84 @@ public class ReactiveFindOperationSupportTests { .verifyComplete(); } + @Test // DATAMONGO-2080 + public void tail() throws InterruptedException { + + blocking.dropCollection(STAR_WARS); + blocking.createCollection(STAR_WARS, CollectionOptions.empty().capped().size(1024 * 1024)); + insertObjects(); + + BlockingQueue collector = new LinkedBlockingQueue<>(); + Flux tail = template.query(Person.class) + .matching(query(new Criteria().orOperator(where("firstname").is("chewbacca"), where("firstname").is("luke")))) + .tail().doOnNext(collector::add); + + Disposable subscription = tail.subscribe(); + + assertThat(collector.poll(1, TimeUnit.SECONDS)).isEqualTo(luke); + assertThat(collector).isEmpty(); + + Person chewbacca = new Person(); + chewbacca.firstname = "chewbacca"; + chewbacca.lastname = "chewie"; + chewbacca.id = "id-3"; + + blocking.save(chewbacca); + + assertThat(collector.poll(10, TimeUnit.SECONDS)).isEqualTo(chewbacca); + + subscription.dispose(); + } + + @Test // DATAMONGO-2080 + public void tailWithProjection() { + + blocking.dropCollection(STAR_WARS); + blocking.createCollection(STAR_WARS, CollectionOptions.empty().capped().size(1024 * 1024)); + insertObjects(); + + StepVerifier + .create(template.query(Person.class).as(Jedi.class).matching(query(where("firstname").is("luke"))).tail()) + .consumeNextWith(it -> assertThat(it).isInstanceOf(Jedi.class)) // + .thenCancel() // + .verify(); + } + + @Test // DATAMONGO-2080 + public void tailWithClosedInterfaceProjection() { + + blocking.dropCollection(STAR_WARS); + blocking.createCollection(STAR_WARS, CollectionOptions.empty().capped().size(1024 * 1024)); + insertObjects(); + + StepVerifier.create( + template.query(Person.class).as(PersonProjection.class).matching(query(where("firstname").is("luke"))).all()) + .consumeNextWith(it -> { + + assertThat(it).isInstanceOf(PersonProjection.class); + assertThat(it.getFirstname()).isEqualTo("luke"); + }) // + .thenCancel() // + .verify(); + } + + @Test // DATAMONGO-2080 + public void tailWithOpenInterfaceProjection() { + + blocking.dropCollection(STAR_WARS); + blocking.createCollection(STAR_WARS, CollectionOptions.empty().capped().size(1024 * 1024)); + insertObjects(); + + StepVerifier.create(template.query(Person.class).as(PersonSpELProjection.class) + .matching(query(where("firstname").is("luke"))).tail()).consumeNextWith(it -> { + + assertThat(it).isInstanceOf(PersonSpELProjection.class); + assertThat(it.getName()).isEqualTo("luke"); + }) // + .thenCancel() // + .verify(); + } + @Test // DATAMONGO-1719 public void firstShouldReturnFirstEntryInCollection() { StepVerifier.create(template.query(Person.class).first()).expectNextCount(1).verifyComplete();