ReactiveMongoTemplate does not return ChangeStreamEvents with PreImages

[PhilippAllstadt]

Hi @ALL,

in our use case we have the requirement to consume changestream events with pre-images. We are using MongoDB version 6.0.7.
Further, our spring version is the newest and we are using the latest mongodb drivers.

Despite it, delete and update events haven´t filled fullDocumentBeforeChange in raw if event is consumed in our spring app.

In database. pre-images should work because we´ve tested consuming via mongosh shell.

This is our code to consume change stream in our spring app:

var collectionChangeStream = reactiveMongoTemplate .changeStream(TransactionEntity.class) .withOptions( (optionsBuilder) -> optionsBuilder .returnFullDocumentBeforeChange() .returnFullDocumentOnUpdate()) .watchCollection("transaction");

if (changeStreamToken.isPresent() && changeStreamToken.get().getToken() != null) {
  collectionChangeStream.resumeAfter(changeStreamToken.get().getToken());
} else {
  collectionChangeStream.resumeAt(Instant.now().minusSeconds(seconds));
}

return collectionChangeStream.listen();

Am I missing something?

Thank you very much.

[christophstrobl]

Thank you @PhilippAllstadt for bringing this to our attention.
At a quick glance it seems ReactiveMongoTemplate#changeStream(String, String, ChangeStreamOptions, Class) does not forward the FullDocumentBeforeChange option to the ChangeStreamPublisher.

[PhilippAllstadt]

Thank you for your response.

Yes, i think the problem is prepareFilter method:

`public Flux<ChangeStreamEvent> changeStream(@nullable String database, @nullable String collectionName,
ChangeStreamOptions options, Class targetType) {

	List<Document> filter = prepareFilter(options);
	FullDocument fullDocument = ClassUtils.isAssignable(Document.class, targetType) ? FullDocument.DEFAULT
			: FullDocument.UPDATE_LOOKUP;

	return ReactiveMongoDatabaseUtils.getDatabase(database, mongoDatabaseFactory) //
			.map(db -> {
				ChangeStreamPublisher<Document> publisher;
				if (StringUtils.hasText(collectionName)) {
					publisher = filter.isEmpty() ? db.getCollection(collectionName).watch(Document.class)
							: db.getCollection(collectionName).watch(filter, Document.class);

				} else {
					publisher = filter.isEmpty() ? db.watch(Document.class) : db.watch(filter, Document.class);
				}

				if (options.isResumeAfter()) {
					publisher = options.getResumeToken().map(BsonValue::asDocument).map(publisher::resumeAfter)
							.orElse(publisher);
				} else if (options.isStartAfter()) {
					publisher = options.getResumeToken().map(BsonValue::asDocument).map(publisher::startAfter)
							.orElse(publisher);
				}
				publisher = options.getCollation().map(Collation::toMongoCollation).map(publisher::collation)
						.orElse(publisher);
				publisher = options.getResumeBsonTimestamp().map(publisher::startAtOperationTime).orElse(publisher);
				return publisher.fullDocument(options.getFullDocumentLookup().orElse(fullDocument));
			}) //
			.flatMapMany(publisher -> Flux.from(publisher)
					.map(document -> new ChangeStreamEvent<>(document, targetType, getConverter())));
}`

prepareFilter:
`

	Object filter = options.getFilter().orElse(Collections.emptyList());

	if (filter instanceof Aggregation agg) {
		AggregationOperationContext context = agg instanceof TypedAggregation typedAggregation
				? new TypeBasedAggregationOperationContext(typedAggregation.getInputType(),
						getConverter().getMappingContext(), queryMapper)
				: new RelaxedTypeBasedAggregationOperationContext(Object.class, mappingContext, queryMapper);

		return agg.toPipeline(new PrefixingDelegatingAggregationOperationContext(context, "fullDocument",
				Arrays.asList("operationType", "fullDocument", "documentKey", "updateDescription", "ns")));
	}

	if (filter instanceof List) {
		return (List<Document>) filter;
	}

	throw new IllegalArgumentException(
			"ChangeStreamRequestOptions.filter mut be either an Aggregation or a plain list of Documents");

`

Changestreamoptions aren´t type of agg and returned list is empty.

So, would you fix that by the next release?

Thank you very much

[Alfagun74]

Oh my god, I've been searching for the cause of this problem for a long time! I'm also having this issue!

[PhilippAllstadt]

Hi there, is there any update to this issue ? This feature would be very important for our use case.

Thank you.