Custom aggregation with WSO2EI and MongoDB

Question

I'm tring to implement aggregation operations for Mongo data service in WSO2DSS (in WSO2EI) since only the basic operations like CRUD and Count are supported out of the box as mentioned here. So I cloned WSO2EI code version 4.4.10 (our team happens to be using this version) and I've successfully added some of my own functionality. However, whenever I try to use org.jongo.MongoCollection.aggregate() function, I get error Error in MongoQuery.runQuery: Command failed with error 9: 'The 'cursor' option is required, except for aggregate with the explain argument'. I checked all the posts regarding this issue and tried to use different versions of Mongo driver (3.4, 3.6..), also changed the actual syntax a lot but no matter what I try, I always get this error if I use the .aggregate() function. Also should I be using org.jongo.MongoCollection.aggregate(String pipelineOperator) or com.mongodb.async.client.aggregate(java.util.List<? extends Bson> pipeline) as stated in the Mongo Java 3.6 documentation here. Example code I used:

private Iterator<MongoTestClass> doAggregate1(MongoCollection collection, String opQuery, Object[] parameters) throws DataServiceFault {

        return collection.aggregate("{$match:{componentType:'Endpoint'}}")
                .as(MongoTestClass.class).iterator();

    }

where componentType is existing field in my MongoDB collection document and 'Endpoint' is it's value. Is my actual synatax wrong? Or is there any other problem? How can I add the 'cursor' to my query so the error goes away?

I can't wrap my head around how this works...much appreciate any help.

Answer 1

From the docs.

MongoDB 3.4 deprecates the use of aggregate command without the cursor option, unless the pipeline includes the explain option. When returning aggregation results inline using the aggregate command, specify the cursor option using the default batch size cursor: {} or specify the batch size in the cursor option cursor: { batchSize: }.

You can pass batchSize with AggregationOptions for Jongo Aggregate method.

AggregationOptions options = AggregationOptions.builder().options.
     batchSize(100).
     outputMode(AggregationOptions.OutputMode.CURSOR).build();

collection.aggregate("{$match:{componentType:'Endpoint'}}").options(options).as(MongoTestClass.class).iterator();

With default batch size

AggregationOptions options = AggregationOptions.builder().options.
     outputMode(AggregationOptions.OutputMode.CURSOR).build();

OR

AggregationOptions options = AggregationOptions.builder().build();

collection.aggregate("{$match:{componentType:'Endpoint'}}").options(options).as(MongoTestClass.class).iterator();