diff --git a/core/src/main/java/de/bwaldvogel/mongo/wire/MongoDatabaseHandler.java b/core/src/main/java/de/bwaldvogel/mongo/wire/MongoDatabaseHandler.java index 41e59d92..eeedde1c 100644 --- a/core/src/main/java/de/bwaldvogel/mongo/wire/MongoDatabaseHandler.java +++ b/core/src/main/java/de/bwaldvogel/mongo/wire/MongoDatabaseHandler.java @@ -60,6 +60,11 @@ protected void channelRead0(ChannelHandlerContext ctx, ClientRequest object) { ctx.channel().writeAndFlush(mongoReply); } else if (object instanceof MongoMessage mongoMessage) { MongoMessage response = handleMessage(mongoMessage); + if (mongoMessage.isMoreToCome()) { + // the client does not read a reply, so sending one would desynchronize the connection + log.debug("not replying to {} with moreToCome set", mongoMessage); + return; + } ctx.channel().writeAndFlush(response); } else { throw new MongoServerException("unknown message: " + object); diff --git a/core/src/main/java/de/bwaldvogel/mongo/wire/MongoWireProtocolHandler.java b/core/src/main/java/de/bwaldvogel/mongo/wire/MongoWireProtocolHandler.java index a8c3b69f..a6519cbe 100644 --- a/core/src/main/java/de/bwaldvogel/mongo/wire/MongoWireProtocolHandler.java +++ b/core/src/main/java/de/bwaldvogel/mongo/wire/MongoWireProtocolHandler.java @@ -142,6 +142,11 @@ private ClientRequest handleMessage(Channel channel, MessageHeader header, ByteB flags.add(MessageFlag.CHECKSUM_PRESENT); } + if (MessageFlag.MORE_TO_COME.isSet(flagBits)) { + flagBits = MessageFlag.MORE_TO_COME.removeFrom(flagBits); + flags.add(MessageFlag.MORE_TO_COME); + } + if (flagBits != 0) { throw new UnsupportedOperationException("flags=" + flagBits + " not yet supported"); } @@ -179,7 +184,7 @@ private ClientRequest handleMessage(Channel channel, MessageHeader header, ByteB Assert.isNull(old); } - return new MongoMessage(channel, header, body); + return new MongoMessage(channel, header, body, flags.contains(MessageFlag.MORE_TO_COME)); } private void decodeKindDocumentSequence(ByteBuf buffer, Document documentSequence) { diff --git a/core/src/main/java/de/bwaldvogel/mongo/wire/message/MongoMessage.java b/core/src/main/java/de/bwaldvogel/mongo/wire/message/MongoMessage.java index cf7051a5..f8213ea9 100644 --- a/core/src/main/java/de/bwaldvogel/mongo/wire/message/MongoMessage.java +++ b/core/src/main/java/de/bwaldvogel/mongo/wire/message/MongoMessage.java @@ -10,10 +10,16 @@ public class MongoMessage extends ClientRequest { private final Document document; private final int flags = 0; + private final boolean moreToCome; public MongoMessage(Channel channel, MessageHeader header, Document document) { + this(channel, header, document, false); + } + + public MongoMessage(Channel channel, MessageHeader header, Document document, boolean moreToCome) { super(channel, header, null); this.document = document; + this.moreToCome = moreToCome; } public Document getDocument() { @@ -24,6 +30,13 @@ public int getFlags() { return flags; } + /** + * Whether the client set the moreToCome flag: it does not wait for a reply, as for an unacknowledged (w=0) write. + */ + public boolean isMoreToCome() { + return moreToCome; + } + @Override public String getDatabaseName() { return (String) document.get("$db"); diff --git a/core/src/test/java/de/bwaldvogel/mongo/wire/MongoDatabaseHandlerTest.java b/core/src/test/java/de/bwaldvogel/mongo/wire/MongoDatabaseHandlerTest.java index d4a06ab5..f7c65b13 100644 --- a/core/src/test/java/de/bwaldvogel/mongo/wire/MongoDatabaseHandlerTest.java +++ b/core/src/test/java/de/bwaldvogel/mongo/wire/MongoDatabaseHandlerTest.java @@ -4,6 +4,7 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -15,6 +16,7 @@ import de.bwaldvogel.mongo.wire.message.MongoMessage; import de.bwaldvogel.mongo.wire.message.MongoQuery; import io.netty.channel.Channel; +import io.netty.channel.ChannelHandlerContext; class MongoDatabaseHandlerTest { @@ -71,4 +73,36 @@ void testHandleMessageUnknownError() throws Exception { assertThat(responseMessageDoc.get("errmsg")).isEqualTo("Unknown error: unexpected"); assertThat(responseMessageDoc.get("ok")).isEqualTo(0); } + + @Test + void testMessageWithMoreToComeIsHandledWithoutReply() throws Exception { + MongoBackend backend = mock(MongoBackend.class); + Channel channel = mock(Channel.class); + ChannelHandlerContext ctx = mock(ChannelHandlerContext.class); + when(ctx.channel()).thenReturn(channel); + + MongoDatabaseHandler handler = new MongoDatabaseHandler(backend, null); + MongoMessage requestMessage = new MongoMessage(channel, new MessageHeader(0, 0), json("insert: 'coll'"), true); + + handler.channelRead0(ctx, requestMessage); + + verify(backend).handleMessage(requestMessage); + verify(channel, never()).writeAndFlush(any()); + } + + @Test + void testMessageWithoutMoreToComeIsReplied() throws Exception { + MongoBackend backend = mock(MongoBackend.class); + Channel channel = mock(Channel.class); + ChannelHandlerContext ctx = mock(ChannelHandlerContext.class); + when(ctx.channel()).thenReturn(channel); + + MongoDatabaseHandler handler = new MongoDatabaseHandler(backend, null); + MongoMessage requestMessage = new MongoMessage(channel, new MessageHeader(0, 0), json("insert: 'coll'")); + + handler.channelRead0(ctx, requestMessage); + + verify(channel).writeAndFlush(any(MongoMessage.class)); + } + } diff --git a/test-common/src/main/java/de/bwaldvogel/mongo/MongoServerTest.java b/test-common/src/main/java/de/bwaldvogel/mongo/MongoServerTest.java index 531feaf4..3f124215 100644 --- a/test-common/src/main/java/de/bwaldvogel/mongo/MongoServerTest.java +++ b/test-common/src/main/java/de/bwaldvogel/mongo/MongoServerTest.java @@ -24,8 +24,10 @@ import com.mongodb.ConnectionString; import com.mongodb.MongoClientSettings; import com.mongodb.MongoException; +import com.mongodb.WriteConcern; import com.mongodb.client.MongoClient; import com.mongodb.client.MongoClients; +import com.mongodb.client.MongoCollection; import de.bwaldvogel.mongo.backend.CollectionUtils; @@ -74,6 +76,26 @@ void testBindAndGetConnectionStringThenConnect() throws Exception { } } + @Test + @Timeout(TEST_TIMEOUT_SECONDS) + void testUnacknowledgedWrites() throws Exception { + // a single connection, so the count is read after the unacknowledged writes were processed + MongoClientSettings clientSettings = MongoClientSettings.builder() + .applyConnectionString(new ConnectionString(server.bindAndGetConnectionString())) + .applyToConnectionPoolSettings(pool -> pool.maxSize(1)) + .build(); + try (MongoClient client = MongoClients.create(clientSettings)) { + MongoCollection collection = client.getDatabase("testdb").getCollection("testcoll"); + MongoCollection unacknowledged = collection.withWriteConcern(WriteConcern.UNACKNOWLEDGED); + + for (int i = 0; i < 3; i++) { + assertThat(unacknowledged.insertOne(new Document("_id", i)).wasAcknowledged()).isFalse(); + } + + assertThat(collection.countDocuments()).isEqualTo(3); + } + } + @Test @Timeout(TEST_TIMEOUT_SECONDS) void testStopListening() throws Exception {