* mongo: fixed MongoCollection.aggregate() pipeline logging

Signed-off-by: neo <[email protected]>

Author: neowu

CHANGELOG.md

@@ -9,6 +9,7 @@
 * ws: remove ws support
   > websocket is not used anymore, use sse/ajax instead
 * kafka: updated to 4.0.0
+* mongo: fixed MongoCollection.aggregate() pipeline logging
 
 ### 9.1.7 (2/26/2025 - 3/6/2025)
 

core-ng-mongo/src/main/java/core/framework/mongo/impl/BsonsLogParam.java

@@ -0,0 +1,37 @@
+package core.framework.mongo.impl;
+
+import core.framework.internal.log.filter.LogParam;
+import org.bson.codecs.configuration.CodecRegistry;
+import org.bson.conversions.Bson;
+
+import java.util.List;
+import java.util.Set;
+
+/**
+ * @author neo
+ */
+class BsonsLogParam implements LogParam {
+    private final List<Bson> bsons;
+    private final CodecRegistry registry;
+
+    BsonsLogParam(List<Bson> bsons, CodecRegistry registry) {
+        this.bsons = bsons;
+        this.registry = registry;
+    }
+
+    @Override
+    public void append(StringBuilder builder, Set<String> maskedFields, int maxParamLength) {
+        builder.append('[');
+        boolean first = true;
+        for (Bson bson : bsons) {
+            if (first) {
+                first = false;
+            } else {
+                builder.append(", ");
+            }
+            new BsonLogParam(bson, registry)
+                .append(builder, maskedFields, maxParamLength);
+        }
+        builder.append(']');
+    }
+}

core-ng-mongo/src/main/java/core/framework/mongo/impl/MongoCollectionImpl.java

@@ -36,7 +36,6 @@
 import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
-import java.util.stream.Collectors;
 
 /**
  * @author neo
@@ -229,7 +228,7 @@ public <V> List<V> aggregate(Aggregate<V> aggregate) {
             int size = results.size();
             logger.debug("aggregate, collection={}, pipeline={}, readPref={}, returnedDocs={}, elapsed={}",
                 collectionName,
-                aggregate.pipeline.stream().map(stage -> new BsonLogParam(stage, mongo.registry)).collect(Collectors.toList()),
+                new BsonsLogParam(aggregate.pipeline, mongo.registry),
                 aggregate.readPreference == null ? null : aggregate.readPreference.getName(),
                 size,
                 elapsed);
@@ -320,7 +319,11 @@ public long delete(Bson filter) {
             return deletedRows;
         } finally {
             long elapsed = watch.elapsed();
-            logger.debug("delete, collection={}, filter={}, deletedRows={}, elapsed={}", collectionName, new BsonLogParam(filter, mongo.registry), deletedRows, elapsed);
+            logger.debug("delete, collection={}, filter={}, deletedRows={}, elapsed={}",
+                collectionName,
+                new BsonLogParam(filter, mongo.registry),
+                deletedRows,
+                elapsed);
             ActionLogContext.track("mongo", elapsed, 0, (int) deletedRows);
         }
     }