Repository: quarkusio/quarkus
Issue: Null Pointer Exception Attempting to Watch Change Stream in Database with ChangeStreamOptions

### Describe the bug

If you supply a ChangeStreamOptions parameter to the watch() method of a Reactive Mongo Database for example in order to resume a change stream after a certain token the watch method returns null.

This appears to be an oversight in the implementation, as the Collection equivalent works as expected.

I.e. this works:

```
this.mongoClient
                .getDatabase(databaseName)
                .getCollection(collectionName)
                .watch(pipeline, Document.class, options);
```

But this returns null:
```
this.mongoClient
                .getDatabase(databaseName)
                .watch(pipeline, Document.class, options);
```


### Expected behavior

Supplying a ChangeStreamOptions parameter to the watch method, e.g.:

```
this.mongoClient
                .getDatabase(databaseName)
                .watch(pipeline, Document.class, options);
```

Would create a watch on the change stream of the database, and return a valid Multi.

### Actual behavior

The method returns null.

### How to Reproduce?

To reproduce:

```
        ChangeStreamOptions options = new ChangeStreamOptions();
        var watch = this.reactiveMongoClient
                .getDatabase(databaseName)
                .getCollection(collectionName)
                .watch(Document.class, options);
        Assertions.notNull("Change Stream Watch", watch);
```



### Output of `uname -a` or `ver`

Darwin ip-192-168-1-4.eu-central-1.compute.internal 21.5.0 Darwin Kernel Version 21.5.0: Tue Apr 26 21:08:29 PDT 2022; root:xnu-8020.121.3~4/RELEASE_ARM64_T8101 arm64

### Output of `java -version`

java version "17.0.1" 2021-10-19 LTS Java(TM) SE Runtime Environment (build 17.0.1+12-LTS-39) Java HotSpot(TM) 64-Bit Server VM (build 17.0.1+12-LTS-39, mixed mode, sharing)

### GraalVM version (if different from Java)

_No response_

### Quarkus version or git rev

2.11.3.Final

### Build tool (ie. output of `mvnw --version` or `gradlew --version`)

Apache Maven 3.8.6 (84538c9988a25aec085021c365c560670ad80f63) Maven home: /Users/lthompson/.m2/wrapper/dists/apache-maven-3.8.6-bin/67568434/apache-maven-3.8.6 Java version: 17.0.1, vendor: Oracle Corporation, runtime: /Library/Java/JavaVirtualMachines/jdk-17.0.1.jdk/Contents/Home Default locale: en_AU, platform encoding: UTF-8 OS name: "mac os x", version: "12.4", arch: "aarch64", family: "mac"

### Additional information

_No response_

Changed files:
extensions/mongodb-client/runtime/src/main/java/io/quarkus/mongodb/impl/ReactiveMongoDatabaseImpl.java

diff --git a/extensions/mongodb-client/runtime/src/main/java/io/quarkus/mongodb/impl/ReactiveMongoDatabaseImpl.java b/extensions/mongodb-client/runtime/src/main/java/io/quarkus/mongodb/impl/ReactiveMongoDatabaseImpl.java
index 42756a44562..fc99cdeebcb 100644
--- a/extensions/mongodb-client/runtime/src/main/java/io/quarkus/mongodb/impl/ReactiveMongoDatabaseImpl.java
+++ b/extensions/mongodb-client/runtime/src/main/java/io/quarkus/mongodb/impl/ReactiveMongoDatabaseImpl.java
@@ -10,6 +10,7 @@
 import com.mongodb.client.model.CreateViewOptions;
 import com.mongodb.client.model.changestream.ChangeStreamDocument;
 import com.mongodb.reactivestreams.client.AggregatePublisher;
+import com.mongodb.reactivestreams.client.ChangeStreamPublisher;
 import com.mongodb.reactivestreams.client.ClientSession;
 import com.mongodb.reactivestreams.client.ListCollectionsPublisher;
 import com.mongodb.reactivestreams.client.MongoDatabase;
@@ -217,7 +218,14 @@ public <T> Multi<ChangeStreamDocument<T>> watch(Class<T> clazz) {
 
     @Override
     public <T> Multi<ChangeStreamDocument<T>> watch(Class<T> clazz, ChangeStreamOptions options) {
-        return null;
+        return Wrappers.toMulti(apply(options, database.watch(clazz)));
+    }
+
+    private <D> ChangeStreamPublisher<D> apply(ChangeStreamOptions options, ChangeStreamPublisher<D> watch) {
+        if (options == null) {
+            return watch;
+        }
+        return options.apply(watch);
     }
 
     @Override
@@ -227,7 +235,7 @@ public Multi<ChangeStreamDocument<Document>> watch(List<? extends Bson> pipeline
 
     @Override
     public Multi<ChangeStreamDocument<Document>> watch(List<? extends Bson> pipeline, ChangeStreamOptions options) {
-        return null;
+        return Wrappers.toMulti(apply(options, database.watch(pipeline)));
     }
 
     @Override
@@ -238,7 +246,7 @@ public <T> Multi<ChangeStreamDocument<T>> watch(List<? extends Bson> pipeline, C
     @Override
     public <T> Multi<ChangeStreamDocument<T>> watch(List<? extends Bson> pipeline, Class<T> clazz,
             ChangeStreamOptions options) {
-        return null;
+        return Wrappers.toMulti(apply(options, database.watch(pipeline, clazz)));
     }
 
     @Override
@@ -248,7 +256,7 @@ public Multi<ChangeStreamDocument<Document>> watch(ClientSession clientSession)
 
     @Override
     public Multi<ChangeStreamDocument<Document>> watch(ClientSession clientSession, ChangeStreamOptions options) {
-        return null;
+        return Wrappers.toMulti(apply(options, database.watch(clientSession)));
     }
 
     @Override
@@ -259,7 +267,7 @@ public <T> Multi<ChangeStreamDocument<T>> watch(ClientSession clientSession, Cla
     @Override
     public <T> Multi<ChangeStreamDocument<T>> watch(ClientSession clientSession, Class<T> clazz,
             ChangeStreamOptions options) {
-        return null;
+        return Wrappers.toMulti(apply(options, database.watch(clientSession, clazz)));
     }
 
     @Override
@@ -270,7 +278,7 @@ public Multi<ChangeStreamDocument<Document>> watch(ClientSession clientSession,
     @Override
     public Multi<ChangeStreamDocument<Document>> watch(ClientSession clientSession, List<? extends Bson> pipeline,
             ChangeStreamOptions options) {
-        return null;
+        return Wrappers.toMulti(apply(options, database.watch(clientSession, pipeline)));
     }
 
     @Override
@@ -282,7 +290,7 @@ public <T> Multi<ChangeStreamDocument<T>> watch(ClientSession clientSession, Lis
     @Override
     public <T> Multi<ChangeStreamDocument<T>> watch(ClientSession clientSession, List<? extends Bson> pipeline,
             Class<T> clazz, ChangeStreamOptions options) {
-        return null;
+        return Wrappers.toMulti(apply(options, database.watch(clientSession, pipeline, clazz)));
     }
 
     @Override