Learn how to migrate some Java code from spring-data-mongodb to pure java mongodb-driver-reactivestreams from example code marked Example 1 and Migration 1 and then migrate the snippet of code marked

To migrate the Java code from spring-data-mongodb to pure java mongodb-driver-reactivestreams, we need to replace the relevant classes and methods from spring-data-mongodb with their corresponding ones from mongodb-driver-reactivestreams.

For example, in Migration 1, we replace the use of mongoTemplate.aggregate with the following code:

Flux.from(dbClient.getCollection(kapokMongoProvider.getMongoBackendCollectionName(projectId)).aggregate(Arrays.asList(
    ...
))).allowDiskUse(true).first()).blockLast())

Here, we use dbClient.getCollection to get a reference to the collection we want to aggregate on, and then use the aggregate method to execute the aggregation pipeline. We also use Flux.from to convert the AggregateIterable returned by aggregate to a Flux, and then use allowDiskUse to enable disk usage for large aggregation results, and first and blockLast to retrieve the first result and block until it's available.

Similarly, we can replace other spring-data-mongodb classes and methods with their corresponding ones from mongodb-driver-reactivestreams, such as Aggregation, AggregationOptions, Aggregates, Filters, Accumulators, Projections, ArrayOperators, ComparisonOperators, ConditionalOperators, Direction, AggregationOperation, and AggregationOperationContext.

For example, in Translate, we can use the following code to migrate the aggregation pipeline:

Flux.from(dbClient.getCollection(kapokMongoProvider.getMongoBreadcrumbsCollectionName(projectId)).aggregate(Arrays.asList(
    KapokLogPresenterApiController.optionalMatchSubmoduleId(submoduleId),

    Aggregates.match(
        Filters.and(
            Filters.eq("category", "xhr"),
            Filters.exists("data.url"),
            Filters.ne("data.url", ""),
            Filters.exists("timestamp"),
            Filters.gt("timestamp", (double)(OffsetDateTime.now().minusDays(1).toInstant().toEpochMilli()) / 1000)
        )
    ),

    Aggregates.lookup(kapokMongoProvider.getMongoBreadcrumbsCollectionName(projectId), "event_id", "event_id", "breadcrumbObjects"),

    Aggregates.unwind("$breadcrumbObjects"),

    Aggregates.match(
        new Document("$expr",
            new Document("$gt", Arrays.asList("$timestamp", "$breadcrumbObjects.timestamp"))
        )
    ),

    Aggregates.sort(Sorts.descending("breadcrumbObjects.timestamp")),
    Aggregates.match(Filters.eq("breadcrumbObjects.category", "navigation")),
    Aggregates.group(new Document("_id", "$_id"))
        .first("timestamp").as("timestamp")
        .first("data.url").as("url")
        .first("breadcrumbObjects").as("breadcrumbObject"),

    Aggregates.match(
        Filters.and(
            Filters.exists("breadcrumbObject.data.to"),
            Filters.ne("breadcrumbObject.data.to", "")
        )
    ),

    Aggregates.group(new Document("timestamp", "$timestamp").append("url", "$url"))
        .first("breadcrumbObject.data.to").as("path"),

    Aggregates.group(null)
        .count().as("pathNum")
        .sum("callNum").as("totalCallNum"),

    Aggregates.project(Projections.fields(
        Projections.excludeId(),
        Projections.computed("callPerPage", new Document("$divide", Arrays.asList("$totalCallNum", "$pathNum")))
    ))
)).blockLast()

Here, we use Filters and Sorts to construct filter and sort criteria, and use Aggregates to construct aggregation pipeline stages. We also use new Document to construct BSON documents for some pipeline stages, and use Projections to construct projection documents for the final stage. Lastly, we use blockLast to block until the aggregation result is available.