JAVA-5767 Support $lookup in CSFLE and QE

driver-core/src/test/resources/client-side-encryption-data/lookup/key-doc.json

@@ -0,0 +1,30 @@
+{
+    "_id": {
+        "$binary": {
+            "base64": "EjRWeBI0mHYSNBI0VniQEg==",
+            "subType": "04"
+        }
+    },
+    "keyMaterial": {
+        "$binary": {
+            "base64": "sHe0kz57YW7v8g9VP9sf/+K1ex4JqKc5rf/URX3n3p8XdZ6+15uXPaSayC6adWbNxkFskuMCOifDoTT+rkqMtFkDclOy884RuGGtUysq3X7zkAWYTKi8QAfKkajvVbZl2y23UqgVasdQu3OVBQCrH/xY00nNAs/52e958nVjBuzQkSb1T8pKJAyjZsHJ60+FtnfafDZSTAIBJYn7UWBCwQ==",
+            "subType": "00"
+        }
+    },
+    "creationDate": {
+        "$date": {
+            "$numberLong": "1648914851981"
+        }
+    },
+    "updateDate": {
+        "$date": {
+            "$numberLong": "1648914851981"
+        }
+    },
+    "status": {
+        "$numberInt": "0"
+    },
+    "masterKey": {
+        "provider": "local"
+    }
+}

driver-core/src/test/resources/client-side-encryption-data/lookup/schema-csfle.json

@@ -0,0 +1,19 @@
+{
+    "properties": {
+        "csfle": {
+            "encrypt": {
+                "keyId": [
+                    {
+                        "$binary": {
+                            "base64": "EjRWeBI0mHYSNBI0VniQEg==",
+                            "subType": "04"
+                        }
+                    }
+                ],
+                "bsonType": "string",
+                "algorithm": "AEAD_AES_256_CBC_HMAC_SHA_512-Deterministic"
+            }
+        }
+    },
+    "bsonType": "object"
+}

driver-core/src/test/resources/client-side-encryption-data/lookup/schema-csfle2.json

@@ -0,0 +1,19 @@
+{
+    "properties": {
+        "csfle2": {
+            "encrypt": {
+                "keyId": [
+                    {
+                        "$binary": {
+                            "base64": "EjRWeBI0mHYSNBI0VniQEg==",
+                            "subType": "04"
+                        }
+                    }
+                ],
+                "bsonType": "string",
+                "algorithm": "AEAD_AES_256_CBC_HMAC_SHA_512-Deterministic"
+            }
+        }
+    },
+    "bsonType": "object"
+}

driver-core/src/test/resources/client-side-encryption-data/lookup/schema-qe.json

@@ -0,0 +1,20 @@
+{
+    "escCollection": "enxcol_.qe.esc",
+    "ecocCollection": "enxcol_.qe.ecoc",
+    "fields": [
+        {
+            "keyId": {
+                "$binary": {
+                    "base64": "EjRWeBI0mHYSNBI0VniQEg==",
+                    "subType": "04"
+                }
+            },
+            "path": "qe",
+            "bsonType": "string",
+            "queries": {
+                "queryType": "equality",
+                "contention": 0
+            }
+        }
+    ]
+}

driver-core/src/test/resources/client-side-encryption-data/lookup/schema-qe2.json

@@ -0,0 +1,20 @@
+{
+    "escCollection": "enxcol_.qe2.esc",
+    "ecocCollection": "enxcol_.qe2.ecoc",
+    "fields": [
+        {
+            "keyId": {
+                "$binary": {
+                    "base64": "EjRWeBI0mHYSNBI0VniQEg==",
+                    "subType": "04"
+                }
+            },
+            "path": "qe2",
+            "bsonType": "string",
+            "queries": {
+                "queryType": "equality",
+                "contention": 0
+            }
+        }
+    ]
+}

driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/crypt/CollectionInfoRetriever.java

@@ -20,7 +20,7 @@
 import com.mongodb.lang.Nullable;
 import com.mongodb.reactivestreams.client.MongoClient;
 import org.bson.BsonDocument;
-import reactor.core.publisher.Mono;
+import reactor.core.publisher.Flux;
 
 import static com.mongodb.assertions.Assertions.notNull;
 import static com.mongodb.reactivestreams.client.internal.TimeoutHelper.databaseWithTimeoutDeferred;
@@ -35,8 +35,8 @@ class CollectionInfoRetriever {
         this.client = notNull("client", client);
     }
 
