Skip to content

Commit 3f09b17

Browse files
christophstroblchristophstrobl
committed
Decouple Bulk from BulkOperations.
For bulk write operations we decoupled the Bulk from the actual execution. This eliminates the need to go through a dedicated BulkOperations interface and enables direct use via MongoOperations. In difference to the existing BulkOperations, Lifecycle events and conversions are triggered on calling the bulk write method on MongoOperations. In case of ongoing transaction the active ClientSession is passed on to method calls on MongoClient. See: #5087 Original Pull Request: #5169
1 parent 3ef4cae commit 3f09b17

30 files changed

Lines changed: 4181 additions & 89 deletions

‎spring-data-mongodb/src/main/java/org/springframework/data/mongodb/MongoClusterCapable.java‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
1+
/*
2+
* Copyright 2026-present the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package org.springframework.data.mongodb;
17+
18+
/**
19+
* Interface that can provide access to a MongoDB cluster.
20+
*
21+
* @param <T> the MongoDB cluster/client type (e.g. {@link com.mongodb.client.MongoCluster} or
22+
* {@link com.mongodb.reactivestreams.client.MongoCluster}).
23+
* @author Christoph Strobl
24+
* @since 5.1
25+
*/
26+
public interface MongoClusterCapable<T> {
27+
28+
/**
29+
* Returns the MongoDB cluster used by this factory.
30+
*
31+
* @return the cluster; never {@literal null}.
32+
* @throws IllegalStateException if cluster cannot be obtained.
33+
*/
34+
T getMongoCluster();
35+
}

‎spring-data-mongodb/src/main/java/org/springframework/data/mongodb/SessionAwareMethodInterceptor.java‎

