From e18593efbf3c9bf2c99f1d61a45a656bac4188e4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ivan=20=E2=80=9CCLOVIS=E2=80=9D=20Canet?= Date: Wed, 1 Oct 2025 14:51:25 +0200 Subject: [PATCH] feat(driver-multiplatform-wire): Prototype a request and a response --- .../src/commonMain/kotlin/Message.kt | 48 ++++ .../src/commonMain/kotlin/MessageSection.kt | 51 +++++ .../src/commonMain/kotlin/MongoClient.kt | 208 +++++++++++++++++- .../src/commonTest/kotlin/1_Connect.kt | 19 +- .../src/commonTest/kotlin/Utils.kt | 14 +- 5 files changed, 330 insertions(+), 10 deletions(-) create mode 100644 driver-multiplatform-wire/src/commonMain/kotlin/Message.kt create mode 100644 driver-multiplatform-wire/src/commonMain/kotlin/MessageSection.kt diff --git a/driver-multiplatform-wire/src/commonMain/kotlin/Message.kt b/driver-multiplatform-wire/src/commonMain/kotlin/Message.kt new file mode 100644 index 00000000..51d29e05 --- /dev/null +++ b/driver-multiplatform-wire/src/commonMain/kotlin/Message.kt @@ -0,0 +1,48 @@ +/* + * Copyright (c) 2025-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.multiplatform.BsonDocument +import opensavvy.ktmongo.bson.multiplatform.BsonFactory +import opensavvy.ktmongo.dsl.LowLevelApi + +interface Message { + + /** + * The message type. + * + * The possible opcodes are documented [in the wire protocol documentation](https://www.mongodb.com/docs/manual/reference/mongodb-wire-protocol/#opcodes). + */ + val opcode: Int + + @LowLevelApi + val content: BsonDocument +} + +data object Find : Message { + + override val opcode: Int + get() = 2013 + + @LowLevelApi + override val content: BsonDocument + get() = BsonFactory().buildDocument { + writeString("find", "test-basic") + writeDocument("filter") {} + writeString("\$db", "java-test") + } +} diff --git a/driver-multiplatform-wire/src/commonMain/kotlin/MessageSection.kt b/driver-multiplatform-wire/src/commonMain/kotlin/MessageSection.kt new file mode 100644 index 00000000..bf8be691 --- /dev/null +++ b/driver-multiplatform-wire/src/commonMain/kotlin/MessageSection.kt @@ -0,0 +1,51 @@ +/* + * Copyright (c) 2025-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.multiplatform.BsonDocument + +sealed interface MessageSection { + + val kind: Byte + + class Body( + val document: BsonDocument, + ) : MessageSection { + override val kind: Byte + get() = Body.kind + + override fun toString() = "MessageSection.Body($document)" + + companion object { + const val kind: Byte = 0 + } + } + + class DocumentSequence( + val id: String, + val documents: List, + ) : MessageSection { + override val kind: Byte + get() = DocumentSequence.kind + + override fun toString() = "MessageSection.DocumentSequence('$id': $documents)" + + companion object { + const val kind: Byte = 1 + } + } +} diff --git a/driver-multiplatform-wire/src/commonMain/kotlin/MongoClient.kt b/driver-multiplatform-wire/src/commonMain/kotlin/MongoClient.kt index 2f78762c..4546aa3c 100644 --- a/driver-multiplatform-wire/src/commonMain/kotlin/MongoClient.kt +++ b/driver-multiplatform-wire/src/commonMain/kotlin/MongoClient.kt @@ -18,22 +18,219 @@ 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.Buffer +import kotlinx.io.readIntLe +import kotlinx.io.writeIntLe +import kotlinx.io.writeUByte +import opensavvy.ktmongo.bson.multiplatform.BsonFactory import opensavvy.ktmongo.dsl.LowLevelApi import kotlin.coroutines.CoroutineContext @LowLevelApi interface MongoClient : AutoCloseable { + suspend fun send( + message: Message, + ): ReceiveChannel + companion object } @LowLevelApi private class MultiplatformMongoClient( private val socket: Socket, + private val factory: BsonFactory, + coroutineScope: CoroutineScope, ) : MongoClient { - private val readChannel = socket.openReadChannel() - private val writeChannel = socket.openWriteChannel() + private class Request( + val data: Buffer, + val output: Channel, + ) + + private class Response( + val length: Int, + val requestId: Int, + val responseTo: Int, + val data: Buffer, + val output: SendChannel, + ) + + private val requestChannel = Channel(Channel.RENDEZVOUS) + + private fun log(message: String) { + println("KtMongo • $message") + } + + init { + val sentChannel = Channel>>(Channel.BUFFERED) + val receivedChannel = Channel(Channel.BUFFERED) + + coroutineScope.launch(CoroutineName("ktmongo-actor-writer")) { + sendActor(sentChannel) + } + + coroutineScope.launch(CoroutineName("ktmongo-actor-reader")) { + readActor(sentChannel, receivedChannel) + } + + repeat(3) { + coroutineScope.launch(CoroutineName("ktmongo-actor-parser-$it")) { + parserActor(receivedChannel) + } + } + } + + private suspend fun sendActor( + sentChannel: SendChannel>>, + ) { + val writeChannel = socket.openWriteChannel() + var nextRequestId = 1 + + while (currentCoroutineContext().isActive && !writeChannel.isClosedForWrite) { + val request = requestChannel.receive() + val requestId = nextRequestId++ + + val buffer = Buffer() + buffer.writeIntLe(request.data.size.toInt() + 8) // + the size itself (4) + the request ID (4) + buffer.writeIntLe(requestId) + buffer.write(request.data, request.data.size) + writeChannel.writeBuffer(buffer) + writeChannel.flush() + + log("$requestId was sent") + sentChannel.send(requestId to request.output) + } + } + + private suspend fun readActor( + sentChannel: ReceiveChannel>>, + receivedChannel: SendChannel, + ) = coroutineScope { + val readChannel = socket.openReadChannel() + val waiting = HashMap>() + + while (currentCoroutineContext().isActive && !readChannel.isClosedForRead) { + select { + sentChannel.onReceive { (requestId, output) -> + log("$requestId expects an answer") + waiting[requestId] = output + } + + async { + readChannel.awaitContent(4 * 4) // standard message header size + }.onAwait { isActive -> + if (isActive) { + log("Received a response from the DB.") + val messageLength = readChannel.readInt().asLittleEndian() + val requestId = readChannel.readInt().asLittleEndian() + val responseTo = readChannel.readInt().asLittleEndian() + + log("Received message $requestId in response to $responseTo, of size $messageLength") + + val data = readChannel.readBuffer(messageLength - (4 * 3)) // don't read the fields we already read + + val handler = waiting[responseTo] + ?: error("Received the message $requestId in response to $responseTo, but no known message with ID $responseTo has been sent by this client.") + receivedChannel.send(Response(messageLength, requestId, responseTo, data, handler)) + } else { + log("MongoDB has stopped sending data to us.") + } + } + } + } + } + + private suspend fun parserActor( + receivedChannel: ReceiveChannel, + ) { + for (received in receivedChannel) { + val buffer = received.data + + buffer.readIntLe() + buffer.readIntLe() + + val sections = ArrayList() + + while (buffer.canRead()) { + when (val kind = buffer.readByte()) { + MessageSection.Body.kind -> { + val size = buffer.peek().readIntLe() + sections += MessageSection.Body(factory.readDocument(buffer.readBytes(size))) + } + + MessageSection.DocumentSequence.kind -> { + TODO() + } + + else -> error("Unrecognized section kind $kind in message ${received.requestId} sent as response to ${received.responseTo}") + } + } + + log("Received: $sections") + } + } + + 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 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 + + // Section kind: Body + buffer.writeUByte(0u) + + buffer.write(message.content.toByteArray()) // TODO: avoid copy + + // endregion + + return buffer + } + + override suspend fun send( + message: Message, + ): ReceiveChannel { + val output = Channel() + log("Preparing to write $message…") + val buffer = writeMessage(message) + log("Message to send is: ${message.content}") + requestChannel.send(Request(buffer, output)) + return output + } override fun close() { socket.close() @@ -46,10 +243,11 @@ private class MultiplatformMongoClient( suspend fun MongoClient( hostName: String, port: Int, - dispatcher: CoroutineContext, + factory: BsonFactory = BsonFactory(), + coroutineContext: CoroutineContext, ): MongoClient { - val selectorManager = SelectorManager(dispatcher) + val selectorManager = SelectorManager(coroutineContext + Dispatchers.Default + CoroutineName("ktmongo-socket")) val socket = aSocket(selectorManager).tcp().connect(hostName, port) - return MultiplatformMongoClient(socket) + return MultiplatformMongoClient(socket, factory, CoroutineScope(coroutineContext + CoroutineName("ktmongo-client"))) } diff --git a/driver-multiplatform-wire/src/commonTest/kotlin/1_Connect.kt b/driver-multiplatform-wire/src/commonTest/kotlin/1_Connect.kt index 481b5fa8..810b3fda 100644 --- a/driver-multiplatform-wire/src/commonTest/kotlin/1_Connect.kt +++ b/driver-multiplatform-wire/src/commonTest/kotlin/1_Connect.kt @@ -18,12 +18,12 @@ package opensavvy.ktmongo.multiplatform.wire +import kotlinx.coroutines.delay import opensavvy.ktmongo.dsl.LowLevelApi import opensavvy.prepared.runner.testballoon.preparedSuite -import opensavvy.prepared.suite.config.CoroutineTimeout -import kotlin.time.Duration.Companion.seconds +import opensavvy.prepared.suite.launchInBackground -val ConnectTest by preparedSuite("1. Connect", preparedConfig = CoroutineTimeout(10.seconds)) { +val ConnectTest by preparedSuite { test("Connect to the database") { val client = MongoClient() @@ -34,4 +34,17 @@ val ConnectTest by preparedSuite("1. Connect", preparedConfig = CoroutineTimeout println("Disconnected.") } + test("Send a hello") { + val client = MongoClient() + + val output = client.send(Find) + + launchInBackground { + for (out in output) { + println("Received: $out 2") + } + } + + delay(10) + } } diff --git a/driver-multiplatform-wire/src/commonTest/kotlin/Utils.kt b/driver-multiplatform-wire/src/commonTest/kotlin/Utils.kt index 6860a68a..7fbf27e8 100644 --- a/driver-multiplatform-wire/src/commonTest/kotlin/Utils.kt +++ b/driver-multiplatform-wire/src/commonTest/kotlin/Utils.kt @@ -1,5 +1,5 @@ /* - * Copyright (c) 2025, OpenSavvy and contributors. + * Copyright (c) 2025-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. @@ -24,6 +24,7 @@ import kotlinx.coroutines.* import opensavvy.ktmongo.dsl.LowLevelApi import opensavvy.prepared.runner.testballoon.preparedSuite import opensavvy.prepared.suite.backgroundScope +import opensavvy.prepared.suite.cleanUp import opensavvy.prepared.suite.prepared import opensavvy.prepared.suite.shared @@ -57,7 +58,16 @@ val mongoAddress by shared { val MongoClient by prepared { val socket = mongoAddress() - MongoClient(socket.hostname, socket.port, backgroundScope.coroutineContext) + + MongoClient( + hostName = socket.hostname, + port = socket.port, + coroutineContext = backgroundScope.coroutineContext + ).also { + cleanUp("Close $it") { + it.close() + } + } } val SocketTest by preparedSuite { -- 2.51.2