-    public Mono<BsonDocument> filter(final String databaseName, final BsonDocument filter, @Nullable final Timeout operationTimeout) {
+    public Flux<BsonDocument> filter(final String databaseName, final BsonDocument filter, @Nullable final Timeout operationTimeout) {
         return databaseWithTimeoutDeferred(client.getDatabase(databaseName), TIMEOUT_ERROR_MESSAGE, operationTimeout)
-                .flatMap(database -> Mono.from(database.listCollections(BsonDocument.class).filter(filter).first()));
+                .flatMapMany(database -> Flux.from(database.listCollections(BsonDocument.class).filter(filter)));
     }
 }

driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/crypt/Crypt.java

@@ -304,10 +304,8 @@ private void collInfo(final MongoCryptContext cryptContext,
         } else {
             collectionInfoRetriever.filter(databaseName, cryptContext.getMongoOperation(), operationTimeout)
                     .contextWrite(sink.contextView())
-                    .doOnSuccess(result -> {
-                        if (result != null) {
-                            cryptContext.addMongoOperationResult(result);
-                        }
+                    .doOnNext(result -> cryptContext.addMongoOperationResult(result))
+                    .doOnComplete(() -> {
                         cryptContext.completeMongoOperation();
                         executeStateMachineWithSink(cryptContext, databaseName, sink, operationTimeout);
                     })

driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/ClientSideEncryption25Lookup.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.mongodb.reactivestreams.client;
+
+import com.mongodb.ClientEncryptionSettings;
+import com.mongodb.MongoClientSettings;
+import com.mongodb.client.MongoClient;
+import com.mongodb.client.vault.ClientEncryption;
+import com.mongodb.reactivestreams.client.syncadapter.SyncClientEncryption;
+import com.mongodb.reactivestreams.client.syncadapter.SyncMongoClient;
+import com.mongodb.reactivestreams.client.vault.ClientEncryptions;
+
+public class ClientSideEncryption25Lookup extends com.mongodb.client.ClientSideEncryption25Lookup {
+
+    @Override
+    protected MongoClient createMongoClient(final MongoClientSettings settings) {
+        return new SyncMongoClient(MongoClients.create(settings));
+    }
+
+    @Override
+    protected ClientEncryption createClientEncryption(final ClientEncryptionSettings settings) {
+        return new SyncClientEncryption(ClientEncryptions.create(settings));
+    }
+
+}

driver-sync/src/main/com/mongodb/client/internal/CollectionInfoRetriever.java

@@ -16,6 +16,7 @@
 
 package com.mongodb.client.internal;
 
+import com.mongodb.client.ListCollectionsIterable;
 import com.mongodb.client.MongoClient;
 import com.mongodb.internal.time.Timeout;
 import com.mongodb.lang.Nullable;
@@ -33,9 +34,8 @@ class CollectionInfoRetriever {
         this.client = notNull("client", client);
     }
 
-    @Nullable
-    public BsonDocument filter(final String databaseName, final BsonDocument filter, @Nullable final Timeout operationTimeout) {
-        return databaseWithTimeout(client.getDatabase(databaseName), TIMEOUT_ERROR_MESSAGE,
-                operationTimeout).listCollections(BsonDocument.class).filter(filter).first();
+    public ListCollectionsIterable<BsonDocument> filter(final String databaseName, final BsonDocument filter, @Nullable final Timeout operationTimeout) {
+        return databaseWithTimeout(client.getDatabase(databaseName), TIMEOUT_ERROR_MESSAGE, operationTimeout)
+                .listCollections(BsonDocument.class).filter(filter);
     }
 }

driver-sync/src/main/com/mongodb/client/internal/Crypt.java

@@ -19,6 +19,7 @@
 import com.mongodb.MongoClientException;
 import com.mongodb.MongoException;
 import com.mongodb.MongoInternalException;
+import com.mongodb.client.ListCollectionsIterable;
 import com.mongodb.client.MongoClient;
 import com.mongodb.client.model.vault.DataKeyOptions;
 import com.mongodb.client.model.vault.EncryptOptions;
@@ -41,6 +42,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
 import java.util.Map;
 import java.util.function.Supplier;
 
@@ -308,9 +310,10 @@ private void fetchCredentials(final MongoCryptContext cryptContext) {
 
     private void collInfo(final MongoCryptContext cryptContext, final String databaseName, @Nullable final Timeout operationTimeout) {
         try {
-            BsonDocument collectionInfo = assertNotNull(collectionInfoRetriever).filter(databaseName, cryptContext.getMongoOperation(), operationTimeout);
-            if (collectionInfo != null) {
-                cryptContext.addMongoOperationResult(collectionInfo);
+            ListCollectionsIterable<BsonDocument> results = assertNotNull(collectionInfoRetriever)
+                    .filter(databaseName, cryptContext.getMongoOperation(), operationTimeout);
+            for (BsonDocument result : results.into(new ArrayList<>())) {
+                cryptContext.addMongoOperationResult(result);
             }
             cryptContext.completeMongoOperation();
         } catch (Throwable t) {