diff --git a/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt b/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt index adaa6410..9a8a0cb9 100644 --- a/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt +++ b/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt @@ -19,15 +19,13 @@ package opensavvy.ktmongo.multiplatform.wire import io.ktor.network.selector.* import io.ktor.network.sockets.* import io.ktor.utils.io.* -import io.ktor.utils.io.core.* import kotlinx.coroutines.* import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.channels.ReceiveChannel import kotlinx.coroutines.channels.SendChannel import kotlinx.coroutines.selects.select -import kotlinx.io.* import kotlinx.io.Buffer -import opensavvy.ktmongo.bson.multiplatform.BsonDocument +import kotlinx.io.writeIntLe import opensavvy.ktmongo.bson.multiplatform.BsonFactory import opensavvy.ktmongo.dsl.LowLevelApi import kotlin.coroutines.CoroutineContext @@ -199,14 +197,11 @@ private class SocketWireClient( val readSocket = socket.openReadChannel() while (currentCoroutineContext().isActive && !readSocket.isClosedForRead) { - val messageLength = readSocket.readInt().asLittleEndian() - val requestId = readSocket.readInt().asLittleEndian() - val responseTo = readSocket.readInt().asLittleEndian() + val response = readSocket.readResponse() - log("Received message $requestId in response to $responseTo, of size $messageLength") + log("Received message ${response.requestId} in response to ${response.responseTo}, of size ${response.messageLength}") - val data = readSocket.readBuffer(messageLength - (4 * 3)) // don't read the fields we already read - receivedChannel.send(Response(requestId, responseTo, data)) + receivedChannel.send(Response(response.requestId, response.responseTo, response.data)) } } @@ -256,152 +251,18 @@ private class SocketWireClient( receivedChannel: ReceiveChannel, ) { for (received in receivedChannel) { - val buffer = received.response.data - - val opcode = buffer.readIntLe() - check(opcode == 2013) { "Currently, only OP_MSG is supported, but found opcode $opcode" } - - buffer.readIntLe() // flag bits - - val sections = ArrayList() - - while (buffer.canRead()) { - when (val kind = buffer.readUByte()) { - MessageSection.Body.kind -> { - val size = buffer.peek().readIntLe() - sections += MessageSection.Body(eager(factory.readDocument(buffer.readBytes(size)))) // TODO: avoid copy - } - - MessageSection.DocumentSequence.kind -> { - // • section size - val size = buffer.readIntLe() - 4 - var read = 4 - - // • section id - val id = buffer.readCString() - read += id.length - read += 1 // null terminator - - // • section documents - val documents = ArrayList>() - while (read < size) { - val documentSize = buffer.peek().readIntLe() - documents += eager(factory.readDocument(buffer.readBytes(documentSize))) // TODO: avoid copy - } - - sections += MessageSection.DocumentSequence(id, documents) - } - - else -> error("Unrecognized section kind $kind in message ${received.response.requestId} sent as response to ${received.response.responseTo}") - } - } - - log("Received: $sections") - - val body = sections.singleOrNull { it is MessageSection.Body } as? MessageSection.Body - ?: error("An OP_MSG message must have a single body section, found: $sections") - - val response = Message.OpMsg( - body, - sections.asSequence() - .filterIsInstance(), + val message = received.response.data.parseMessage( + factory = factory, + requestId = received.response.requestId, + responseTo = received.response.responseTo, ) - when (received.output) { - is ResponseHandler.Single -> received.output.result.complete(response) - is ResponseHandler.Multiple -> received.output.result.send(response) - } - } - } - - private fun Int.asLittleEndian(): Int { - return ((this and 0xFF) shl 24) or - ((this and 0xFF00) shl 8) or - ((this and 0xFF0000) shr 8) or - ((this and 0xFF000000.toInt()) ushr 24) - } - - private fun Buffer.writeCString(value: String) { - val text = value - .takeUnless { 0.toChar() in it } - ?: value.filterNot { it == 0.toChar() } - - writeString(text) - writeUByte(0u) - } + log("Received: $message") - private fun Buffer.readCString(): String { - val peek = peek() - var byteCount = 0L - while (peek.request(1) && peek.readByte() != 0.toByte()) - byteCount++ - - return readString(byteCount) - .also { skip(1) } // null-terminator - } - - private fun writeMessage(message: Message): Buffer { - val buffer = Buffer() - - // region Message header - // https://www.mongodb.com/docs/manual/reference/mongodb-wire-protocol/#standard-message-header - - // Writes the complete message to the buffer EXCEPT the first 2 fields: - // • message length - // • request ID - // The writer actor will add these two fields. - - // • response to - buffer.writeIntLe(0) - - // • opcode - buffer.writeIntLe(message.opcode) - - // endregion - // region Message flags - // https://www.mongodb.com/docs/manual/reference/mongodb-wire-protocol/#flag-bits - - buffer.writeIntLe(0) - - // endregion - // region Sections - - when (message) { - is Message.OpMsg -> { - writeOpMsg(message, buffer) - } - } - - // endregion - - return buffer - } - - private fun writeOpMsg(message: Message.OpMsg, buffer: Buffer) { - // First, write the body (any order is allowed in the spec) - - // • body section kind - buffer.writeUByte(message.body.kind) - - // • body content - buffer.write(message.body.document.toByteArray()) // TODO: avoid copy - - // Next, read the sequences, if any - for (sequence in message.sequences) { - // • section kind - buffer.writeUByte(sequence.kind) - - val payload = Buffer() - payload.writeCString(sequence.id) - for (document in sequence.documents) { - payload.write(document.toByteArray()) // TODO: avoid copy + when (received.output) { + is ResponseHandler.Single -> received.output.result.complete(message) + is ResponseHandler.Multiple -> received.output.result.send(message) } - - // • size - buffer.writeIntLe(payload.size.toInt() + 4) - - // • documents - buffer.write(payload, payload.size) } } diff --git a/driver-multiplatform-wire/src/commonMain/kotlin/Primitives.read.kt b/driver-multiplatform-wire/src/commonMain/kotlin/Primitives.read.kt new file mode 100644 index 00000000..b46e0729 --- /dev/null +++ b/driver-multiplatform-wire/src/commonMain/kotlin/Primitives.read.kt @@ -0,0 +1,117 @@ +/* + * Copyright (c) 2026, OpenSavvy and contributors. + * + * 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 opensavvy.ktmongo.multiplatform.wire + +import io.ktor.utils.io.* +import io.ktor.utils.io.core.* +import kotlinx.io.Buffer +import kotlinx.io.readIntLe +import kotlinx.io.readString +import kotlinx.io.readUByte +import opensavvy.ktmongo.bson.multiplatform.BsonDocument +import opensavvy.ktmongo.bson.multiplatform.BsonFactory + +internal class ResponsePayload( + val messageLength: Int, + val requestId: Int, + val responseTo: Int, + val data: Buffer, +) + +internal suspend fun ByteReadChannel.readResponse(): ResponsePayload { + val messageLength = readInt().asLittleEndian() + val requestId = readInt().asLittleEndian() + val responseTo = readInt().asLittleEndian() + + val data = readBuffer(messageLength - (4 * 3)) // don't read the fields we already read + + return ResponsePayload(messageLength, requestId, responseTo, data) +} + +internal fun Buffer.parseMessage( + factory: BsonFactory, + requestId: Int, + responseTo: Int, +): Message { + val buffer = this + + val opcode = buffer.readIntLe() + check(opcode == 2013) { "Currently, only OP_MSG is supported, but found opcode $opcode" } + + buffer.readIntLe() // flag bits + + val sections = ArrayList() + + while (buffer.canRead()) { + when (val kind = buffer.readUByte()) { + MessageSection.Body.kind -> { + val size = buffer.peek().readIntLe() + sections += MessageSection.Body(eager(factory.readDocument(buffer.readBytes(size)))) // TODO: avoid copy + } + + MessageSection.DocumentSequence.kind -> { + // • section size + val size = buffer.readIntLe() - 4 + var read = 4 + + // • section id + val id = buffer.readCString() + read += id.length + read += 1 // null terminator + + // • section documents + val documents = ArrayList>() + while (read < size) { + val documentSize = buffer.peek().readIntLe() + documents += eager(factory.readDocument(buffer.readBytes(documentSize))) // TODO: avoid copy + } + + sections += MessageSection.DocumentSequence(id, documents) + } + + else -> error("Unrecognized section kind $kind in message $requestId sent as response to $responseTo") + } + } + + val body = sections.singleOrNull { it is MessageSection.Body } as? MessageSection.Body + ?: error("An OP_MSG message must have a single body section, found: $sections") + + val response = Message.OpMsg( + body, + sections.asSequence() + .filterIsInstance(), + ) + + return response +} + +private fun Int.asLittleEndian(): Int { + return ((this and 0xFF) shl 24) or + ((this and 0xFF00) shl 8) or + ((this and 0xFF0000) shr 8) or + ((this and 0xFF000000.toInt()) ushr 24) +} + +private fun Buffer.readCString(): String { + val peek = peek() + var byteCount = 0L + while (peek.request(1) && peek.readByte() != 0.toByte()) + byteCount++ + + return readString(byteCount) + .also { skip(1) } // null-terminator +} diff --git a/driver-multiplatform-wire/src/commonMain/kotlin/Primitives.write.kt b/driver-multiplatform-wire/src/commonMain/kotlin/Primitives.write.kt new file mode 100644 index 00000000..f3f0ec2a --- /dev/null +++ b/driver-multiplatform-wire/src/commonMain/kotlin/Primitives.write.kt @@ -0,0 +1,96 @@ +/* + * Copyright (c) 2026, OpenSavvy and contributors. + * + * 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 opensavvy.ktmongo.multiplatform.wire + +import kotlinx.io.Buffer +import kotlinx.io.writeIntLe +import kotlinx.io.writeString +import kotlinx.io.writeUByte + +internal fun writeMessage(message: Message): Buffer { + val buffer = Buffer() + + // region Message header + // https://www.mongodb.com/docs/manual/reference/mongodb-wire-protocol/#standard-message-header + + // Writes the complete message to the buffer EXCEPT the first 2 fields: + // • message length + // • request ID + // The writer actor will add these two fields. + + // • response to + buffer.writeIntLe(0) + + // • opcode + buffer.writeIntLe(message.opcode) + + // endregion + // region Message flags + // https://www.mongodb.com/docs/manual/reference/mongodb-wire-protocol/#flag-bits + + buffer.writeIntLe(0) + + // endregion + // region Sections + + when (message) { + is Message.OpMsg -> { + writeOpMsg(message, buffer) + } + } + + // endregion + + return buffer +} + +internal fun writeOpMsg(message: Message.OpMsg, buffer: Buffer) { + // First, write the body (any order is allowed in the spec) + + // • body section kind + buffer.writeUByte(message.body.kind) + + // • body content + buffer.write(message.body.document.toByteArray()) // TODO: avoid copy + + // Next, read the sequences, if any + for (sequence in message.sequences) { + // • section kind + buffer.writeUByte(sequence.kind) + + val payload = Buffer() + payload.writeCString(sequence.id) + for (document in sequence.documents) { + payload.write(document.toByteArray()) // TODO: avoid copy + } + + // • size + buffer.writeIntLe(payload.size.toInt() + 4) + + // • documents + buffer.write(payload, payload.size) + } +} + +private fun Buffer.writeCString(value: String) { + val text = value + .takeUnless { 0.toChar() in it } + ?: value.filterNot { it == 0.toChar() } + + writeString(text) + writeUByte(0u) +} -- 2.51.2 From 111801b1b5753103d6406e0f65c0b7d9cdeb6f3c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ivan=20=E2=80=9CCLOVIS=E2=80=9D=20Canet?= Date: Sat, 15 Aug 2026 11:17:30 +0200 Subject: [PATCH 2/4] refactor(driver-multiplatform-wire): Introduce a MongoSocket abstraction --- .../src/commonMain/kotlin/MongoSocket.kt | 60 +++++++++++++++++++ .../src/commonMain/kotlin/MongoWireClient.kt | 11 ++-- 2 files changed, 64 insertions(+), 7 deletions(-) create mode 100644 driver-multiplatform-wire/src/commonMain/kotlin/MongoSocket.kt diff --git a/driver-multiplatform-wire/src/commonMain/kotlin/MongoSocket.kt b/driver-multiplatform-wire/src/commonMain/kotlin/MongoSocket.kt new file mode 100644 index 00000000..5c1d5c01 --- /dev/null +++ b/driver-multiplatform-wire/src/commonMain/kotlin/MongoSocket.kt @@ -0,0 +1,60 @@ +/* + * Copyright (c) 2026, OpenSavvy and contributors. + * + * 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 opensavvy.ktmongo.multiplatform.wire + +import io.ktor.network.selector.* +import io.ktor.network.sockets.* +import io.ktor.utils.io.* +import kotlinx.coroutines.isActive + +/** + * Custom abstraction to allow creating fake sockets for testing. + */ +internal interface MongoSocket : AutoCloseable { + + val isActive: Boolean + + fun openReadChannel(): ByteReadChannel + + fun openWriteChannel(): ByteWriteChannel +} + +private class KtorMongoSocket( + private val socket: Socket, + private val selectorManager: SelectorManager, +) : MongoSocket { + + override val isActive: Boolean + get() = socket.isActive + + override fun openReadChannel(): ByteReadChannel = + socket.openReadChannel() + + override fun openWriteChannel(): ByteWriteChannel = + socket.openWriteChannel() + + override fun close() { + socket.close() + selectorManager.close() + } + + override fun toString(): String = + "Ktor ${socket.remoteAddress}" +} + +internal fun MongoSocket(socket: Socket, selectorManager: SelectorManager): MongoSocket = + KtorMongoSocket(socket, selectorManager) diff --git a/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt b/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt index 9a8a0cb9..c1a6e445 100644 --- a/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt +++ b/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt @@ -64,8 +64,7 @@ interface MongoWireClient : AutoCloseable { */ @LowLevelApi private class SocketWireClient( - private val socket: Socket, - private val selectorManager: SelectorManager, + private val socket: MongoSocket, private val factory: BsonFactory, coroutineScope: CoroutineScope, // Should contain a Job dedicated to this client ) : MongoWireClient { @@ -113,7 +112,7 @@ private class SocketWireClient( } init { - log("Creating client for socket ${socket.remoteAddress}") + log("Creating client for socket $socket") // Ensure that no resources can leak actorsJob.invokeOnCompletion { close() } @@ -288,10 +287,9 @@ private class SocketWireClient( override fun close() { actorsJob.cancel("${this::class}.close() has been called") socket.close() - selectorManager.close() } - override fun toString() = "MongoWireClient(${socket.remoteAddress})" + override fun toString() = "MongoWireClient($socket)" } @LowLevelApi @@ -309,8 +307,7 @@ suspend fun MongoWireClient( } return SocketWireClient( - socket = socket, - selectorManager = selectorManager, + socket = MongoSocket(socket, selectorManager), factory = factory, coroutineScope = CoroutineScope(coroutineContext + innerJob + CoroutineName("ktmongo-client")) ) -- 2.51.2 From 219d801169cf24782dc4c11a4b1a594d999bcd23 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ivan=20=E2=80=9CCLOVIS=E2=80=9D=20Canet?= Date: Sat, 15 Aug 2026 11:52:00 +0200 Subject: [PATCH 3/4] test(driver-multiplatform-wire): Use a fake server to verify specific scenarii --- .../src/commonMain/kotlin/MongoWireClient.kt | 17 ++ .../src/commonTest/kotlin/fake/FakeServer.kt | 254 ++++++++++++++++++ .../commonTest/kotlin/fake/FakeServerTest.kt | 76 ++++++ 3 files changed, 347 insertions(+) create mode 100644 driver-multiplatform-wire/src/commonTest/kotlin/fake/FakeServer.kt create mode 100644 driver-multiplatform-wire/src/commonTest/kotlin/fake/FakeServerTest.kt diff --git a/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt b/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt index c1a6e445..224decd8 100644 --- a/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt +++ b/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt @@ -292,6 +292,23 @@ private class SocketWireClient( override fun toString() = "MongoWireClient($socket)" } +/** + * Creates a [MongoWireClient] wrapping an existing [socket]. + * + * Used by tests to inject fake sockets instead of connecting to a real server. + */ +@LowLevelApi +internal fun MongoWireClient( + socket: MongoSocket, + factory: BsonFactory = BsonFactory(), + coroutineScope: CoroutineScope, +): MongoWireClient = + SocketWireClient( + socket = socket, + factory = factory, + coroutineScope = coroutineScope, + ) + @LowLevelApi suspend fun MongoWireClient( hostName: String, diff --git a/driver-multiplatform-wire/src/commonTest/kotlin/fake/FakeServer.kt b/driver-multiplatform-wire/src/commonTest/kotlin/fake/FakeServer.kt new file mode 100644 index 00000000..d3ffcb9a --- /dev/null +++ b/driver-multiplatform-wire/src/commonTest/kotlin/fake/FakeServer.kt @@ -0,0 +1,254 @@ +/* + * Copyright (c) 2026, OpenSavvy and contributors. + * + * 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. + */ + +@file:OptIn(LowLevelApi::class, ExperimentalBsonDiffApi::class) + +package opensavvy.ktmongo.multiplatform.wire.fake + +import io.ktor.utils.io.* +import kotlinx.coroutines.* +import kotlinx.io.Buffer +import kotlinx.io.writeIntLe +import opensavvy.ktmongo.bson.ExperimentalBsonDiffApi +import opensavvy.ktmongo.bson.diff +import opensavvy.ktmongo.bson.multiplatform.BsonFactory +import opensavvy.ktmongo.dsl.LowLevelApi +import opensavvy.ktmongo.multiplatform.wire.* +import opensavvy.prepared.suite.TestDsl +import opensavvy.prepared.suite.cleanUp +import opensavvy.prepared.suite.foregroundScope + +private fun logFake(message: String) { + println("» Fake: $message") +} + +/** + * A fake in-memory [MongoSocket], backed by an in-memory pipe instead of a real network connection. + * + * Reading from this socket returns the bytes written to its [peer][FakeMongoSocket] socket, and vice-versa. + */ +private class FakeMongoSocket( + private val readChannel: ByteReadChannel, + private val writeChannel: ByteWriteChannel, +) : MongoSocket { + + override val isActive: Boolean + get() = !readChannel.isClosedForRead && !writeChannel.isClosedForWrite + + override fun openReadChannel(): ByteReadChannel = readChannel + + override fun openWriteChannel(): ByteWriteChannel = writeChannel + + override fun close() { + readChannel.cancel(null) + writeChannel.cancel(null) + } + + override fun toString() = "Fake socket" + + companion object { + /** + * Creates a pair of [FakeMongoSocket]s connected to each other, as if by a network connection: + * anything written to one of them can be read from the other one. + */ + fun createLinked(): Pair { + val clientToServer = ByteChannel() + val serverToClient = ByteChannel() + + val client = FakeMongoSocket(readChannel = serverToClient, writeChannel = clientToServer) + val server = FakeMongoSocket(readChannel = clientToServer, writeChannel = serverToClient) + + return client to server + } + } +} + +/** + * Writes a full wire-protocol frame for [message], as a response to the request identified by [responseTo]. + */ +private fun writeResponseFrame(message: Message.OpMsg, requestId: Int, responseTo: Int): Buffer { + val payload = Buffer() + payload.writeIntLe(responseTo) + payload.writeIntLe(message.opcode) + payload.writeIntLe(0) // flag bits + writeOpMsg(message, payload) + + val frame = Buffer() + frame.writeIntLe(payload.size.toInt() + 8) // + the size itself (4) + the request ID (4) + frame.writeIntLe(requestId) + frame.write(payload, payload.size) + return frame +} + +@DslMarker +annotation class FakeServerDsl + +@FakeServerDsl +class FakeServer private constructor( + private val currentTest: TestDsl, + private val scenario: FakeServerScenario, +) : AutoCloseable { + + private val sockets = FakeMongoSocket.createLinked() + private val clientSocket get() = sockets.first + private val serverSocket get() = sockets.second + + init { + currentTest.foregroundScope.launch(CoroutineName("fake-server")) { + run() + } + } + + override fun close() { + serverSocket.close() + clientSocket.close() + } + + private suspend fun run() { + val readChannel = serverSocket.openReadChannel() + val writeChannel = serverSocket.openWriteChannel() + + var lastRequestId = 0 + var nextResponseId = 1 + + for (event in scenario.events) { + when (event) { + is FakeServerScenario.Event.Expect -> { + lastRequestId = verifyExpect(event, readChannel) + } + + is FakeServerScenario.Event.Respond -> { + verifyRespond(event, writeChannel, requestId = nextResponseId++, responseTo = lastRequestId) + } + } + } + + logFake("No more events to execute.") + serverSocket.close() + } + + private suspend fun verifyExpect( + event: FakeServerScenario.Event.Expect, + readChannel: ByteReadChannel, + ): Int { + logFake("Expecting $event") + val expected = event.message + check(expected is Message.OpMsg) { "Other kinds of messages are not supported yet" } + + val responsePayload = readChannel.readResponse() + + val actual = responsePayload.data.parseMessage( + factory = BsonFactory(), + requestId = responsePayload.requestId, + responseTo = responsePayload.responseTo, + ) + + logFake("Received $actual") + + check(actual is Message.OpMsg) { "Other kinds of messages are not supported yet" } + check(actual.body.document == expected.body.document) { "The received document doesn't match the expected document:\n${actual.body.document diff expected.body.document}" } + + val expectedSequences = expected.sequences.toList() + .sortedBy { it.id } + val actualSequences = actual.sequences.toList() + .sortedBy { it.id } + + check(expectedSequences.size == actualSequences.size) + + for ((expectedSequence, actualSequence) in expectedSequences.zip(actualSequences)) { + check(expectedSequence.id == actualSequence.id) + + val expectedDocuments = expectedSequence.documents.toList() + val actualDocuments = actualSequence.documents.toList() + check(expectedDocuments.size == actualDocuments.size) + + for ((expectedDocument, actualDocument) in expectedDocuments.zip(actualDocuments)) { + check(expectedDocument == actualDocument) { "${expectedDocument diff actualDocument}" } + } + } + + logFake("Expectation verified") + + return responsePayload.requestId + } + + private suspend fun verifyRespond( + event: FakeServerScenario.Event.Respond, + writeChannel: ByteWriteChannel, + requestId: Int, + responseTo: Int, + ) { + logFake("Responding $event") + val message = event.message + check(message is Message.OpMsg) { "Other kinds of messages are not supported yet" } + + val frame = writeResponseFrame(message, requestId, responseTo) + writeChannel.writeBuffer(frame) + writeChannel.flush() + + logFake("Response sent") + } + + suspend fun createClient(): MongoWireClient { + return MongoWireClient( + socket = clientSocket, + coroutineScope = CoroutineScope(currentTest.foregroundScope.coroutineContext + Job(currentTest.foregroundScope.coroutineContext.job)), + ).also { + currentTest.cleanUp("Fake client") { + it.close() + } + } + } + + @FakeServerDsl + companion object { + suspend fun TestDsl.fakeServer( + stub: FakeServerScenario.() -> Unit, + ): FakeServer { + val scenario = FakeServerScenario().apply(stub) + + return FakeServer(this, scenario) + .also { + cleanUp("Fake server") { + it.close() + } + } + } + } +} + +@FakeServerDsl +class FakeServerScenario { + sealed class Event { + data class Expect(val message: Message) : Event() { + override fun toString() = "Expect $message" + } + + data class Respond(val message: Message) : Event() { + override fun toString() = "Respond $message" + } + } + + val events = ArrayList() + + fun expect(message: Message) { + events += Event.Expect(message) + } + + fun respond(message: Message) { + events += Event.Respond(message) + } +} diff --git a/driver-multiplatform-wire/src/commonTest/kotlin/fake/FakeServerTest.kt b/driver-multiplatform-wire/src/commonTest/kotlin/fake/FakeServerTest.kt new file mode 100644 index 00000000..b53820e9 --- /dev/null +++ b/driver-multiplatform-wire/src/commonTest/kotlin/fake/FakeServerTest.kt @@ -0,0 +1,76 @@ +/* + * Copyright (c) 2026, OpenSavvy and contributors. + * + * 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. + */ + +@file:OptIn(LowLevelApi::class) + +package opensavvy.ktmongo.multiplatform.wire.fake + +import opensavvy.ktmongo.bson.multiplatform.BsonFactory +import opensavvy.ktmongo.dsl.LowLevelApi +import opensavvy.ktmongo.multiplatform.wire.Message +import opensavvy.ktmongo.multiplatform.wire.MessageSection +import opensavvy.ktmongo.multiplatform.wire.eager +import opensavvy.ktmongo.multiplatform.wire.fake.FakeServer.Companion.fakeServer +import opensavvy.prepared.runner.testballoon.preparedSuite + +val FakeServerTest by preparedSuite { + + test("Create a fake server") { + val server = fakeServer {} + println(server) + } + + test("Create a fake client") { + val client = fakeServer {} + .createClient() + println(client) + } + + test("Round-trip hello") { + val helloMessage = Message.OpMsg( + MessageSection.Body( + eager( + BsonFactory().buildDocument { + writeInt32("hello", 1) + } + ) + ) + ) + + val okMessage = Message.OpMsg( + MessageSection.Body( + eager( + BsonFactory().buildDocument { + writeDouble("ok", 1.0) + } + ) + ) + ) + + val server = fakeServer { + expect(helloMessage) + respond(okMessage) + } + + val client = server.createClient() + + val response = client.sendSingle(helloMessage) + + check(response is Message.OpMsg) + check(response.body.document["ok"]?.decodeDouble() == 1.0) + } + +} -- 2.51.2 From e99ba36bab059427f50ce5e5f9ab7da6b8a7ffe4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ivan=20=E2=80=9CCLOVIS=E2=80=9D=20Canet?= Date: Sat, 15 Aug 2026 12:10:23 +0200 Subject: [PATCH 4/4] test(driver-multiplatform-wire): Simplify the creation of OpMsg messages --- .../src/commonTest/kotlin/OpMsgUtils.kt | 51 +++++++++++++++++++ .../commonTest/kotlin/fake/FakeServerTest.kt | 28 +++------- 2 files changed, 58 insertions(+), 21 deletions(-) create mode 100644 driver-multiplatform-wire/src/commonTest/kotlin/OpMsgUtils.kt diff --git a/driver-multiplatform-wire/src/commonTest/kotlin/OpMsgUtils.kt b/driver-multiplatform-wire/src/commonTest/kotlin/OpMsgUtils.kt new file mode 100644 index 00000000..6c6d7936 --- /dev/null +++ b/driver-multiplatform-wire/src/commonTest/kotlin/OpMsgUtils.kt @@ -0,0 +1,51 @@ +/* + * Copyright (c) 2026, OpenSavvy and contributors. + * + * 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 opensavvy.ktmongo.multiplatform.wire + +import opensavvy.ktmongo.bson.BsonFieldWriter +import opensavvy.ktmongo.bson.multiplatform.BsonFactory +import opensavvy.ktmongo.dsl.LowLevelApi +import opensavvy.ktmongo.multiplatform.wire.Message.OpMsg + +/** + * Creates an [OpMsg] message with the given [body] and no [OpMsg.sequences]. + */ +@OptIn(LowLevelApi::class) +fun OpMsg(body: BsonFieldWriter.() -> Unit): OpMsg = + OpMsg( + MessageSection.Body( + eager( + BsonFactory().buildDocument(body) + ) + ), + sequences = emptySequence() + ) + +/** + * Creates a copy of this [OpMsg], concatenating a new sequence named [id] and composed of the given [documents]. + */ +@OptIn(LowLevelApi::class) +fun OpMsg.withSequence( + id: String, + vararg documents: BsonFieldWriter.() -> Unit, +): OpMsg = OpMsg( + body, + sequences + MessageSection.DocumentSequence( + id, + documents.map { eager(BsonFactory().buildDocument(it)) } + ) +) diff --git a/driver-multiplatform-wire/src/commonTest/kotlin/fake/FakeServerTest.kt b/driver-multiplatform-wire/src/commonTest/kotlin/fake/FakeServerTest.kt index b53820e9..4fb05486 100644 --- a/driver-multiplatform-wire/src/commonTest/kotlin/fake/FakeServerTest.kt +++ b/driver-multiplatform-wire/src/commonTest/kotlin/fake/FakeServerTest.kt @@ -18,11 +18,9 @@ package opensavvy.ktmongo.multiplatform.wire.fake -import opensavvy.ktmongo.bson.multiplatform.BsonFactory import opensavvy.ktmongo.dsl.LowLevelApi import opensavvy.ktmongo.multiplatform.wire.Message -import opensavvy.ktmongo.multiplatform.wire.MessageSection -import opensavvy.ktmongo.multiplatform.wire.eager +import opensavvy.ktmongo.multiplatform.wire.OpMsg import opensavvy.ktmongo.multiplatform.wire.fake.FakeServer.Companion.fakeServer import opensavvy.prepared.runner.testballoon.preparedSuite @@ -40,25 +38,13 @@ val FakeServerTest by preparedSuite { } test("Round-trip hello") { - val helloMessage = Message.OpMsg( - MessageSection.Body( - eager( - BsonFactory().buildDocument { - writeInt32("hello", 1) - } - ) - ) - ) + val helloMessage = OpMsg { + writeInt32("hello", 1) + } - val okMessage = Message.OpMsg( - MessageSection.Body( - eager( - BsonFactory().buildDocument { - writeDouble("ok", 1.0) - } - ) - ) - ) + val okMessage = OpMsg { + writeDouble("ok", 1.0) + } val server = fakeServer { expect(helloMessage)