Support OP_MSG moreToCome for unacknowledged writes

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);

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) {

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");

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));
+    }
+
 }

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<Document> collection = client.getDatabase("testdb").getCollection("testcoll");
+            MongoCollection<Document> 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 {