Как использовать строку JSON агрегатов MongoDB в качестве аргумента «arg0» в Spring Data ChangeStreamOptions.builder().f ⇐ JAVA

[Anonymous]

Я реализовал Reactive Async REST API с Spring Web Flux и MongoDB Change Streams, который работает правильно. Но вместо передачи объекта агрегации с критериями в метод ".filter(arg0)" я хотел бы передать в метод фильтра строку агрегирования MongoDB, например:

Код: Выделить всё

ChangeStreamOptions.builder()
.filter(Document.parse("[{$match: {age: {$gt: 18}}}, {'operationType': 'insert'}]"))
.returnFullDocumentOnUpdate().build();

Приведенный выше фрагмент кода не работает, поскольку он неверен.

Вот моя рабочая реализация ниже.
И метод

Код: Выделить всё

public Flux
 watchAgeGreaterThan(Integer ageParam){...}

вызывается через конечную точку REST "/api/rest/age/greaterthan/{ageParam}" для асинхронного возврата потока каждый раз. время, когда документ вставлен в MongoDB «person_collection» и его поле «age» больше, чем «ageParam».

Код: Выделить всё

import org.springframework.data.mongodb.core.ChangeStreamEvent;
import org.springframework.data.mongodb.core.ChangeStreamOptions;
import org.springframework.data.mongodb.core.ReactiveMongoTemplate;
import org.springframework.data.mongodb.core.aggregation.Aggregation;
import org.springframework.data.mongodb.core.query.Criteria;

@Service
public class Watch {

@Autowired
private ReactiveMongoTemplate reactiveMongoTemplate;

public Flux
 watchAgeGreaterThan(Integer ageParam) {

ChangeStreamOptions options = ChangeStreamOptions.builder()
.filter(Aggregation.newAggregation(Person.class,
Aggregation.match(
Criteria.where("operationType").is("insert")
.and("fullDocument.age").gt(ageParam)
))
).returnFullDocumentOnUpdate().build();

return reactiveMongoTemplate.changeStream("person_collection", options, Person.class)
.map(ChangeStreamEvent::getBody)
.doOnError(throwable -> log.error("Error on 'person' change stream event :: "
+ throwable.getMessage(), throwable));
}

}

Я попробовал еще одну реализацию, которую вы можете увидеть ниже:

Код: Выделить всё

public class CustomAggregationOperation implements AggregationOperation {
private String aggregationJsonString;

public CustomAggregationOperation(String aggregationJsonString) {
this.aggregationJsonString = aggregationJsonString;
}

@Override
public Document toDocument(AggregationOperationContext context) {
return context.getMappedObject(Document.parse(aggregationJsonString));
}
}

@Service
public class Watch {

@Autowired
private ReactiveMongoTemplate reactiveMongoTemplate;

public Flux
 watchAgeGreaterThan(Integer ageParam) {
String queryJsonString = "{$match: {age: {$gt: 18}}}";

AggregationOperation matchOperationType = Aggregation.match(Criteria.where("operationType").is("insert"));

TypedAggregation aggregation = Aggregation.newAggregation(Person.class,
matchOperationType, new CustomAggregationOperation(queryJsonString));

ChangeStreamOptions options2 = ChangeStreamOptions.builder().filter(aggregation).returnFullDocumentOnUpdate().build();

return reactiveMongoTemplate.changeStream("person_collection", options2, Person.class)
.map(ChangeStreamEvent::getBody)
.doOnError(throwable -> log.error("Error on 'person' change stream event :: "
+ throwable.getMessage(), throwable));
}
}

Но этот не работает.


Подробнее здесь: https://stackoverflow.com/questions/603 ... ng-data-ch