This class is not part of the public API and may be removed or changed at any time
*/
public final class CommandMessage extends RequestMessage {
+ /**
+ * Specifies that the `OP_MSG` section payload is a BSON document.
+ */
+ private static final byte PAYLOAD_TYPE_0_DOCUMENT = 0;
+ /**
+ * Specifies that the `OP_MSG` section payload is a sequence of BSON documents.
+ */
+ private static final byte PAYLOAD_TYPE_1_DOCUMENT_SEQUENCE = 1;
+
private final MongoNamespace namespace;
private final BsonDocument command;
private final FieldNameValidator commandFieldNameValidator;
private final ReadPreference readPreference;
private final boolean exhaustAllowed;
- private final SplittablePayload payload;
- private final FieldNameValidator payloadFieldNameValidator;
+ private final MessageSequences sequences;
private final boolean responseExpected;
+ /**
+ * {@code null} iff either {@link #sequences} is not of the {@link DualMessageSequences} type,
+ * or it is of that type, but it has not been {@linkplain #encodeMessageBodyWithMetadata(ByteBufferBsonOutput, OperationContext) encoded}.
+ */
+ @Nullable
+ private Boolean dualMessageSequencesRequireResponse;
private final ClusterConnectionMode clusterConnectionMode;
private final ServerApi serverApi;
CommandMessage(final MongoNamespace namespace, final BsonDocument command, final FieldNameValidator commandFieldNameValidator,
final ReadPreference readPreference, final MessageSettings settings, final ClusterConnectionMode clusterConnectionMode,
@Nullable final ServerApi serverApi) {
- this(namespace, command, commandFieldNameValidator, readPreference, settings, true, null, null,
+ this(namespace, command, commandFieldNameValidator, readPreference, settings, true, EmptyMessageSequences.INSTANCE,
clusterConnectionMode, serverApi);
}
CommandMessage(final MongoNamespace namespace, final BsonDocument command, final FieldNameValidator commandFieldNameValidator,
final ReadPreference readPreference, final MessageSettings settings, final boolean exhaustAllowed,
final ClusterConnectionMode clusterConnectionMode, @Nullable final ServerApi serverApi) {
- this(namespace, command, commandFieldNameValidator, readPreference, settings, true, exhaustAllowed, null, null,
+ this(namespace, command, commandFieldNameValidator, readPreference, settings, true, exhaustAllowed, EmptyMessageSequences.INSTANCE,
clusterConnectionMode, serverApi);
}
CommandMessage(final MongoNamespace namespace, final BsonDocument command, final FieldNameValidator commandFieldNameValidator,
final ReadPreference readPreference, final MessageSettings settings, final boolean responseExpected,
- @Nullable final SplittablePayload payload, @Nullable final FieldNameValidator payloadFieldNameValidator,
+ final MessageSequences sequences,
final ClusterConnectionMode clusterConnectionMode, @Nullable final ServerApi serverApi) {
- this(namespace, command, commandFieldNameValidator, readPreference, settings, responseExpected, false, payload,
- payloadFieldNameValidator, clusterConnectionMode, serverApi);
+ this(namespace, command, commandFieldNameValidator, readPreference, settings, responseExpected, false,
+ sequences, clusterConnectionMode, serverApi);
}
CommandMessage(final MongoNamespace namespace, final BsonDocument command, final FieldNameValidator commandFieldNameValidator,
final ReadPreference readPreference, final MessageSettings settings,
final boolean responseExpected, final boolean exhaustAllowed,
- @Nullable final SplittablePayload payload, @Nullable final FieldNameValidator payloadFieldNameValidator,
+ final MessageSequences sequences,
final ClusterConnectionMode clusterConnectionMode, @Nullable final ServerApi serverApi) {
super(namespace.getFullName(), getOpCode(settings, clusterConnectionMode, serverApi), settings);
this.namespace = namespace;
@@ -108,9 +127,9 @@ public final class CommandMessage extends RequestMessage {
this.commandFieldNameValidator = commandFieldNameValidator;
this.readPreference = readPreference;
this.responseExpected = responseExpected;
+ this.dualMessageSequencesRequireResponse = null;
this.exhaustAllowed = exhaustAllowed;
- this.payload = payload;
- this.payloadFieldNameValidator = payloadFieldNameValidator;
+ this.sequences = sequences;
this.clusterConnectionMode = notNull("clusterConnectionMode", clusterConnectionMode);
this.serverApi = serverApi;
assertTrue(useOpMsg() || responseExpected);
@@ -119,8 +138,8 @@ public final class CommandMessage extends RequestMessage {
/**
* Create a BsonDocument representing the logical document encoded by an OP_MSG.
*
- * The returned document will contain all the fields from the Body (Kind 0) Section, as well as all fields represented by
- * OP_MSG Document Sequence (Kind 1) Sections.
+ * The returned document will contain all the fields from the `PAYLOAD_TYPE_0_DOCUMENT` section, as well as all fields represented by
+ * `PAYLOAD_TYPE_1_DOCUMENT_SEQUENCE` sections.
*/
BsonDocument getCommandDocument(final ByteBufferBsonOutput bsonOutput) {
List byteBuffers = bsonOutput.getByteBuffers();
@@ -130,14 +149,14 @@ BsonDocument getCommandDocument(final ByteBufferBsonOutput bsonOutput) {
byteBuf.position(getEncodingMetadata().getFirstDocumentPosition());
ByteBufBsonDocument byteBufBsonDocument = createOne(byteBuf);
- // If true, it means there is at least one Kind 1:Document Sequence in the OP_MSG
+ // If true, it means there is at least one `PAYLOAD_TYPE_1_DOCUMENT_SEQUENCE` section in the OP_MSG
if (byteBuf.hasRemaining()) {
BsonDocument commandBsonDocument = byteBufBsonDocument.toBaseBsonDocument();
// Each loop iteration processes one Document Sequence
// When there are no more bytes remaining, there are no more Document Sequences
while (byteBuf.hasRemaining()) {
- // skip reading the payload type, we know it is 1
+ // skip reading the payload type, we know it is `PAYLOAD_TYPE_1`
byteBuf.position(byteBuf.position() + 1);
int sequenceStart = byteBuf.position();
int sequenceSizeInBytes = byteBuf.getInt();
@@ -170,7 +189,7 @@ BsonDocument getCommandDocument(final ByteBufferBsonOutput bsonOutput) {
/**
* Get the field name from a buffer positioned at the start of the document sequence identifier of an OP_MSG Section of type
- * Document Sequence (Kind 1).
+ * `PAYLOAD_TYPE_1_DOCUMENT_SEQUENCE`.
*
* Upon normal completion of the method, the buffer will be positioned at the start of the first BSON object in the sequence.
*/
@@ -192,7 +211,15 @@ boolean isResponseExpected() {
if (responseExpected) {
return true;
} else {
- return payload != null && payload.isOrdered() && payload.hasAnotherSplit();
+ if (sequences instanceof SplittablePayload) {
+ SplittablePayload payload = (SplittablePayload) sequences;
+ return payload.isOrdered() && payload.hasAnotherSplit();
+ } else if (sequences instanceof DualMessageSequences) {
+ return assertNotNull(dualMessageSequencesRequireResponse);
+ } else if (!(sequences instanceof EmptyMessageSequences)) {
+ fail(sequences.toString());
+ }
+ return false;
}
}
@@ -201,47 +228,74 @@ MongoNamespace getNamespace() {
}
@Override
- protected EncodingMetadata encodeMessageBodyWithMetadata(final BsonOutput bsonOutput, final OperationContext operationContext) {
+ protected EncodingMetadata encodeMessageBodyWithMetadata(final ByteBufferBsonOutput bsonOutput, final OperationContext operationContext) {
+ int commandStartPosition = useOpMsg() ? writeOpMsg(bsonOutput, operationContext) : writeOpQuery(bsonOutput);
+ return new EncodingMetadata(commandStartPosition);
+ }
+
+ @SuppressWarnings("try")
+ private int writeOpMsg(final ByteBufferBsonOutput bsonOutput, final OperationContext operationContext) {
int messageStartPosition = bsonOutput.getPosition() - MESSAGE_PROLOGUE_LENGTH;
- int commandStartPosition;
- if (useOpMsg()) {
- int flagPosition = bsonOutput.getPosition();
- bsonOutput.writeInt32(0); // flag bits
- bsonOutput.writeByte(0); // payload type
- commandStartPosition = bsonOutput.getPosition();
-
- addDocument(command, bsonOutput, commandFieldNameValidator, getExtraElements(operationContext));
-
- if (payload != null) {
- bsonOutput.writeByte(1); // payload type
- int payloadBsonOutputStartPosition = bsonOutput.getPosition();
- bsonOutput.writeInt32(0); // size
- bsonOutput.writeCString(payload.getPayloadName());
- writePayload(new BsonBinaryWriter(bsonOutput, payloadFieldNameValidator), bsonOutput, getSettings(),
- messageStartPosition, payload, getSettings().getMaxDocumentSize());
-
- int payloadBsonOutputLength = bsonOutput.getPosition() - payloadBsonOutputStartPosition;
- bsonOutput.writeInt32(payloadBsonOutputStartPosition, payloadBsonOutputLength);
+ int flagPosition = bsonOutput.getPosition();
+ bsonOutput.writeInt32(0); // flag bits
+ bsonOutput.writeByte(PAYLOAD_TYPE_0_DOCUMENT);
+ int commandStartPosition = bsonOutput.getPosition();
+ List extraElements = getExtraElements(operationContext);
+
+ int commandDocumentSizeInBytes = writeDocument(command, bsonOutput, commandFieldNameValidator);
+ if (sequences instanceof SplittablePayload) {
+ appendElementsToDocument(bsonOutput, commandStartPosition, extraElements);
+ SplittablePayload payload = (SplittablePayload) sequences;
+ try (FinishOpMsgSectionWithPayloadType1 finishSection = startOpMsgSectionWithPayloadType1(
+ bsonOutput, payload.getPayloadName())) {
+ writePayload(
+ new BsonBinaryWriter(bsonOutput, payload.getFieldNameValidator()),
+ bsonOutput, getSettings(), messageStartPosition, payload, getSettings().getMaxDocumentSize());
}
-
- // Write the flag bits
- bsonOutput.writeInt32(flagPosition, getOpMsgFlagBits());
+ } else if (sequences instanceof DualMessageSequences) {
+ DualMessageSequences dualMessageSequences = (DualMessageSequences) sequences;
+ try (ByteBufferBsonOutput.Branch bsonOutputBranch2 = bsonOutput.branch();
+ ByteBufferBsonOutput.Branch bsonOutputBranch1 = bsonOutput.branch()) {
+ DualMessageSequences.EncodeDocumentsResult encodeDocumentsResult;
+ try (FinishOpMsgSectionWithPayloadType1 finishSection1 = startOpMsgSectionWithPayloadType1(
+ bsonOutputBranch1, dualMessageSequences.getFirstSequenceId());
+ FinishOpMsgSectionWithPayloadType1 finishSection2 = startOpMsgSectionWithPayloadType1(
+ bsonOutputBranch2, dualMessageSequences.getSecondSequenceId())) {
+ encodeDocumentsResult = writeDocumentsOfDualMessageSequences(
+ dualMessageSequences, commandDocumentSizeInBytes, bsonOutputBranch1,
+ bsonOutputBranch2, getSettings());
+ }
+ dualMessageSequencesRequireResponse = encodeDocumentsResult.isServerResponseRequired();
+ extraElements.addAll(encodeDocumentsResult.getExtraElements());
+ appendElementsToDocument(bsonOutput, commandStartPosition, extraElements);
+ }
+ } else if (sequences instanceof EmptyMessageSequences) {
+ appendElementsToDocument(bsonOutput, commandStartPosition, extraElements);
} else {
- bsonOutput.writeInt32(0);
- bsonOutput.writeCString(namespace.getFullName());
- bsonOutput.writeInt32(0);
- bsonOutput.writeInt32(-1);
+ fail(sequences.toString());
+ }
+
+ // Write the flag bits
+ bsonOutput.writeInt32(flagPosition, getOpMsgFlagBits());
+ return commandStartPosition;
+ }
- commandStartPosition = bsonOutput.getPosition();
+ private int writeOpQuery(final ByteBufferBsonOutput bsonOutput) {
+ bsonOutput.writeInt32(0);
+ bsonOutput.writeCString(namespace.getFullName());
+ bsonOutput.writeInt32(0);
+ bsonOutput.writeInt32(-1);
- List elements = null;
- if (serverApi != null) {
- elements = new ArrayList<>(3);
- addServerApiElements(elements);
- }
- addDocument(command, bsonOutput, commandFieldNameValidator, elements);
+ int commandStartPosition = bsonOutput.getPosition();
+
+ List elements = null;
+ if (serverApi != null) {
+ elements = new ArrayList<>(3);
+ addServerApiElements(elements);
}
- return new EncodingMetadata(commandStartPosition);
+ writeDocument(command, bsonOutput, commandFieldNameValidator);
+ appendElementsToDocument(bsonOutput, commandStartPosition, elements);
+ return commandStartPosition;
}
private int getOpMsgFlagBits() {
@@ -269,7 +323,7 @@ private List getExtraElements(final OperationContext operationConte
SessionContext sessionContext = operationContext.getSessionContext();
TimeoutContext timeoutContext = operationContext.getTimeoutContext();
- List extraElements = new ArrayList<>();
+ ArrayList extraElements = new ArrayList<>();
if (!getSettings().isCryptd()) {
timeoutContext.runMaxTimeMS(maxTimeMS ->
extraElements.add(new BsonElement("maxTimeMS", new BsonInt64(maxTimeMS)))
@@ -341,6 +395,19 @@ private void addReadConcernDocument(final List extraElements, final
}
}
+ /**
+ * @param sequenceId The identifier of the sequence contained in the {@code OP_MSG} section to be written.
+ * @see OP_MSG
+ */
+ private FinishOpMsgSectionWithPayloadType1 startOpMsgSectionWithPayloadType1(final ByteBufferBsonOutput bsonOutput, final String sequenceId) {
+ bsonOutput.writeByte(PAYLOAD_TYPE_1_DOCUMENT_SEQUENCE);
+ int sequenceStart = bsonOutput.getPosition();
+ // size to be patched back later
+ bsonOutput.writeInt32(0);
+ bsonOutput.writeCString(sequenceId);
+ return () -> backpatchLength(sequenceStart, bsonOutput);
+ }
+
private static OpCode getOpCode(final MessageSettings settings, final ClusterConnectionMode clusterConnectionMode,
@Nullable final ServerApi serverApi) {
return isServerVersionKnown(settings) || clusterConnectionMode == LOAD_BALANCED || serverApi != null
@@ -351,4 +418,9 @@ private static OpCode getOpCode(final MessageSettings settings, final ClusterCon
private static boolean isServerVersionKnown(final MessageSettings settings) {
return settings.getMaxWireVersion() >= FOUR_DOT_ZERO_WIRE_VERSION;
}
+
+ @FunctionalInterface
+ private interface FinishOpMsgSectionWithPayloadType1 extends AutoCloseable {
+ void close();
+ }
}
diff --git a/driver-core/src/main/com/mongodb/internal/connection/CommandProtocolImpl.java b/driver-core/src/main/com/mongodb/internal/connection/CommandProtocolImpl.java
index de9e0666d40..eb4d6d49516 100644
--- a/driver-core/src/main/com/mongodb/internal/connection/CommandProtocolImpl.java
+++ b/driver-core/src/main/com/mongodb/internal/connection/CommandProtocolImpl.java
@@ -26,17 +26,15 @@
import org.bson.FieldNameValidator;
import org.bson.codecs.Decoder;
-import static com.mongodb.assertions.Assertions.isTrueArgument;
import static com.mongodb.assertions.Assertions.notNull;
import static com.mongodb.internal.connection.ProtocolHelper.getMessageSettings;
class CommandProtocolImpl implements CommandProtocol {
private final MongoNamespace namespace;
private final BsonDocument command;
- private final SplittablePayload payload;
+ private final MessageSequences sequences;
private final ReadPreference readPreference;
private final FieldNameValidator commandFieldNameValidator;
- private final FieldNameValidator payloadFieldNameValidator;
private final Decoder commandResultDecoder;
private final boolean responseExpected;
private final ClusterConnectionMode clusterConnectionMode;
@@ -44,8 +42,7 @@ class CommandProtocolImpl implements CommandProtocol {
CommandProtocolImpl(final String database, final BsonDocument command, final FieldNameValidator commandFieldNameValidator,
@Nullable final ReadPreference readPreference, final Decoder commandResultDecoder, final boolean responseExpected,
- @Nullable final SplittablePayload payload, @Nullable final FieldNameValidator payloadFieldNameValidator,
- final ClusterConnectionMode clusterConnectionMode, final OperationContext operationContext) {
+ final MessageSequences sequences, final ClusterConnectionMode clusterConnectionMode, final OperationContext operationContext) {
notNull("database", database);
this.namespace = new MongoNamespace(notNull("database", database), MongoNamespace.COMMAND_COLLECTION_NAME);
this.command = notNull("command", command);
@@ -53,13 +50,9 @@ class CommandProtocolImpl implements CommandProtocol {
this.readPreference = readPreference;
this.commandResultDecoder = notNull("commandResultDecoder", commandResultDecoder);
this.responseExpected = responseExpected;
- this.payload = payload;
- this.payloadFieldNameValidator = payloadFieldNameValidator;
+ this.sequences = sequences;
this.clusterConnectionMode = notNull("clusterConnectionMode", clusterConnectionMode);
this.operationContext = operationContext;
-
- isTrueArgument("payloadFieldNameValidator cannot be null if there is a payload.",
- payload == null || payloadFieldNameValidator != null);
}
@Nullable
@@ -87,13 +80,13 @@ public void executeAsync(final InternalConnection connection, final SingleResult
@Override
public CommandProtocolImpl withSessionContext(final SessionContext sessionContext) {
return new CommandProtocolImpl<>(namespace.getDatabaseName(), command, commandFieldNameValidator, readPreference,
- commandResultDecoder, responseExpected, payload, payloadFieldNameValidator, clusterConnectionMode,
+ commandResultDecoder, responseExpected, sequences, clusterConnectionMode,
operationContext.withSessionContext(sessionContext));
}
private CommandMessage getCommandMessage(final InternalConnection connection) {
return new CommandMessage(namespace, command, commandFieldNameValidator, readPreference,
- getMessageSettings(connection.getDescription(), connection.getInitialServerDescription()), responseExpected, payload,
- payloadFieldNameValidator, clusterConnectionMode, operationContext.getServerApi());
+ getMessageSettings(connection.getDescription(), connection.getInitialServerDescription()), responseExpected,
+ sequences, clusterConnectionMode, operationContext.getServerApi());
}
}
diff --git a/driver-core/src/main/com/mongodb/internal/connection/CompressedMessage.java b/driver-core/src/main/com/mongodb/internal/connection/CompressedMessage.java
index 9880ef3fb0b..6764135daa1 100644
--- a/driver-core/src/main/com/mongodb/internal/connection/CompressedMessage.java
+++ b/driver-core/src/main/com/mongodb/internal/connection/CompressedMessage.java
@@ -17,7 +17,6 @@
package com.mongodb.internal.connection;
import org.bson.ByteBuf;
-import org.bson.io.BsonOutput;
import java.util.List;
@@ -37,7 +36,7 @@ class CompressedMessage extends RequestMessage {
}
@Override
- protected EncodingMetadata encodeMessageBodyWithMetadata(final BsonOutput bsonOutput, final OperationContext operationContext) {
+ protected EncodingMetadata encodeMessageBodyWithMetadata(final ByteBufferBsonOutput bsonOutput, final OperationContext operationContext) {
bsonOutput.writeInt32(wrappedOpcode.getValue());
bsonOutput.writeInt32(getWrappedMessageSize(wrappedMessageBuffers) - MESSAGE_HEADER_LENGTH);
bsonOutput.writeByte(compressor.getId());
diff --git a/driver-core/src/main/com/mongodb/internal/connection/Connection.java b/driver-core/src/main/com/mongodb/internal/connection/Connection.java
index 95094b240c1..219fb9ae6b9 100644
--- a/driver-core/src/main/com/mongodb/internal/connection/Connection.java
+++ b/driver-core/src/main/com/mongodb/internal/connection/Connection.java
@@ -51,7 +51,7 @@ T command(String database, BsonDocument command, FieldNameValidator fieldNam
@Nullable
T command(String database, BsonDocument command, FieldNameValidator commandFieldNameValidator,
@Nullable ReadPreference readPreference, Decoder commandResultDecoder, OperationContext operationContext,
- boolean responseExpected, @Nullable SplittablePayload payload, @Nullable FieldNameValidator payloadFieldNameValidator);
+ boolean responseExpected, MessageSequences sequences);
enum PinningMode {
diff --git a/driver-core/src/main/com/mongodb/internal/connection/DefaultServer.java b/driver-core/src/main/com/mongodb/internal/connection/DefaultServer.java
index 8f3d0f09fd9..008cdbefcb7 100644
--- a/driver-core/src/main/com/mongodb/internal/connection/DefaultServer.java
+++ b/driver-core/src/main/com/mongodb/internal/connection/DefaultServer.java
@@ -302,9 +302,9 @@ public T command(final String database, final BsonDocument command, final Fi
public T command(final String database, final BsonDocument command, final FieldNameValidator commandFieldNameValidator,
@Nullable final ReadPreference readPreference, final Decoder commandResultDecoder,
final OperationContext operationContext, final boolean responseExpected,
- @Nullable final SplittablePayload payload, @Nullable final FieldNameValidator payloadFieldNameValidator) {
+ final MessageSequences sequences) {
return wrapped.command(database, command, commandFieldNameValidator, readPreference, commandResultDecoder, operationContext,
- responseExpected, payload, payloadFieldNameValidator);
+ responseExpected, sequences);
}
@Override
@@ -364,10 +364,10 @@ public void commandAsync(final String database, final BsonDocument command,
@Override
public void commandAsync(final String database, final BsonDocument command, final FieldNameValidator commandFieldNameValidator,
@Nullable final ReadPreference readPreference, final Decoder commandResultDecoder,
- final OperationContext operationContext, final boolean responseExpected, @Nullable final SplittablePayload payload,
- @Nullable final FieldNameValidator payloadFieldNameValidator, final SingleResultCallback callback) {
+ final OperationContext operationContext, final boolean responseExpected, final MessageSequences sequences,
+ final SingleResultCallback callback) {
wrapped.commandAsync(database, command, commandFieldNameValidator, readPreference, commandResultDecoder,
- operationContext, responseExpected, payload, payloadFieldNameValidator, callback);
+ operationContext, responseExpected, sequences, callback);
}
@Override
diff --git a/driver-core/src/main/com/mongodb/internal/connection/DefaultServerConnection.java b/driver-core/src/main/com/mongodb/internal/connection/DefaultServerConnection.java
index 01d5f587fdc..143ef5b76ae 100644
--- a/driver-core/src/main/com/mongodb/internal/connection/DefaultServerConnection.java
+++ b/driver-core/src/main/com/mongodb/internal/connection/DefaultServerConnection.java
@@ -20,6 +20,7 @@
import com.mongodb.connection.ClusterConnectionMode;
import com.mongodb.connection.ConnectionDescription;
import com.mongodb.internal.async.SingleResultCallback;
+import com.mongodb.internal.connection.MessageSequences.EmptyMessageSequences;
import com.mongodb.internal.diagnostics.logging.Logger;
import com.mongodb.internal.diagnostics.logging.Loggers;
import com.mongodb.internal.session.SessionContext;
@@ -70,18 +71,17 @@ public ConnectionDescription getDescription() {
@Override
public T command(final String database, final BsonDocument command, final FieldNameValidator fieldNameValidator,
@Nullable final ReadPreference readPreference, final Decoder commandResultDecoder, final OperationContext operationContext) {
- return command(database, command, fieldNameValidator, readPreference, commandResultDecoder, operationContext, true, null, null);
+ return command(database, command, fieldNameValidator, readPreference, commandResultDecoder, operationContext, true, EmptyMessageSequences.INSTANCE);
}
@Nullable
@Override
public T command(final String database, final BsonDocument command, final FieldNameValidator commandFieldNameValidator,
@Nullable final ReadPreference readPreference, final Decoder commandResultDecoder,
- final OperationContext operationContext, final boolean responseExpected,
- @Nullable final SplittablePayload payload, @Nullable final FieldNameValidator payloadFieldNameValidator) {
+ final OperationContext operationContext, final boolean responseExpected, final MessageSequences sequences) {
return executeProtocol(
new CommandProtocolImpl<>(database, command, commandFieldNameValidator, readPreference, commandResultDecoder,
- responseExpected, payload, payloadFieldNameValidator, clusterConnectionMode, operationContext),
+ responseExpected, sequences, clusterConnectionMode, operationContext),
operationContext.getSessionContext());
}
@@ -90,16 +90,15 @@ public void commandAsync(final String database, final BsonDocument command,
@Nullable final ReadPreference readPreference, final Decoder commandResultDecoder, final OperationContext operationContext,
final SingleResultCallback callback) {
commandAsync(database, command, fieldNameValidator, readPreference, commandResultDecoder,
- operationContext, true, null, null, callback);
+ operationContext, true, EmptyMessageSequences.INSTANCE, callback);
}
@Override
public void commandAsync(final String database, final BsonDocument command, final FieldNameValidator commandFieldNameValidator,
@Nullable final ReadPreference readPreference, final Decoder commandResultDecoder, final OperationContext operationContext,
- final boolean responseExpected, @Nullable final SplittablePayload payload,
- @Nullable final FieldNameValidator payloadFieldNameValidator, final SingleResultCallback callback) {
+ final boolean responseExpected, final MessageSequences sequences, final SingleResultCallback callback) {
executeProtocolAsync(new CommandProtocolImpl<>(database, command, commandFieldNameValidator, readPreference,
- commandResultDecoder, responseExpected, payload, payloadFieldNameValidator, clusterConnectionMode, operationContext),
+ commandResultDecoder, responseExpected, sequences, clusterConnectionMode, operationContext),
operationContext.getSessionContext(), callback);
}
diff --git a/driver-core/src/main/com/mongodb/internal/connection/DualMessageSequences.java b/driver-core/src/main/com/mongodb/internal/connection/DualMessageSequences.java
new file mode 100644
index 00000000000..0c5a3430c22
--- /dev/null
+++ b/driver-core/src/main/com/mongodb/internal/connection/DualMessageSequences.java
@@ -0,0 +1,122 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.mongodb.internal.connection;
+
+import org.bson.BsonBinaryWriter;
+import org.bson.BsonElement;
+import org.bson.FieldNameValidator;
+
+import java.util.List;
+
+/**
+ * Two sequences that may either be coupled or independent.
+ *
+ * This class is not part of the public API and may be removed or changed at any time.
+ * May be called multiple times per {@link #encodeDocuments(WritersProviderAndLimitsChecker)}.
+ */
+ WriteResult tryWrite(WriteAction write);
+
+ /**
+ * @see #doAndGetBatchCount(BsonBinaryWriter, BsonBinaryWriter)
+ */
+ interface WriteAction {
+ /**
+ * Writes documents to the sequences using the provided writers.
+ *
+ * @return The resulting batch count since the beginning of {@link #encodeDocuments(WritersProviderAndLimitsChecker)}.
+ * It is generally allowed to be greater than {@link MessageSettings#getMaxBatchCount()}.
+ */
+ int doAndGetBatchCount(BsonBinaryWriter firstWriter, BsonBinaryWriter secondWriter);
+ }
+
+ enum WriteResult {
+ FAIL_LIMIT_EXCEEDED,
+ OK_LIMIT_REACHED,
+ OK_LIMIT_NOT_REACHED
+ }
+ }
+
+ public static final class EncodeDocumentsResult {
+ private final boolean serverResponseRequired;
+ private final List extraElements;
+
+ /**
+ * @param extraElements See {@link #getExtraElements()}.
+ */
+ public EncodeDocumentsResult(final boolean serverResponseRequired, final List extraElements) {
+ this.serverResponseRequired = serverResponseRequired;
+ this.extraElements = extraElements;
+ }
+
+ boolean isServerResponseRequired() {
+ return serverResponseRequired;
+ }
+
+ /**
+ * {@linkplain BsonElement Key/value pairs} to be added to the document contained in the {@code OP_MSG} section with payload type 0.
+ */
+ List getExtraElements() {
+ return extraElements;
+ }
+ }
+}
diff --git a/driver-core/src/main/com/mongodb/internal/connection/ElementExtendingBsonWriter.java b/driver-core/src/main/com/mongodb/internal/connection/ElementExtendingBsonWriter.java
deleted file mode 100644
index d0ed5234d50..00000000000
--- a/driver-core/src/main/com/mongodb/internal/connection/ElementExtendingBsonWriter.java
+++ /dev/null
@@ -1,57 +0,0 @@
-/*
- * Copyright 2008-present MongoDB, Inc.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.mongodb.internal.connection;
-
-import org.bson.BsonBinaryWriter;
-import org.bson.BsonElement;
-import org.bson.BsonReader;
-
-import java.util.List;
-
-import static com.mongodb.internal.connection.BsonWriterHelper.writeElements;
-
-/**
- *
This class is not part of the public API and may be removed or changed at any time
- */
-public class ElementExtendingBsonWriter extends LevelCountingBsonWriter {
- private final BsonBinaryWriter writer;
- private final List extraElements;
-
-
- public ElementExtendingBsonWriter(final BsonBinaryWriter writer, final List extraElements) {
- super(writer);
- this.writer = writer;
- this.extraElements = extraElements;
- }
-
- @Override
- public void writeEndDocument() {
- if (getCurrentLevel() == 0) {
- writeElements(writer, extraElements);
- }
- super.writeEndDocument();
- }
-
- @Override
- public void pipe(final BsonReader reader) {
- if (getCurrentLevel() == -1) {
- writer.pipe(reader, extraElements);
- } else {
- writer.pipe(reader);
- }
- }
-}
diff --git a/driver-core/src/main/com/mongodb/internal/connection/IdHoldingBsonWriter.java b/driver-core/src/main/com/mongodb/internal/connection/IdHoldingBsonWriter.java
index 4120dbdfb17..c73a5d4fd86 100644
--- a/driver-core/src/main/com/mongodb/internal/connection/IdHoldingBsonWriter.java
+++ b/driver-core/src/main/com/mongodb/internal/connection/IdHoldingBsonWriter.java
@@ -92,11 +92,11 @@ public void writeStartDocument() {
@Override
public void writeEndDocument() {
if (isWritingId()) {
- if (getIdBsonWriterCurrentLevel() >= 0) {
+ if (getIdBsonWriterCurrentLevel() > DEFAULT_INITIAL_LEVEL) {
getIdBsonWriter().writeEndDocument();
}
- if (getIdBsonWriterCurrentLevel() == -1) {
+ if (getIdBsonWriterCurrentLevel() == DEFAULT_INITIAL_LEVEL) {
if (id != null && id.isJavaScriptWithScope()) {
id = new BsonJavaScriptWithScope(id.asJavaScriptWithScope().getCode(), new RawBsonDocument(getBytes()));
} else if (id == null) {
@@ -105,7 +105,7 @@ public void writeEndDocument() {
}
}
- if (getCurrentLevel() == 0 && id == null) {
+ if (getCurrentLevel() == DEFAULT_INITIAL_LEVEL + 1 && id == null) {
id = fallbackId == null ? new BsonObjectId() : fallbackId;
writeObjectId(ID_FIELD_NAME, id.asObjectId().getValue());
}
@@ -115,7 +115,7 @@ public void writeEndDocument() {
@Override
public void writeStartArray() {
if (isWritingId()) {
- if (getIdBsonWriterCurrentLevel() == -1) {
+ if (getIdBsonWriterCurrentLevel() == DEFAULT_INITIAL_LEVEL) {
idFieldIsAnArray = true;
getIdBsonWriter().writeStartDocument();
getIdBsonWriter().writeName(ID_FIELD_NAME);
@@ -129,7 +129,7 @@ public void writeStartArray() {
public void writeStartArray(final String name) {
setCurrentFieldName(name);
if (isWritingId()) {
- if (getIdBsonWriterCurrentLevel() == -1) {
+ if (getIdBsonWriterCurrentLevel() == DEFAULT_INITIAL_LEVEL) {
getIdBsonWriter().writeStartDocument();
}
getIdBsonWriter().writeStartArray(name);
@@ -141,7 +141,7 @@ public void writeStartArray(final String name) {
public void writeEndArray() {
if (isWritingId()) {
getIdBsonWriter().writeEndArray();
- if (getIdBsonWriterCurrentLevel() == 0 && idFieldIsAnArray) {
+ if (getIdBsonWriterCurrentLevel() == DEFAULT_INITIAL_LEVEL + 1 && idFieldIsAnArray) {
getIdBsonWriter().writeEndDocument();
id = new RawBsonDocument(getBytes()).get(ID_FIELD_NAME);
}
@@ -308,7 +308,7 @@ public void writeMinKey() {
@Override
public void writeName(final String name) {
setCurrentFieldName(name);
- if (getIdBsonWriterCurrentLevel() >= 0) {
+ if (getIdBsonWriterCurrentLevel() > DEFAULT_INITIAL_LEVEL) {
getIdBsonWriter().writeName(name);
}
super.writeName(name);
@@ -433,13 +433,13 @@ private void setCurrentFieldName(final String name) {
}
private boolean isWritingId() {
- return getIdBsonWriterCurrentLevel() >= 0 || (getCurrentLevel() == 0 && currentFieldName != null
+ return getIdBsonWriterCurrentLevel() > DEFAULT_INITIAL_LEVEL || (getCurrentLevel() == DEFAULT_INITIAL_LEVEL + 1 && currentFieldName != null
&& currentFieldName.equals(ID_FIELD_NAME));
}
private void addBsonValue(final Supplier value, final Runnable writeValue) {
if (isWritingId()) {
- if (getIdBsonWriterCurrentLevel() >= 0) {
+ if (getIdBsonWriterCurrentLevel() > DEFAULT_INITIAL_LEVEL) {
writeValue.run();
} else {
id = value.get();
@@ -448,7 +448,7 @@ private void addBsonValue(final Supplier value, final Runnable writeV
}
private int getIdBsonWriterCurrentLevel() {
- return idBsonBinaryWriter == null ? -1 : idBsonBinaryWriter.getCurrentLevel();
+ return idBsonBinaryWriter == null ? DEFAULT_INITIAL_LEVEL : idBsonBinaryWriter.getCurrentLevel();
}
private LevelCountingBsonWriter getIdBsonWriter() {
diff --git a/driver-core/src/main/com/mongodb/internal/connection/LevelCountingBsonWriter.java b/driver-core/src/main/com/mongodb/internal/connection/LevelCountingBsonWriter.java
index 44889765fbf..3e9d0324bd7 100644
--- a/driver-core/src/main/com/mongodb/internal/connection/LevelCountingBsonWriter.java
+++ b/driver-core/src/main/com/mongodb/internal/connection/LevelCountingBsonWriter.java
@@ -20,10 +20,21 @@
abstract class LevelCountingBsonWriter extends BsonWriterDecorator {
- private int level = -1;
+ static final int DEFAULT_INITIAL_LEVEL = -1;
+
+ private int level;
LevelCountingBsonWriter(final BsonWriter bsonWriter) {
+ this(bsonWriter, DEFAULT_INITIAL_LEVEL);
+ }
+
+ /**
+ * @param initialLevel This parameter allows initializing the {@linkplain #getCurrentLevel() current level}
+ * with a value different from {@link #DEFAULT_INITIAL_LEVEL}.
+ */
+ LevelCountingBsonWriter(final BsonWriter bsonWriter, final int initialLevel) {
super(bsonWriter);
+ level = initialLevel;
}
int getCurrentLevel() {
diff --git a/driver-core/src/main/com/mongodb/internal/connection/MessageSequences.java b/driver-core/src/main/com/mongodb/internal/connection/MessageSequences.java
new file mode 100644
index 00000000000..19600007404
--- /dev/null
+++ b/driver-core/src/main/com/mongodb/internal/connection/MessageSequences.java
@@ -0,0 +1,31 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.mongodb.internal.connection;
+
+/**
+ * Zero or more identifiable sequences contained in the {@code OP_MSG} section with payload type 1.
+ *
+ * This class is not part of the public API and may be removed or changed at any time.
+ * @see OP_MSG
+ */
+public abstract class MessageSequences {
+ public static final class EmptyMessageSequences extends MessageSequences {
+ public static final EmptyMessageSequences INSTANCE = new EmptyMessageSequences();
+
+ private EmptyMessageSequences() {
+ }
+ }
+}
diff --git a/driver-core/src/main/com/mongodb/internal/connection/MessageSettings.java b/driver-core/src/main/com/mongodb/internal/connection/MessageSettings.java
index 7a5734bc140..91fd862ce4b 100644
--- a/driver-core/src/main/com/mongodb/internal/connection/MessageSettings.java
+++ b/driver-core/src/main/com/mongodb/internal/connection/MessageSettings.java
@@ -42,6 +42,13 @@ public final class MessageSettings {
* {@code maxWriteBatchSize}.
*/
private static final int DEFAULT_MAX_BATCH_COUNT = 1000;
+ /**
+ * The headroom for documents that are not intended to be stored in a database.
+ * A command document is an example of such a document.
+ * This headroom allows a command document to specify a document that is intended to be stored in a database,
+ * even if the specified document is of the maximum size.
+ */
+ static final int DOCUMENT_HEADROOM_SIZE = 16 * (1 << 10);
private final int maxDocumentSize;
private final int maxMessageSize;
diff --git a/driver-core/src/main/com/mongodb/internal/connection/RequestMessage.java b/driver-core/src/main/com/mongodb/internal/connection/RequestMessage.java
index 86e2ebd1dbe..dd09a59f763 100644
--- a/driver-core/src/main/com/mongodb/internal/connection/RequestMessage.java
+++ b/driver-core/src/main/com/mongodb/internal/connection/RequestMessage.java
@@ -18,24 +18,16 @@
import com.mongodb.lang.Nullable;
import org.bson.BsonBinaryWriter;
-import org.bson.BsonBinaryWriterSettings;
import org.bson.BsonDocument;
-import org.bson.BsonElement;
-import org.bson.BsonWriter;
-import org.bson.BsonWriterSettings;
import org.bson.FieldNameValidator;
-import org.bson.codecs.BsonValueCodecProvider;
-import org.bson.codecs.Codec;
-import org.bson.codecs.Encoder;
-import org.bson.codecs.EncoderContext;
-import org.bson.codecs.configuration.CodecRegistry;
import org.bson.io.BsonOutput;
-import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import static com.mongodb.assertions.Assertions.notNull;
-import static org.bson.codecs.configuration.CodecRegistries.fromProviders;
+import static com.mongodb.internal.connection.BsonWriterHelper.backpatchLength;
+import static com.mongodb.internal.connection.BsonWriterHelper.createBsonBinaryWriter;
+import static com.mongodb.internal.connection.BsonWriterHelper.encodeUsingRegistry;
/**
* Abstract base class for all MongoDB Wire Protocol request messages.
@@ -46,12 +38,6 @@ abstract class RequestMessage {
static final int MESSAGE_PROLOGUE_LENGTH = 16;
- // Allow an extra 16K to the maximum allowed size of a query or command document, so that, for example,
- // a 16M document can be upserted via findAndModify
- private static final int DOCUMENT_HEADROOM = 16 * 1024;
-
- private static final CodecRegistry REGISTRY = fromProviders(new BsonValueCodecProvider());
-
private final String collectionName;
private final MessageSettings settings;
private final int id;
@@ -128,12 +114,12 @@ public MessageSettings getSettings() {
* @param bsonOutput the output
* @param operationContext the session context
*/
- public void encode(final BsonOutput bsonOutput, final OperationContext operationContext) {
+ public void encode(final ByteBufferBsonOutput bsonOutput, final OperationContext operationContext) {
notNull("operationContext", operationContext);
int messageStartPosition = bsonOutput.getPosition();
writeMessagePrologue(bsonOutput);
EncodingMetadata encodingMetadata = encodeMessageBodyWithMetadata(bsonOutput, operationContext);
- backpatchMessageLength(messageStartPosition, bsonOutput);
+ backpatchLength(messageStartPosition, bsonOutput);
this.encodingMetadata = encodingMetadata;
}
@@ -165,23 +151,13 @@ protected void writeMessagePrologue(final BsonOutput bsonOutput) {
* @param operationContext the session context
* @return the encoding metadata
*/
- protected abstract EncodingMetadata encodeMessageBodyWithMetadata(BsonOutput bsonOutput, OperationContext operationContext);
-
- protected void addDocument(final BsonDocument document, final BsonOutput bsonOutput,
- final FieldNameValidator validator, @Nullable final List extraElements) {
- addDocument(document, getCodec(document), EncoderContext.builder().build(), bsonOutput, validator,
- settings.getMaxDocumentSize() + DOCUMENT_HEADROOM, extraElements);
- }
+ protected abstract EncodingMetadata encodeMessageBodyWithMetadata(ByteBufferBsonOutput bsonOutput, OperationContext operationContext);
- /**
- * Backpatches the message length into the beginning of the message.
- *
- * @param startPosition the start position of the message
- * @param bsonOutput the output
- */
- protected void backpatchMessageLength(final int startPosition, final BsonOutput bsonOutput) {
- int messageLength = bsonOutput.getPosition() - startPosition;
- bsonOutput.writeInt32(bsonOutput.getPosition() - messageLength, messageLength);
+ protected int writeDocument(final BsonDocument document, final BsonOutput bsonOutput, final FieldNameValidator validator) {
+ BsonBinaryWriter writer = createBsonBinaryWriter(bsonOutput, validator, getSettings());
+ int documentStart = bsonOutput.getPosition();
+ encodeUsingRegistry(writer, document);
+ return bsonOutput.getPosition() - documentStart;
}
/**
@@ -192,20 +168,4 @@ protected void backpatchMessageLength(final int startPosition, final BsonOutput
protected String getCollectionName() {
return collectionName;
}
-
- @SuppressWarnings("unchecked")
- Codec getCodec(final BsonDocument document) {
- return (Codec) REGISTRY.get(document.getClass());
- }
-
- private void addDocument(final T obj, final Encoder encoder, final EncoderContext encoderContext,
- final BsonOutput bsonOutput, final FieldNameValidator validator, final int maxDocumentSize,
- @Nullable final List extraElements) {
- BsonBinaryWriter bsonBinaryWriter = new BsonBinaryWriter(new BsonWriterSettings(), new BsonBinaryWriterSettings(maxDocumentSize),
- bsonOutput, validator);
- BsonWriter bsonWriter = extraElements == null
- ? bsonBinaryWriter
- : new ElementExtendingBsonWriter(bsonBinaryWriter, extraElements);
- encoder.encode(bsonWriter, obj, encoderContext);
- }
}
diff --git a/driver-core/src/main/com/mongodb/internal/connection/SplittablePayload.java b/driver-core/src/main/com/mongodb/internal/connection/SplittablePayload.java
index d628a39238d..b091d654aa7 100644
--- a/driver-core/src/main/com/mongodb/internal/connection/SplittablePayload.java
+++ b/driver-core/src/main/com/mongodb/internal/connection/SplittablePayload.java
@@ -26,6 +26,7 @@
import org.bson.BsonObjectId;
import org.bson.BsonValue;
import org.bson.BsonWriter;
+import org.bson.FieldNameValidator;
import org.bson.codecs.BsonValueCodecProvider;
import org.bson.codecs.Codec;
import org.bson.codecs.Encoder;
@@ -54,8 +55,9 @@
*
*
This class is not part of the public API and may be removed or changed at any time
*/
-public final class SplittablePayload {
+public final class SplittablePayload extends MessageSequences {
private static final CodecRegistry REGISTRY = fromProviders(new BsonValueCodecProvider());
+ private final FieldNameValidator fieldNameValidator;
private final WriteRequestEncoder writeRequestEncoder = new WriteRequestEncoder();
private final Type payloadType;
private final List writeRequestWithIndexes;
@@ -94,10 +96,19 @@ public enum Type {
* @param payloadType the payload type
* @param writeRequestWithIndexes the writeRequests
*/
- public SplittablePayload(final Type payloadType, final List writeRequestWithIndexes, final boolean ordered) {
+ public SplittablePayload(
+ final Type payloadType,
+ final List writeRequestWithIndexes,
+ final boolean ordered,
+ final FieldNameValidator fieldNameValidator) {
this.payloadType = notNull("batchType", payloadType);
this.writeRequestWithIndexes = notNull("writeRequests", writeRequestWithIndexes);
this.ordered = ordered;
+ this.fieldNameValidator = notNull("fieldNameValidator", fieldNameValidator);
+ }
+
+ public FieldNameValidator getFieldNameValidator() {
+ return fieldNameValidator;
}
/**
@@ -175,7 +186,7 @@ boolean isOrdered() {
public SplittablePayload getNextSplit() {
isTrue("hasAnotherSplit", hasAnotherSplit());
List nextPayLoad = writeRequestWithIndexes.subList(position, writeRequestWithIndexes.size());
- return new SplittablePayload(payloadType, nextPayLoad, ordered);
+ return new SplittablePayload(payloadType, nextPayLoad, ordered, fieldNameValidator);
}
/**
@@ -204,7 +215,8 @@ public void encode(final BsonWriter writer, final WriteRequestWithIndex writeReq
writer,
// Reuse `writeRequestDocumentId` if it may have been generated
// by `IdHoldingBsonWriter` in a previous attempt.
- // If its type is not `BsonObjectId`, we know it could not have been generated.
+ // If its type is not `BsonObjectId`, which happens only if `_id` was specified by the application,
+ // we know it could not have been generated.
writeRequestDocumentId instanceof BsonObjectId ? writeRequestDocumentId.asObjectId() : null);
getCodec(document).encode(idHoldingBsonWriter, document,
EncoderContext.builder().isEncodingCollectibleDocument(true).build());
diff --git a/driver-core/src/main/com/mongodb/internal/connection/SplittablePayloadBsonWriter.java b/driver-core/src/main/com/mongodb/internal/connection/SplittablePayloadBsonWriter.java
index ecff2c95a0c..e679a3b557c 100644
--- a/driver-core/src/main/com/mongodb/internal/connection/SplittablePayloadBsonWriter.java
+++ b/driver-core/src/main/com/mongodb/internal/connection/SplittablePayloadBsonWriter.java
@@ -63,7 +63,7 @@ public void writeStartDocument() {
@Override
public void writeEndDocument() {
- if (getCurrentLevel() == 0 && payload.hasPayload()) {
+ if (getCurrentLevel() == DEFAULT_INITIAL_LEVEL + 1 && payload.hasPayload()) {
writePayloadArray(writer, bsonOutput, settings, messageStartPosition, payload, maxSplittableDocumentSize);
}
super.writeEndDocument();
diff --git a/driver-core/src/main/com/mongodb/internal/operation/BulkWriteBatch.java b/driver-core/src/main/com/mongodb/internal/operation/BulkWriteBatch.java
index 8da0f13e312..1064bee14d3 100644
--- a/driver-core/src/main/com/mongodb/internal/operation/BulkWriteBatch.java
+++ b/driver-core/src/main/com/mongodb/internal/operation/BulkWriteBatch.java
@@ -154,7 +154,7 @@ private BulkWriteBatch(final MongoNamespace namespace, final ConnectionDescripti
this.indexMap = indexMap;
this.unprocessed = unprocessedItems;
- this.payload = new SplittablePayload(getPayloadType(batchType), payloadItems, ordered);
+ this.payload = new SplittablePayload(getPayloadType(batchType), payloadItems, ordered, getFieldNameValidator());
this.operationContext = operationContext;
this.comment = comment;
this.variables = variables;
@@ -270,7 +270,7 @@ BulkWriteBatch getNextBatch() {
}
}
- FieldNameValidator getFieldNameValidator() {
+ private FieldNameValidator getFieldNameValidator() {
if (batchType == UPDATE || batchType == REPLACE) {
Map rootMap;
if (batchType == REPLACE) {
diff --git a/driver-core/src/main/com/mongodb/internal/operation/ClientBulkWriteOperation.java b/driver-core/src/main/com/mongodb/internal/operation/ClientBulkWriteOperation.java
index b2f6b131723..eb68a5e1983 100644
--- a/driver-core/src/main/com/mongodb/internal/operation/ClientBulkWriteOperation.java
+++ b/driver-core/src/main/com/mongodb/internal/operation/ClientBulkWriteOperation.java
@@ -16,7 +16,6 @@
package com.mongodb.internal.operation;
import com.mongodb.ClientBulkWriteException;
-import com.mongodb.Function;
import com.mongodb.MongoClientSettings;
import com.mongodb.MongoCommandException;
import com.mongodb.MongoException;
@@ -31,27 +30,32 @@
import com.mongodb.bulk.WriteConcernError;
import com.mongodb.client.cursor.TimeoutMode;
import com.mongodb.client.model.bulk.ClientBulkWriteOptions;
-import com.mongodb.client.model.bulk.ClientNamespacedReplaceOneModel;
-import com.mongodb.client.model.bulk.ClientNamespacedUpdateOneModel;
-import com.mongodb.client.model.bulk.ClientNamespacedWriteModel;
import com.mongodb.client.model.bulk.ClientBulkWriteResult;
import com.mongodb.client.model.bulk.ClientDeleteResult;
import com.mongodb.client.model.bulk.ClientInsertOneResult;
+import com.mongodb.client.model.bulk.ClientNamespacedReplaceOneModel;
+import com.mongodb.client.model.bulk.ClientNamespacedUpdateOneModel;
+import com.mongodb.client.model.bulk.ClientNamespacedWriteModel;
import com.mongodb.client.model.bulk.ClientUpdateResult;
import com.mongodb.connection.ConnectionDescription;
import com.mongodb.internal.TimeoutContext;
+import com.mongodb.internal.VisibleForTesting;
import com.mongodb.internal.async.function.RetryState;
import com.mongodb.internal.binding.ConnectionSource;
import com.mongodb.internal.binding.WriteBinding;
import com.mongodb.internal.client.model.bulk.AbstractClientDeleteModel;
import com.mongodb.internal.client.model.bulk.AbstractClientNamespacedWriteModel;
import com.mongodb.internal.client.model.bulk.AbstractClientUpdateModel;
+import com.mongodb.internal.client.model.bulk.AcknowledgedSummaryClientBulkWriteResult;
+import com.mongodb.internal.client.model.bulk.AcknowledgedVerboseClientBulkWriteResult;
import com.mongodb.internal.client.model.bulk.ClientWriteModel;
import com.mongodb.internal.client.model.bulk.ConcreteClientBulkWriteOptions;
import com.mongodb.internal.client.model.bulk.ConcreteClientDeleteManyModel;
import com.mongodb.internal.client.model.bulk.ConcreteClientDeleteOneModel;
import com.mongodb.internal.client.model.bulk.ConcreteClientDeleteOptions;
+import com.mongodb.internal.client.model.bulk.ConcreteClientDeleteResult;
import com.mongodb.internal.client.model.bulk.ConcreteClientInsertOneModel;
+import com.mongodb.internal.client.model.bulk.ConcreteClientInsertOneResult;
import com.mongodb.internal.client.model.bulk.ConcreteClientNamespacedDeleteManyModel;
import com.mongodb.internal.client.model.bulk.ConcreteClientNamespacedDeleteOneModel;
import com.mongodb.internal.client.model.bulk.ConcreteClientNamespacedInsertOneModel;
@@ -63,26 +67,26 @@
import com.mongodb.internal.client.model.bulk.ConcreteClientUpdateManyModel;
import com.mongodb.internal.client.model.bulk.ConcreteClientUpdateOneModel;
import com.mongodb.internal.client.model.bulk.ConcreteClientUpdateOptions;
-import com.mongodb.internal.client.model.bulk.AcknowledgedSummaryClientBulkWriteResult;
-import com.mongodb.internal.client.model.bulk.AcknowledgedVerboseClientBulkWriteResult;
-import com.mongodb.internal.client.model.bulk.ConcreteClientDeleteResult;
-import com.mongodb.internal.client.model.bulk.ConcreteClientInsertOneResult;
import com.mongodb.internal.client.model.bulk.ConcreteClientUpdateResult;
import com.mongodb.internal.client.model.bulk.UnacknowledgedClientBulkWriteResult;
import com.mongodb.internal.connection.Connection;
+import com.mongodb.internal.connection.DualMessageSequences;
import com.mongodb.internal.connection.IdHoldingBsonWriter;
import com.mongodb.internal.connection.MongoWriteConcernWithResponseException;
import com.mongodb.internal.connection.OperationContext;
import com.mongodb.internal.operation.retry.AttachmentKeys;
import com.mongodb.internal.session.SessionContext;
-import com.mongodb.internal.validator.MappedFieldNameValidator;
import com.mongodb.internal.validator.NoOpFieldNameValidator;
import com.mongodb.internal.validator.ReplacingDocumentFieldNameValidator;
import com.mongodb.internal.validator.UpdateFieldNameValidator;
import com.mongodb.lang.Nullable;
import org.bson.BsonArray;
+import org.bson.BsonBinaryWriter;
+import org.bson.BsonBoolean;
import org.bson.BsonDocument;
-import org.bson.BsonDocumentWrapper;
+import org.bson.BsonElement;
+import org.bson.BsonInt32;
+import org.bson.BsonInt64;
import org.bson.BsonObjectId;
import org.bson.BsonValue;
import org.bson.BsonWriter;
@@ -106,11 +110,15 @@
import static com.mongodb.assertions.Assertions.assertNotNull;
import static com.mongodb.assertions.Assertions.assertTrue;
import static com.mongodb.assertions.Assertions.fail;
+import static com.mongodb.internal.VisibleForTesting.AccessModifier.PACKAGE;
+import static com.mongodb.internal.VisibleForTesting.AccessModifier.PRIVATE;
+import static com.mongodb.internal.connection.DualMessageSequences.WritersProviderAndLimitsChecker.WriteResult.FAIL_LIMIT_EXCEEDED;
+import static com.mongodb.internal.connection.DualMessageSequences.WritersProviderAndLimitsChecker.WriteResult.OK_LIMIT_NOT_REACHED;
import static com.mongodb.internal.operation.BulkWriteBatch.logWriteModelDoesNotSupportRetries;
+import static com.mongodb.internal.operation.CommandOperationHelper.commandWriteConcern;
import static com.mongodb.internal.operation.CommandOperationHelper.initialRetryState;
import static com.mongodb.internal.operation.CommandOperationHelper.shouldAttemptToRetryWriteAndAddRetryableLabel;
import static com.mongodb.internal.operation.CommandOperationHelper.transformWriteException;
-import static com.mongodb.internal.operation.CommandOperationHelper.commandWriteConcern;
import static com.mongodb.internal.operation.CommandOperationHelper.validateAndGetEffectiveWriteConcern;
import static com.mongodb.internal.operation.OperationHelper.isRetryableWrite;
import static com.mongodb.internal.operation.SyncOperationHelper.cursorDocumentToBatchCursor;
@@ -118,7 +126,8 @@
import static com.mongodb.internal.operation.SyncOperationHelper.withSourceAndConnection;
import static java.util.Collections.emptyList;
import static java.util.Collections.emptyMap;
-import static java.util.Collections.singletonMap;
+import static java.util.Collections.singletonList;
+import static java.util.Optional.ofNullable;
import static java.util.Spliterator.IMMUTABLE;
import static java.util.Spliterator.ORDERED;
import static java.util.Spliterators.spliteratorUnknownSize;
@@ -193,8 +202,8 @@ private void executeAllBatches(
/**
* @return The start model index of the next batch, provided that the operation
- * {@linkplain ExhaustiveBulkWriteCommandOkResponse#operationMayContinue(ConcreteClientBulkWriteOptions) may continue}
- * and there are unexecuted models left.
+ * {@linkplain ExhaustiveClientBulkWriteCommandOkResponse#operationMayContinue(ConcreteClientBulkWriteOptions) may continue}
+ * and there are unexecuted {@linkplain ClientNamespacedWriteModel models} left.
*/
@Nullable
private Integer executeBatch(
@@ -203,12 +212,13 @@ private Integer executeBatch(
final WriteBinding binding,
final ResultAccumulator resultAccumulator) {
List unexecutedModels = models.subList(batchStartModelIndex, models.size());
+ assertFalse(unexecutedModels.isEmpty());
OperationContext operationContext = binding.getOperationContext();
SessionContext sessionContext = operationContext.getSessionContext();
TimeoutContext timeoutContext = operationContext.getTimeoutContext();
RetryState retryState = initialRetryState(retryWritesSetting, timeoutContext);
BatchEncoder batchEncoder = new BatchEncoder();
- Supplier retryingBatchExecutor = decorateWriteWithRetries(
+ Supplier retryingBatchExecutor = decorateWriteWithRetries(
retryState, operationContext,
// Each batch re-selects a server and re-checks out a connection because this is simpler,
// and it is allowed by https://jira.mongodb.org/browse/DRIVERS-2502.
@@ -216,21 +226,21 @@ private Integer executeBatch(
// and `ClientSession`, `TransactionContext` are aware of that.
() -> withSourceAndConnection(binding::getWriteConnectionSource, true, (connectionSource, connection) -> {
ConnectionDescription connectionDescription = connection.getDescription();
- boolean effectiveRetryWrites = isRetryableWrite(retryWritesSetting, effectiveWriteConcern, connectionDescription, sessionContext);
+ boolean effectiveRetryWrites = isRetryableWrite(
+ retryWritesSetting, effectiveWriteConcern, connectionDescription, sessionContext);
retryState.breakAndThrowIfRetryAnd(() -> !effectiveRetryWrites);
resultAccumulator.onNewServerAddress(connectionDescription.getServerAddress());
retryState.attach(AttachmentKeys.maxWireVersion(), connectionDescription.getMaxWireVersion(), true)
.attach(AttachmentKeys.commandDescriptionSupplier(), () -> BULK_WRITE_COMMAND_NAME, false);
- BsonDocumentWrapper lazilyEncodedBulkWriteCommand = createBulkWriteCommand(
- effectiveRetryWrites, effectiveWriteConcern, sessionContext, unexecutedModels, batchEncoder,
+ ClientBulkWriteCommand bulkWriteCommand = createBulkWriteCommand(
+ retryState, effectiveRetryWrites, effectiveWriteConcern, sessionContext, unexecutedModels, batchEncoder,
() -> retryState.attach(AttachmentKeys.retryableCommandFlag(), true, true));
return executeBulkWriteCommandAndExhaustOkResponse(
- retryState, connectionSource, connection, lazilyEncodedBulkWriteCommand, unexecutedModels,
- effectiveWriteConcern, operationContext);
+ retryState, connectionSource, connection, bulkWriteCommand, effectiveWriteConcern, operationContext);
})
);
try {
- ExhaustiveBulkWriteCommandOkResponse bulkWriteCommandOkResponse = retryingBatchExecutor.get();
+ ExhaustiveClientBulkWriteCommandOkResponse bulkWriteCommandOkResponse = retryingBatchExecutor.get();
return resultAccumulator.onBulkWriteCommandOkResponseOrNoResponse(
batchStartModelIndex, bulkWriteCommandOkResponse, batchEncoder.intoEncodedBatchInfo());
} catch (MongoWriteConcernWithResponseException mongoWriteConcernWithOkResponseException) {
@@ -251,36 +261,34 @@ private Integer executeBatch(
/**
* @throws MongoWriteConcernWithResponseException This internal exception must be handled to avoid it being observed by an application.
- * It {@linkplain MongoWriteConcernWithResponseException#getResponse() bears} the OK response to the {@code lazilyEncodedCommand},
+ * It {@linkplain MongoWriteConcernWithResponseException#getResponse() bears} the OK response to the {@code bulkWriteCommand},
* which must be
* {@linkplain ResultAccumulator#onBulkWriteCommandOkResponseWithWriteConcernError(int, MongoWriteConcernWithResponseException, BatchEncoder.EncodedBatchInfo) accumulated}
* iff this exception is the failed result of retries.
*/
@Nullable
- private ExhaustiveBulkWriteCommandOkResponse executeBulkWriteCommandAndExhaustOkResponse(
+ private ExhaustiveClientBulkWriteCommandOkResponse executeBulkWriteCommandAndExhaustOkResponse(
final RetryState retryState,
final ConnectionSource connectionSource,
final Connection connection,
- final BsonDocumentWrapper lazilyEncodedCommand,
- final List unexecutedModels,
+ final ClientBulkWriteCommand bulkWriteCommand,
final WriteConcern effectiveWriteConcern,
final OperationContext operationContext) throws MongoWriteConcernWithResponseException {
BsonDocument bulkWriteCommandOkResponse = connection.command(
"admin",
- lazilyEncodedCommand,
- FieldNameValidators.createUpdateModsFieldValidator(unexecutedModels),
+ bulkWriteCommand.getCommandDocument(),
+ NoOpFieldNameValidator.INSTANCE,
null,
CommandResultDocumentCodec.create(codecRegistry.get(BsonDocument.class), CommandBatchCursorHelper.FIRST_BATCH),
operationContext,
effectiveWriteConcern.isAcknowledged(),
- null,
- null);
+ bulkWriteCommand.getOpsAndNsInfo());
if (bulkWriteCommandOkResponse == null) {
return null;
}
List> cursorExhaustBatches = doWithRetriesDisabledForCommand(retryState, "getMore", () ->
exhaustBulkWriteCommandOkResponseCursor(connectionSource, connection, bulkWriteCommandOkResponse));
- ExhaustiveBulkWriteCommandOkResponse exhaustiveBulkWriteCommandOkResponse = new ExhaustiveBulkWriteCommandOkResponse(
+ ExhaustiveClientBulkWriteCommandOkResponse exhaustiveBulkWriteCommandOkResponse = new ExhaustiveClientBulkWriteCommandOkResponse(
bulkWriteCommandOkResponse, cursorExhaustBatches);
// `Connection.command` does not throw `MongoWriteConcernException`, so we have to construct it ourselves
MongoWriteConcernException writeConcernException = Exceptions.createWriteConcernException(
@@ -313,7 +321,7 @@ private List> exhaustBulkWriteCommandOkResponseCursor(
final Connection connection,
final BsonDocument response) {
int serverDefaultCursorBatchSize = 0;
- try (BatchCursor cursor = cursorDocumentToBatchCursor(
+ try (CommandBatchCursor cursor = cursorDocumentToBatchCursor(
TimeoutMode.CURSOR_LIFETIME,
response,
serverDefaultCursorBatchSize,
@@ -321,78 +329,42 @@ private List> exhaustBulkWriteCommandOkResponseCursor(
options.getComment().orElse(null),
connectionSource,
connection)) {
+ cursor.setCloseWithoutTimeoutReset(true);
return stream(spliteratorUnknownSize(cursor, ORDERED | IMMUTABLE), false).collect(toList());
}
}
- private BsonDocumentWrapper createBulkWriteCommand(
+ private ClientBulkWriteCommand createBulkWriteCommand(
+ final RetryState retryState,
final boolean effectiveRetryWrites,
final WriteConcern effectiveWriteConcern,
final SessionContext sessionContext,
final List unexecutedModels,
final BatchEncoder batchEncoder,
- final Runnable ifCommandIsRetryable) {
- return new BsonDocumentWrapper<>(
- BULK_WRITE_COMMAND_NAME,
- new Encoder() {
- @Override
- public void encode(final BsonWriter writer, final String commandName, final EncoderContext encoderContext) {
- batchEncoder.reset();
- writer.writeStartDocument();
- writer.writeInt32(commandName, 1);
- writer.writeBoolean("errorsOnly", !options.isVerboseResults());
- writer.writeBoolean("ordered", options.isOrdered());
- options.isBypassDocumentValidation().ifPresent(value -> writer.writeBoolean("bypassDocumentValidation", value));
- options.getComment().ifPresent(value -> {
- writer.writeName("comment");
- encodeUsingRegistry(writer, value);
- });
- options.getLet().ifPresent(value -> {
- writer.writeName("let");
- encodeUsingRegistry(writer, value);
- });
- Function modelSupportsRetries = model ->
- !(model instanceof ConcreteClientUpdateManyModel || model instanceof ConcreteClientDeleteManyModel);
- assertFalse(unexecutedModels.isEmpty());
- LinkedHashMap indexedNamespaces = new LinkedHashMap<>();
- writer.writeStartArray("ops");
- boolean commandIsRetryable = effectiveRetryWrites;
- for (int modelIndexInBatch = 0; modelIndexInBatch < unexecutedModels.size(); modelIndexInBatch++) {
- AbstractClientNamespacedWriteModel modelWithNamespace = getNamespacedModel(unexecutedModels, modelIndexInBatch);
- ClientWriteModel model = modelWithNamespace.getModel();
- if (commandIsRetryable && !modelSupportsRetries.apply(model)) {
- commandIsRetryable = false;
- logWriteModelDoesNotSupportRetries();
- }
- int namespaceIndexInBatch = indexedNamespaces.computeIfAbsent(
- modelWithNamespace.getNamespace(), k -> indexedNamespaces.size());
- batchEncoder.encodeWriteModel(writer, model, modelIndexInBatch, namespaceIndexInBatch);
- }
- writer.writeEndArray();
- writer.writeStartArray("nsInfo");
- indexedNamespaces.keySet().forEach(namespace -> {
- writer.writeStartDocument();
- writer.writeString("ns", namespace.getFullName());
- writer.writeEndDocument();
- });
- writer.writeEndArray();
- if (commandIsRetryable) {
- batchEncoder.encodeTxnNumber(writer, sessionContext);
- ifCommandIsRetryable.run();
- }
- commandWriteConcern(effectiveWriteConcern, sessionContext).ifPresent(value -> {
- writer.writeName("writeConcern");
- encodeUsingRegistry(writer, value.asDocument());
- });
- writer.writeEndDocument();
- }
-
- @Override
- public Class getEncoderClass() {
- throw fail();
- }
- }
- );
+ final Runnable retriesEnabler) {
+ BsonDocument commandDocument = new BsonDocument(BULK_WRITE_COMMAND_NAME, new BsonInt32(1))
+ .append("errorsOnly", BsonBoolean.valueOf(!options.isVerboseResults()))
+ .append("ordered", BsonBoolean.valueOf(options.isOrdered()));
+ options.isBypassDocumentValidation().ifPresent(value ->
+ commandDocument.append("bypassDocumentValidation", BsonBoolean.valueOf(value)));
+ options.getComment().ifPresent(value ->
+ commandDocument.append("comment", value));
+ options.getLet().ifPresent(let ->
+ commandDocument.append("let", let.toBsonDocument(BsonDocument.class, codecRegistry)));
+ commandWriteConcern(effectiveWriteConcern, sessionContext).ifPresent(value->
+ commandDocument.append("writeConcern", value.asDocument()));
+ return new ClientBulkWriteCommand(
+ commandDocument,
+ new ClientBulkWriteCommand.OpsAndNsInfo(
+ effectiveRetryWrites, unexecutedModels,
+ batchEncoder,
+ options,
+ () -> {
+ retriesEnabler.run();
+ return retryState.isFirstAttempt()
+ ? sessionContext.advanceTransactionNumber()
+ : sessionContext.getTransactionNumber();
+ }));
}
private void encodeUsingRegistry(final BsonWriter writer, final T value) {
@@ -401,8 +373,8 @@ private void encodeUsingRegistry(final BsonWriter writer, final T value) {
private void encodeUsingRegistry(final BsonWriter writer, final T value, final EncoderContext encoderContext) {
@SuppressWarnings("unchecked")
- Encoder collationEncoder = (Encoder) codecRegistry.get(value.getClass());
- collationEncoder.encode(writer, value, encoderContext);
+ Encoder encoder = (Encoder) codecRegistry.get(value.getClass());
+ encoder.encode(writer, value, encoderContext);
}
private static AbstractClientNamespacedWriteModel getNamespacedModel(
@@ -418,7 +390,7 @@ public static Optional serverAddressFromException(@Nullable final
} else if (exception instanceof MongoSocketException) {
serverAddress = ((MongoSocketException) exception).getServerAddress();
}
- return Optional.ofNullable(serverAddress);
+ return ofNullable(serverAddress);
}
@Nullable
@@ -438,139 +410,7 @@ private static MongoWriteConcernException createWriteConcernException(
}
}
- private static final class FieldNameValidators {
- /**
- * The server supports only the {@code update} individual write operation in the {@code ops} array field, while the driver supports
- * {@link ClientNamespacedUpdateOneModel}, {@link ClientNamespacedUpdateOneModel}, {@link ClientNamespacedReplaceOneModel}.
- * The difference between updating and replacing is only in the document specified via the {@code updateMods} field:
- *
- *
if the name of the first field starts with {@code '$'}, then the document is interpreted as specifying update operators;
- *
if the name of the first field does not start with {@code '$'}, then the document is interpreted as a replacement.
- *
- */
- private static FieldNameValidator createUpdateModsFieldValidator(final List models) {
- return new MappedFieldNameValidator(
- NoOpFieldNameValidator.INSTANCE,
- singletonMap("ops", new FieldNameValidators.OpsArrayFieldValidator(models)));
- }
-
- static final class OpsArrayFieldValidator implements FieldNameValidator {
- private static final Set OPERATION_DISCRIMINATOR_FIELD_NAMES = Stream.of("insert", "update", "delete").collect(toSet());
-
- private final List models;
- private final ReplacingUpdateModsFieldValidator replacingValidator;
- private final UpdatingUpdateModsFieldValidator updatingValidator;
- private int currentIndividualOperationIndex;
-
- OpsArrayFieldValidator(final List models) {
- this.models = models;
- replacingValidator = new ReplacingUpdateModsFieldValidator();
- updatingValidator = new UpdatingUpdateModsFieldValidator();
- currentIndividualOperationIndex = -1;
- }
-
- @Override
- public boolean validate(final String fieldName) {
- if (OPERATION_DISCRIMINATOR_FIELD_NAMES.contains(fieldName)) {
- currentIndividualOperationIndex++;
- }
- return true;
- }
-
- @Override
- public FieldNameValidator getValidatorForField(final String fieldName) {
- if (fieldName.equals("updateMods")) {
- return currentIndividualOperationIsReplace() ? replacingValidator.reset() : updatingValidator.reset();
- }
- return NoOpFieldNameValidator.INSTANCE;
- }
-
- private boolean currentIndividualOperationIsReplace() {
- return getNamespacedModel(models, currentIndividualOperationIndex) instanceof ConcreteClientNamespacedReplaceOneModel;
- }
- }
-
- private static final class ReplacingUpdateModsFieldValidator implements FieldNameValidator {
- private boolean firstFieldSinceLastReset;
-
- ReplacingUpdateModsFieldValidator() {
- firstFieldSinceLastReset = true;
- }
-
- @Override
- public boolean validate(final String fieldName) {
- if (firstFieldSinceLastReset) {
- // we must validate only the first field, and leave the rest up to the server
- firstFieldSinceLastReset = false;
- return ReplacingDocumentFieldNameValidator.INSTANCE.validate(fieldName);
- }
- return true;
- }
-
- @Override
- public String getValidationErrorMessage(final String fieldName) {
- return ReplacingDocumentFieldNameValidator.INSTANCE.getValidationErrorMessage(fieldName);
- }
-
- @Override
- public FieldNameValidator getValidatorForField(final String fieldName) {
- return NoOpFieldNameValidator.INSTANCE;
- }
-
- ReplacingUpdateModsFieldValidator reset() {
- firstFieldSinceLastReset = true;
- return this;
- }
- }
-
- private static final class UpdatingUpdateModsFieldValidator implements FieldNameValidator {
- private final UpdateFieldNameValidator delegate;
- private boolean firstFieldSinceLastReset;
-
- UpdatingUpdateModsFieldValidator() {
- delegate = new UpdateFieldNameValidator();
- firstFieldSinceLastReset = true;
- }
-
- @Override
- public boolean validate(final String fieldName) {
- if (firstFieldSinceLastReset) {
- // we must validate only the first field, and leave the rest up to the server
- firstFieldSinceLastReset = false;
- return delegate.validate(fieldName);
- }
- return true;
- }
-
- @Override
- public String getValidationErrorMessage(final String fieldName) {
- return delegate.getValidationErrorMessage(fieldName);
- }
-
- @Override
- public FieldNameValidator getValidatorForField(final String fieldName) {
- return NoOpFieldNameValidator.INSTANCE;
- }
-
- @Override
- public void start() {
- delegate.start();
- }
-
- @Override
- public void end() {
- delegate.end();
- }
-
- UpdatingUpdateModsFieldValidator reset() {
- delegate.reset();
- firstFieldSinceLastReset = true;
- return this;
- }
- }
- }
-
- private static final class ExhaustiveBulkWriteCommandOkResponse {
+ private static final class ExhaustiveClientBulkWriteCommandOkResponse {
/**
* The number of unsuccessful individual write operations.
*/
@@ -582,7 +422,7 @@ private static final class ExhaustiveBulkWriteCommandOkResponse {
private final int nDeleted;
private final List cursorExhaust;
- ExhaustiveBulkWriteCommandOkResponse(
+ ExhaustiveClientBulkWriteCommandOkResponse(
final BsonDocument bulkWriteCommandOkResponse,
final List> cursorExhaustBatches) {
this.nErrors = bulkWriteCommandOkResponse.getInt32("nErrors").getValue();
@@ -656,8 +496,8 @@ private final class ResultAccumulator {
*/
ClientBulkWriteResult build(@Nullable final MongoException topLevelError, final WriteConcern effectiveWriteConcern) throws MongoException {
boolean verboseResultsSetting = options.isVerboseResults();
- boolean haveResponses = false;
- boolean haveSuccessfulIndividualOperations = false;
+ boolean batchResultsHaveResponses = false;
+ boolean batchResultsHaveInfoAboutSuccessfulIndividualOperations = false;
long insertedCount = 0;
long upsertedCount = 0;
long matchedCount = 0;
@@ -670,15 +510,18 @@ ClientBulkWriteResult build(@Nullable final MongoException topLevelError, final
Map writeErrors = new HashMap<>();
for (BatchResult batchResult : batchResults) {
if (batchResult.hasResponse()) {
- haveResponses = true;
+ batchResultsHaveResponses = true;
MongoWriteConcernException writeConcernException = batchResult.getWriteConcernException();
if (writeConcernException != null) {
writeConcernErrors.add(writeConcernException.getWriteConcernError());
}
int batchStartModelIndex = batchResult.getBatchStartModelIndex();
- ExhaustiveBulkWriteCommandOkResponse response = batchResult.getResponse();
- haveSuccessfulIndividualOperations = haveSuccessfulIndividualOperations
- || response.getNErrors() < batchResult.getBatchModelsCount();
+ ExhaustiveClientBulkWriteCommandOkResponse response = batchResult.getResponse();
+ boolean orderedSetting = options.isOrdered();
+ int nErrors = response.getNErrors();
+ batchResultsHaveInfoAboutSuccessfulIndividualOperations = batchResultsHaveInfoAboutSuccessfulIndividualOperations
+ || (orderedSetting && nErrors == 0)
+ || (!orderedSetting && nErrors < batchResult.getBatchModelsCount());
insertedCount += response.getNInserted();
upsertedCount += response.getNUpserted();
matchedCount += response.getNMatched();
@@ -714,6 +557,8 @@ ClientBulkWriteResult build(@Nullable final MongoException topLevelError, final
fail(writeModel.getClass().toString());
}
} else {
+ batchResultsHaveInfoAboutSuccessfulIndividualOperations = batchResultsHaveInfoAboutSuccessfulIndividualOperations
+ || (orderedSetting && individualOperationIndexInBatch > 0);
WriteError individualOperationWriteError = new WriteError(
individualOperationResponse.getInt32("code").getValue(),
individualOperationResponse.getString("errmsg").getValue(),
@@ -733,8 +578,8 @@ ClientBulkWriteResult build(@Nullable final MongoException topLevelError, final
} else {
return UnacknowledgedClientBulkWriteResult.INSTANCE;
}
- } else if (haveResponses) {
- AcknowledgedSummaryClientBulkWriteResult partialSummaryResult = haveSuccessfulIndividualOperations
+ } else if (batchResultsHaveResponses) {
+ AcknowledgedSummaryClientBulkWriteResult partialSummaryResult = batchResultsHaveInfoAboutSuccessfulIndividualOperations
? new AcknowledgedSummaryClientBulkWriteResult(insertedCount, upsertedCount, matchedCount, modifiedCount, deletedCount)
: null;
throw new ClientBulkWriteException(
@@ -758,7 +603,7 @@ void onNewServerAddress(final ServerAddress serverAddress) {
Integer onBulkWriteCommandOkResponseOrNoResponse(
final int batchStartModelIndex,
@Nullable
- final ExhaustiveBulkWriteCommandOkResponse response,
+ final ExhaustiveClientBulkWriteCommandOkResponse response,
final BatchEncoder.EncodedBatchInfo encodedBatchInfo) {
return onBulkWriteCommandOkResponseOrNoResponse(batchStartModelIndex, response, null, encodedBatchInfo);
}
@@ -773,7 +618,7 @@ Integer onBulkWriteCommandOkResponseWithWriteConcernError(
final BatchEncoder.EncodedBatchInfo encodedBatchInfo) {
MongoWriteConcernException writeConcernException = (MongoWriteConcernException) exception.getCause();
onNewServerAddress(writeConcernException.getServerAddress());
- ExhaustiveBulkWriteCommandOkResponse response = (ExhaustiveBulkWriteCommandOkResponse) exception.getResponse();
+ ExhaustiveClientBulkWriteCommandOkResponse response = (ExhaustiveClientBulkWriteCommandOkResponse) exception.getResponse();
return onBulkWriteCommandOkResponseOrNoResponse(batchStartModelIndex, response, writeConcernException, encodedBatchInfo);
}
@@ -784,7 +629,7 @@ Integer onBulkWriteCommandOkResponseWithWriteConcernError(
private Integer onBulkWriteCommandOkResponseOrNoResponse(
final int batchStartModelIndex,
@Nullable
- final ExhaustiveBulkWriteCommandOkResponse response,
+ final ExhaustiveClientBulkWriteCommandOkResponse response,
@Nullable
final MongoWriteConcernException writeConcernException,
final BatchEncoder.EncodedBatchInfo encodedBatchInfo) {
@@ -807,18 +652,236 @@ void onBulkWriteCommandErrorWithoutResponse(final MongoException exception) {
}
}
+ public static final class ClientBulkWriteCommand {
+ private final BsonDocument commandDocument;
+ private final OpsAndNsInfo opsAndNsInfo;
+
+ ClientBulkWriteCommand(
+ final BsonDocument commandDocument,
+ final OpsAndNsInfo opsAndNsInfo) {
+ this.commandDocument = commandDocument;
+ this.opsAndNsInfo = opsAndNsInfo;
+ }
+
+ BsonDocument getCommandDocument() {
+ return commandDocument;
+ }
+
+ OpsAndNsInfo getOpsAndNsInfo() {
+ return opsAndNsInfo;
+ }
+
+ public static final class OpsAndNsInfo extends DualMessageSequences {
+ private final boolean effectiveRetryWrites;
+ private final List models;
+ private final BatchEncoder batchEncoder;
+ private final ConcreteClientBulkWriteOptions options;
+ private final Supplier doIfCommandIsRetryableAndAdvanceGetTxnNumber;
+
+ @VisibleForTesting(otherwise = PACKAGE)
+ public OpsAndNsInfo(
+ final boolean effectiveRetryWrites,
+ final List models,
+ final BatchEncoder batchEncoder,
+ final ConcreteClientBulkWriteOptions options,
+ final Supplier doIfCommandIsRetryableAndAdvanceGetTxnNumber) {
+ super("ops", new OpsFieldNameValidator(models), "nsInfo", NoOpFieldNameValidator.INSTANCE);
+ this.effectiveRetryWrites = effectiveRetryWrites;
+ this.models = models;
+ this.batchEncoder = batchEncoder;
+ this.options = options;
+ this.doIfCommandIsRetryableAndAdvanceGetTxnNumber = doIfCommandIsRetryableAndAdvanceGetTxnNumber;
+ }
+
+ @Override
+ public EncodeDocumentsResult encodeDocuments(final WritersProviderAndLimitsChecker writersProviderAndLimitsChecker) {
+ // We must call `batchEncoder.reset` lazily, that is here, and not eagerly before a command retry attempt,
+ // because a retry attempt may fail before encoding,
+ // in which case we need the information gathered by `batchEncoder` at a previous attempt.
+ batchEncoder.reset();
+ LinkedHashMap indexedNamespaces = new LinkedHashMap<>();
+ WritersProviderAndLimitsChecker.WriteResult writeResult = OK_LIMIT_NOT_REACHED;
+ boolean commandIsRetryable = effectiveRetryWrites;
+ int maxModelIndexInBatch = -1;
+ for (int modelIndexInBatch = 0; modelIndexInBatch < models.size() && writeResult == OK_LIMIT_NOT_REACHED; modelIndexInBatch++) {
+ AbstractClientNamespacedWriteModel namespacedModel = getNamespacedModel(models, modelIndexInBatch);
+ MongoNamespace namespace = namespacedModel.getNamespace();
+ int indexedNamespacesSizeBeforeCompute = indexedNamespaces.size();
+ int namespaceIndexInBatch = indexedNamespaces.computeIfAbsent(namespace, k -> indexedNamespacesSizeBeforeCompute);
+ boolean writeNewNamespace = indexedNamespaces.size() != indexedNamespacesSizeBeforeCompute;
+ int finalModelIndexInBatch = modelIndexInBatch;
+ writeResult = writersProviderAndLimitsChecker.tryWrite((opsWriter, nsInfoWriter) -> {
+ batchEncoder.encodeWriteModel(opsWriter, namespacedModel.getModel(), finalModelIndexInBatch, namespaceIndexInBatch);
+ if (writeNewNamespace) {
+ nsInfoWriter.writeStartDocument();
+ nsInfoWriter.writeString("ns", namespace.getFullName());
+ nsInfoWriter.writeEndDocument();
+ }
+ return finalModelIndexInBatch + 1;
+ });
+ if (writeResult == FAIL_LIMIT_EXCEEDED) {
+ batchEncoder.reset(finalModelIndexInBatch);
+ } else {
+ maxModelIndexInBatch = finalModelIndexInBatch;
+ if (commandIsRetryable && doesNotSupportRetries(namespacedModel)) {
+ commandIsRetryable = false;
+ logWriteModelDoesNotSupportRetries();
+ }
+ }
+ }
+ return new EncodeDocumentsResult(
+ // we will execute more batches, so we must request a response to maintain the order of individual write operations
+ options.isOrdered() && maxModelIndexInBatch < models.size() - 1,
+ commandIsRetryable
+ ? singletonList(new BsonElement("txnNumber", new BsonInt64(doIfCommandIsRetryableAndAdvanceGetTxnNumber.get())))
+ : emptyList());
+ }
+
+ private static boolean doesNotSupportRetries(final AbstractClientNamespacedWriteModel model) {
+ return model instanceof ConcreteClientNamespacedUpdateManyModel || model instanceof ConcreteClientNamespacedDeleteManyModel;
+ }
+
+ /**
+ * The server supports only the {@code update} individual write operation in the {@code ops} array field, while the driver supports
+ * {@link ClientNamespacedUpdateOneModel}, {@link ClientNamespacedUpdateOneModel}, {@link ClientNamespacedReplaceOneModel}.
+ * The difference between updating and replacing is only in the document specified via the {@code updateMods} field:
+ *
+ *
if the name of the first field starts with {@code '$'}, then the document is interpreted as specifying update operators;
+ *
if the name of the first field does not start with {@code '$'}, then the document is interpreted as a replacement.
+ *
+ *
+ * @see
+ * Update vs. replace document validation
+ */
+ private static final class OpsFieldNameValidator implements FieldNameValidator {
+ private static final Set OPERATION_DISCRIMINATOR_FIELD_NAMES = Stream.of("insert", "update", "delete").collect(toSet());
+
+ private final List models;
+ private final ReplacingUpdateModsFieldValidator replacingValidator;
+ private final UpdatingUpdateModsFieldValidator updatingValidator;
+ private int currentIndividualOperationIndex;
+
+ OpsFieldNameValidator(final List models) {
+ this.models = models;
+ replacingValidator = new ReplacingUpdateModsFieldValidator();
+ updatingValidator = new UpdatingUpdateModsFieldValidator();
+ currentIndividualOperationIndex = -1;
+ }
+
+ @Override
+ public boolean validate(final String fieldName) {
+ if (OPERATION_DISCRIMINATOR_FIELD_NAMES.contains(fieldName)) {
+ currentIndividualOperationIndex++;
+ }
+ return true;
+ }
+
+ @Override
+ public FieldNameValidator getValidatorForField(final String fieldName) {
+ if (fieldName.equals("updateMods")) {
+ return currentIndividualOperationIsReplace() ? replacingValidator.reset() : updatingValidator.reset();
+ }
+ return NoOpFieldNameValidator.INSTANCE;
+ }
+
+ private boolean currentIndividualOperationIsReplace() {
+ return getNamespacedModel(models, currentIndividualOperationIndex) instanceof ConcreteClientNamespacedReplaceOneModel;
+ }
+
+ private static final class ReplacingUpdateModsFieldValidator implements FieldNameValidator {
+ private boolean firstFieldSinceLastReset;
+
+ ReplacingUpdateModsFieldValidator() {
+ firstFieldSinceLastReset = true;
+ }
+
+ @Override
+ public boolean validate(final String fieldName) {
+ if (firstFieldSinceLastReset) {
+ // we must validate only the first field, and leave the rest up to the server
+ firstFieldSinceLastReset = false;
+ return ReplacingDocumentFieldNameValidator.INSTANCE.validate(fieldName);
+ }
+ return true;
+ }
+
+ @Override
+ public String getValidationErrorMessage(final String fieldName) {
+ return ReplacingDocumentFieldNameValidator.INSTANCE.getValidationErrorMessage(fieldName);
+ }
+
+ @Override
+ public FieldNameValidator getValidatorForField(final String fieldName) {
+ return NoOpFieldNameValidator.INSTANCE;
+ }
+
+ ReplacingUpdateModsFieldValidator reset() {
+ firstFieldSinceLastReset = true;
+ return this;
+ }
+ }
+
+ private static final class UpdatingUpdateModsFieldValidator implements FieldNameValidator {
+ private final UpdateFieldNameValidator delegate;
+ private boolean firstFieldSinceLastReset;
+
+ UpdatingUpdateModsFieldValidator() {
+ delegate = new UpdateFieldNameValidator();
+ firstFieldSinceLastReset = true;
+ }
+
+ @Override
+ public boolean validate(final String fieldName) {
+ if (firstFieldSinceLastReset) {
+ // we must validate only the first field, and leave the rest up to the server
+ firstFieldSinceLastReset = false;
+ return delegate.validate(fieldName);
+ }
+ return true;
+ }
+
+ @Override
+ public String getValidationErrorMessage(final String fieldName) {
+ return delegate.getValidationErrorMessage(fieldName);
+ }
+
+ @Override
+ public FieldNameValidator getValidatorForField(final String fieldName) {
+ return NoOpFieldNameValidator.INSTANCE;
+ }
+
+ @Override
+ public void start() {
+ delegate.start();
+ }
+
+ @Override
+ public void end() {
+ delegate.end();
+ }
+
+ UpdatingUpdateModsFieldValidator reset() {
+ delegate.reset();
+ firstFieldSinceLastReset = true;
+ return this;
+ }
+ }
+ }
+ }
+ }
+
static final class BatchResult {
private final int batchStartModelIndex;
private final BatchEncoder.EncodedBatchInfo encodedBatchInfo;
@Nullable
- private final ExhaustiveBulkWriteCommandOkResponse response;
+ private final ExhaustiveClientBulkWriteCommandOkResponse response;
@Nullable
private final MongoWriteConcernException writeConcernException;
static BatchResult okResponse(
final int batchStartModelIndex,
final BatchEncoder.EncodedBatchInfo encodedBatchInfo,
- final ExhaustiveBulkWriteCommandOkResponse response,
+ final ExhaustiveClientBulkWriteCommandOkResponse response,
@Nullable final MongoWriteConcernException writeConcernException) {
return new BatchResult(batchStartModelIndex, encodedBatchInfo, assertNotNull(response), writeConcernException);
}
@@ -830,7 +893,7 @@ static BatchResult noResponse(final int batchStartModelIndex, final BatchEncoder
private BatchResult(
final int batchStartModelIndex,
final BatchEncoder.EncodedBatchInfo encodedBatchInfo,
- @Nullable final ExhaustiveBulkWriteCommandOkResponse response,
+ @Nullable final ExhaustiveClientBulkWriteCommandOkResponse response,
@Nullable final MongoWriteConcernException writeConcernException) {
this.batchStartModelIndex = batchStartModelIndex;
this.encodedBatchInfo = encodedBatchInfo;
@@ -853,7 +916,7 @@ boolean hasResponse() {
return response != null;
}
- ExhaustiveBulkWriteCommandOkResponse getResponse() {
+ ExhaustiveClientBulkWriteCommandOkResponse getResponse() {
return assertNotNull(response);
}
@@ -875,16 +938,19 @@ Map getInsertModelDocumentIds() {
/**
* Exactly one instance must be used per {@linkplain #executeBatch(int, WriteConcern, WriteBinding, ResultAccumulator) batch}.
*/
- private final class BatchEncoder {
+ @VisibleForTesting(otherwise = PRIVATE)
+ public final class BatchEncoder {
private EncodedBatchInfo encodedBatchInfo;
- BatchEncoder() {
+ @VisibleForTesting(otherwise = PACKAGE)
+ public BatchEncoder() {
encodedBatchInfo = new EncodedBatchInfo();
}
/**
* Must be called at most once.
- * Must not be called before calling {@link #encodeWriteModel(BsonWriter, ClientWriteModel, int, int)} at least once.
+ * Must not be called before calling
+ * {@link #encodeWriteModel(BsonBinaryWriter, ClientWriteModel, int, int)} at least once.
* Renders {@code this} unusable.
*/
EncodedBatchInfo intoEncodedBatchInfo() {
@@ -899,16 +965,13 @@ void reset() {
assertNotNull(encodedBatchInfo).modelsCount = 0;
}
- void encodeTxnNumber(final BsonWriter writer, final SessionContext sessionContext) {
- EncodedBatchInfo localEncodedBatchInfo = assertNotNull(encodedBatchInfo);
- if (localEncodedBatchInfo.txnNumber == EncodedBatchInfo.UNINITIALIZED_TXN_NUMBER) {
- localEncodedBatchInfo.txnNumber = sessionContext.advanceTransactionNumber();
- }
- writer.writeInt64("txnNumber", localEncodedBatchInfo.txnNumber);
+ void reset(final int modelIndexInBatch) {
+ assertNotNull(encodedBatchInfo).modelsCount -= 1;
+ encodedBatchInfo.insertModelDocumentIds.remove(modelIndexInBatch);
}
void encodeWriteModel(
- final BsonWriter writer,
+ final BsonBinaryWriter writer,
final ClientWriteModel model,
final int modelIndexInBatch,
final int namespaceIndexInBatch) {
@@ -942,18 +1005,20 @@ void encodeWriteModel(
writer.writeEndDocument();
}
- private void encodeWriteModelInternals(final BsonWriter writer, final ConcreteClientInsertOneModel model, final int modelIndexInBatch) {
+ private void encodeWriteModelInternals(
+ final BsonBinaryWriter writer,
+ final ConcreteClientInsertOneModel model,
+ final int modelIndexInBatch) {
writer.writeName("document");
Object document = model.getDocument();
- @SuppressWarnings("unchecked")
- Encoder