Lines changed: 15 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,7 @@ public class SessionAwareMethodInterceptor<D, C> implements MethodInterceptor {
5454
private final ClientSessionOperator databaseDecorator;
5555
private final Object target;
5656
private final Class<?> targetType;
57+
private final Class<?> clientType;
5758
private final Class<?> collectionType;
5859
private final Class<?> databaseType;
5960
private final Class<? extends ClientSession> sessionType;
@@ -63,15 +64,17 @@ public class SessionAwareMethodInterceptor<D, C> implements MethodInterceptor {
6364
*
6465
* @param session the {@link ClientSession} to be used on invocation.
6566
* @param target the original target object.
66-
* @param databaseType the MongoDB database type
67+
* @param clientType the MongoDB cluster/client type (e.g. {@link com.mongodb.client.MongoCluster}).
68+
* @param sessionType the {@link ClientSession} type.
69+
* @param databaseType the MongoDB database type.
6770
* @param databaseDecorator a {@link ClientSessionOperator} used to create the proxy for an imperative / reactive
6871
* {@code MongoDatabase}.
6972
* @param collectionType the MongoDB collection type.
7073
* @param collectionDecorator a {@link ClientSessionOperator} used to create the proxy for an imperative / reactive
7174
* {@code MongoCollection}.
7275
* @param <T> target object type.
7376
*/
74-
public <T> SessionAwareMethodInterceptor(ClientSession session, T target, Class<? extends ClientSession> sessionType,
77+
public <T> SessionAwareMethodInterceptor(ClientSession session, T target, Class<?> clientType, Class<? extends ClientSession> sessionType,
7578
Class<D> databaseType, ClientSessionOperator<D> databaseDecorator, Class<C> collectionType,
7679
ClientSessionOperator<C> collectionDecorator) {
7780

@@ -85,15 +88,24 @@ public <T> SessionAwareMethodInterceptor(ClientSession session, T target, Class<
8588

8689
this.session = session;
8790
this.target = target;
91+
this.clientType = ClassUtils.getUserClass(clientType);
8892
this.databaseType = ClassUtils.getUserClass(databaseType);
8993
this.collectionType = ClassUtils.getUserClass(collectionType);
9094
this.collectionDecorator = collectionDecorator;
9195
this.databaseDecorator = databaseDecorator;
9296

93-
this.targetType = ClassUtils.isAssignable(databaseType, target.getClass()) ? databaseType : collectionType;
97+
this.targetType = targetType(target.getClass());
9498
this.sessionType = sessionType;
9599
}
96100

101+
Class<?> targetType(@Nullable Class<?> targetType) {
102+
103+
if(ClassUtils.isAssignable(clientType, targetType)) {
104+
return clientType;
105+
}
106+
return ClassUtils.isAssignable(databaseType, targetType) ? databaseType : collectionType;
107+
}
108+
97109
@Override
98110
public @Nullable Object invoke(MethodInvocation methodInvocation) throws Throwable {
99111

‎spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/BulkWriteSupport.java‎

Lines changed: 163 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,163 @@
1+
/*
2+
* Copyright 2026-present the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package org.springframework.data.mongodb.core;
17+
18+
import java.util.List;
19+
20+
import org.bson.Document;
21+
import org.springframework.util.ClassUtils;
22+
23+
import com.mongodb.MongoNamespace;
24+
import com.mongodb.client.model.DeleteManyModel;
25+
import com.mongodb.client.model.DeleteOneModel;
26+
import com.mongodb.client.model.DeleteOptions;
27+
import com.mongodb.client.model.ReplaceOneModel;
28+
import com.mongodb.client.model.ReplaceOptions;
29+
import com.mongodb.client.model.UpdateManyModel;
30+
import com.mongodb.client.model.UpdateOneModel;
31+
import com.mongodb.client.model.UpdateOptions;
32+
import com.mongodb.client.model.WriteModel;
33+
import com.mongodb.client.model.bulk.ClientDeleteManyOptions;
34+
import com.mongodb.client.model.bulk.ClientDeleteOneOptions;
35+
import com.mongodb.client.model.bulk.ClientNamespacedWriteModel;
36+
import com.mongodb.client.model.bulk.ClientReplaceOneOptions;
37+
import com.mongodb.client.model.bulk.ClientUpdateManyOptions;
38+
import com.mongodb.client.model.bulk.ClientUpdateOneOptions;
39+
40+
/**
41+
* @author Christoph Strobl
42+
*/
43+
abstract class BulkWriteSupport {
44+
45+
static WriteModel<Document> updateMany(Document query, Object update, UpdateOptions updateOptions) {
46+
47+
if (update instanceof List<?> pipeline) {
48+
return new UpdateManyModel<>(query, (List<Document>) pipeline, updateOptions);
49+
} else if (update instanceof Document updateDocument) {
50+
return new UpdateManyModel<>(query, updateDocument, updateOptions);
51+
} else {
52+
throw new IllegalArgumentException(
53+
"Update needs to be either a List or a Document, but was [%s]".formatted(ClassUtils.getUserClass(update)));
54+
}
55+
}
56+
57+
static WriteModel<Document> updateOne(Document query, Object update, UpdateOptions updateOptions) {
58+
59+
if (update instanceof List<?> pipeline) {
60+
return new UpdateOneModel<>(query, (List<Document>) pipeline, updateOptions);
61+
} else if (update instanceof Document updateDocument) {
62+
return new UpdateOneModel<>(query, updateDocument, updateOptions);
63+
} else {
64+
throw new IllegalArgumentException(
65+
"Update needs to be either a List or a Document, but was [%s]".formatted(ClassUtils.getUserClass(update)));
66+
}
67+
}
68+
69+
static WriteModel<Document> removeMany(Document query, DeleteOptions deleteOptions) {
70+
return new DeleteManyModel<>(query, deleteOptions);
71+
}
72+
73+
static WriteModel<Document> removeOne(Document query, DeleteOptions deleteOptions) {
74+
return new DeleteOneModel<>(query, deleteOptions);
75+
}
76+
77+
static WriteModel<Document> replaceOne(Document query, Document replacement, UpdateOptions updateOptions) {
78+
79+
ReplaceOptions replaceOptions = new ReplaceOptions();
80+
replaceOptions.collation(updateOptions.getCollation());
81+
replaceOptions.upsert(updateOptions.isUpsert());
82+
replaceOptions.sort(updateOptions.getSort());
83+
replaceOptions.hint(updateOptions.getHint());
84+
replaceOptions.hintString(updateOptions.getHintString());
85+
86+
return new ReplaceOneModel<>(query, replacement, replaceOptions);
87+
}
88+
89+
static ClientNamespacedWriteModel updateMany(MongoNamespace namespace, Document query, Object update,
90+
UpdateOptions updateOptions) {
91+
92+
ClientUpdateManyOptions updateManyOptions = ClientUpdateManyOptions.clientUpdateManyOptions();
93+
updateManyOptions.arrayFilters(updateOptions.getArrayFilters());
94+
updateManyOptions.collation(updateOptions.getCollation());
95+
updateManyOptions.upsert(updateOptions.isUpsert());
96+
updateManyOptions.hint(updateOptions.getHint());
97+
updateManyOptions.hintString(updateOptions.getHintString());
98+
99+
if (update instanceof List<?> pipeline) {
100+
return ClientNamespacedWriteModel.updateMany(namespace, query, (List<Document>) pipeline, updateManyOptions);
101+
} else if (update instanceof Document updateDocument) {
102+
return ClientNamespacedWriteModel.updateMany(namespace, query, updateDocument, updateManyOptions);
103+
} else {
104+
throw new IllegalArgumentException(
105+
"Update needs to be either a List or a Document, but was [%s]".formatted(ClassUtils.getUserClass(update)));
106+
}
107+
}
108+
109+
static ClientNamespacedWriteModel updateOne(MongoNamespace namespace, Document query, Object update,
110+
UpdateOptions updateOptions) {
111+
112+
ClientUpdateOneOptions updateOneOptions = ClientUpdateOneOptions.clientUpdateOneOptions();
113+
updateOneOptions.sort(updateOptions.getSort());
114+
updateOneOptions.arrayFilters(updateOptions.getArrayFilters());
115+
updateOneOptions.collation(updateOptions.getCollation());
116+
updateOneOptions.upsert(updateOptions.isUpsert());
117+
updateOneOptions.hint(updateOptions.getHint());
118+
updateOneOptions.hintString(updateOptions.getHintString());
119+
120+
if (update instanceof List<?> pipeline) {
121+
return ClientNamespacedWriteModel.updateOne(namespace, query, (List<Document>) pipeline, updateOneOptions);
122+
} else if (update instanceof Document updateDocument) {
123+
return ClientNamespacedWriteModel.updateOne(namespace, query, updateDocument, updateOneOptions);
124+
} else {
125+
throw new IllegalArgumentException(
126+
"Update needs to be either a List or a Document, but was [%s]".formatted(ClassUtils.getUserClass(update)));
127+
}
128+
}
129+
130+
static ClientNamespacedWriteModel removeMany(MongoNamespace namespace, Document query, DeleteOptions deleteOptions) {
131+
132+
ClientDeleteManyOptions clientDeleteManyOptions = ClientDeleteManyOptions.clientDeleteManyOptions();
133+
clientDeleteManyOptions.collation(deleteOptions.getCollation());
134+
clientDeleteManyOptions.hint(deleteOptions.getHint());
135+
clientDeleteManyOptions.hintString(deleteOptions.getHintString());
136+
137+
return ClientNamespacedWriteModel.deleteMany(namespace, query, clientDeleteManyOptions);
138+
}
139+
140+
static ClientNamespacedWriteModel removeOne(MongoNamespace namespace, Document query, DeleteOptions deleteOptions) {
141+
142+
ClientDeleteOneOptions clientDeleteOneOptions = ClientDeleteOneOptions.clientDeleteOneOptions();
143+
// TODO: open an issue with MongoDB to enable sort for deleteOne
144+
clientDeleteOneOptions.collation(deleteOptions.getCollation());
145+
clientDeleteOneOptions.hint(deleteOptions.getHint());
146+
clientDeleteOneOptions.hintString(deleteOptions.getHintString());
147+
148+
return ClientNamespacedWriteModel.deleteOne(namespace, query, clientDeleteOneOptions);
149+
}
150+
151+
static ClientNamespacedWriteModel replaceOne(MongoNamespace namespace, Document query, Document replacement,
152+
UpdateOptions updateOptions) {
153+
154+
ClientReplaceOneOptions replaceOptions = ClientReplaceOneOptions.clientReplaceOneOptions();
155+
replaceOptions.sort(updateOptions.getSort());
156+
replaceOptions.upsert(updateOptions.isUpsert());
157+
replaceOptions.hint(updateOptions.getHint());
158+
replaceOptions.hintString(updateOptions.getHintString());
159+
replaceOptions.collation(updateOptions.getCollation());
160+
161+
return ClientNamespacedWriteModel.replaceOne(namespace, query, replacement, replaceOptions);
162+
}
163+
}

0 commit comments

Comments
 (0)