feat(monkeys): establish connections between users [WPB-2855] (#1860)

monkeys/src/commonMain/kotlin/com/wire/kalium/monkeys/DefaultConversationCreation.kt

@@ -22,15 +22,16 @@ import com.wire.kalium.logic.feature.conversation.CreateGroupConversationUseCase
 
 class DefaultConversationCreation : ConversationCreation {
     override suspend fun invoke(
-        monkeyGroups: List<List<Monkey>>
+        monkeyGroups: List<List<Monkey>>,
+        protocol: ConversationOptions.Protocol
     ): List<MonkeyConversation> = monkeyGroups.map { group ->
         val groupCreator = group.first()
         val userScope = groupCreator.operationScope
 
         val conversationResult = userScope.conversations.createGroupConversation(
             name = "By Monkey '${groupCreator.user.email}'",
             userIdList = group.map { it.user.userId },
-            options = ConversationOptions(protocol = MonkeyApplication.GROUP_TYPE)
+            options = ConversationOptions(protocol = protocol)
         )
 
         if (conversationResult !is CreateGroupConversationUseCase.Result.Success) {

monkeys/src/commonMain/kotlin/com/wire/kalium/monkeys/DefaultSetup.kt

@@ -29,8 +29,6 @@ import com.wire.kalium.logic.feature.client.RegisterClientResult
 import com.wire.kalium.logic.feature.client.RegisterClientUseCase
 import com.wire.kalium.logic.feature.client.SelfClientsResult
 import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.async
-import kotlinx.coroutines.awaitAll
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.launch
 import kotlinx.datetime.Instant
@@ -46,21 +44,63 @@ class DefaultSetup(setupParallelism: Int) : SetupStep {
         println("### LOGGING IN ALL USERS")
         val monkeyGroups = accountGroups.map { group ->
             group.map { accountData ->
-                async(dispatcher) {
-                    println("### Getting versioned backend")
-                    val authScope = getAuthScope(coreLogic, accountData.backend)
-                    acquireMonkeySessions(accountData, authScope, coreLogic)
-                }
-            }.awaitAll()
+                println("### Getting versioned backend")
+                val authScope = getAuthScope(coreLogic, accountData.backend)
+                acquireMonkeySessions(accountData, authScope, coreLogic)
+            }
         }
 
         registerAllClients(monkeyGroups)
 
-        // TODO: Connection requests between multiple federated backends
+        establishConnectionBetweenMonkeys(monkeyGroups)
 
         monkeyGroups
     }
 
+    private suspend fun establishConnectionBetweenMonkeys(monkeyGroups: List<List<Monkey>>) {
+        val backendMonkeyMap = monkeyGroups.flatten()
+            .groupBy { it.user.backend }
+
+        val alreadyProcessedBackends = mutableSetOf<Backend>()
+        for ((backend, monkeys) in backendMonkeyMap) {
+            for ((otherBackend, otherMonkeys) in backendMonkeyMap) {
+                if (otherBackend == backend) continue
+                if (alreadyProcessedBackends.contains(otherBackend)) {
+                    // Accept connection requests
+                    acceptAllConnectionRequests(monkeys, otherMonkeys)
+                } else {
+                    // Send connection requests
+                    sendConnectionRequests(monkeys, otherMonkeys)
+                }
+            }
+            alreadyProcessedBackends += backend
+        }
+    }
+
+    private suspend fun sendConnectionRequests(
+        senders: List<Monkey>,
+        otherMonkeys: List<Monkey>
+    ) {
+        senders.forEach { monkey ->
+            otherMonkeys.forEach { otherMonkey ->
+                monkey.operationScope.connection.sendConnectionRequest(otherMonkey.user.userId)
+                // TODO: Handle error
+            }
+        }
+    }
+
+    private suspend fun acceptAllConnectionRequests(
+        receivers: List<Monkey>,
+        otherMonkeys: List<Monkey>
+    ) {
+        receivers.forEach { monkey ->
+            otherMonkeys.forEach { otherMonkey ->
+                monkey.operationScope.connection.acceptConnectionRequest(otherMonkey.user.userId)
+                // TODO: Handle error
+            }
+        }
+    }
+
     private suspend fun registerAllClients(monkeyGroups: List<List<Monkey>>) = coroutineScope {
         for (monkey in monkeyGroups.flatten()) {
             val userData = monkey.user

monkeys/src/commonMain/kotlin/com/wire/kalium/monkeys/MonkeyApplication.kt

@@ -24,6 +24,7 @@ import co.touchlab.kermit.LogWriter
 import com.github.ajalt.clikt.core.CliktCommand
 import com.github.ajalt.clikt.parameters.arguments.argument
 import com.github.ajalt.clikt.parameters.options.default
+import com.github.ajalt.clikt.parameters.options.flag
 import com.github.ajalt.clikt.parameters.options.option
 import com.github.ajalt.clikt.parameters.types.enum
 import com.wire.kalium.logger.KaliumLogLevel
@@ -40,6 +41,7 @@ class MonkeyApplication : CliktCommand(allowMultipleSubcommands = true) {
     private val logLevel by option(help = "log level").enum<KaliumLogLevel>().default(KaliumLogLevel.VERBOSE)
     private val logOutputFile by option(help = "output file for logs")
     private val fileLogger: LogWriter by lazy { fileLogger(logOutputFile ?: "kalium.log") }
+    private val proteus by option(help = "run using proteus protocol").flag()
 
     override fun run() = runBlocking {
         val coreLogic = coreLogic(
@@ -67,17 +69,17 @@ class MonkeyApplication : CliktCommand(allowMultipleSubcommands = true) {
         users: List<UserData>,
         testSequence: TestSequence
     ) = with(testSequence) {
+        val protocol = if (proteus) ConversationOptions.Protocol.PROTEUS else ConversationOptions.Protocol.MLS
         val monkeyGroups = split(users)
         val monkeyScopes = setup(coreLogic, monkeyGroups)
-        val conversations = createConversations(monkeyScopes)
+        val conversations = createConversations(monkeyScopes, protocol)
         commands.forEach { command ->
             command(conversations)
         }
     }
 
     companion object {
         val HOME_DIRECTORY: String = homeDirectory()
-        val GROUP_TYPE = ConversationOptions.Protocol.MLS
     }
 
 }

monkeys/src/commonMain/kotlin/com/wire/kalium/monkeys/TestSequence.kt

@@ -48,7 +48,10 @@ fun interface MonkeyCommand {
  * Creates a conversation between a watch monkey and a list of monkeys.
  */
 fun interface ConversationCreation {
-    suspend operator fun invoke(monkeyGroups: List<List<Monkey>>): List<MonkeyConversation>
+    suspend operator fun invoke(
+        monkeyGroups: kotlin.collections.List<kotlin.collections.List<com.wire.kalium.monkeys.Monkey>>,
+        protocol: com.wire.kalium.logic.data.conversation.ConversationOptions.Protocol
+    ): List<MonkeyConversation>
 }
 
 /**

monkeys/src/commonMain/kotlin/com/wire/kalium/monkeys/command/BananaPerGroupCommand.kt

@@ -46,6 +46,7 @@ class BananaPerGroupCommand(private val amountOfMessages: Int) : MonkeyCommand {
             }
 
             if (firstConversation != null) {
+                println("Requesting banana=$currentMessage from ${randomUser.user.email}")
                 userScope.messages.sendTextMessage(
                     firstConversation.id,
                     "give me $currentMessage bananas! ${emoji.random()}",

monkeys/users.json

@@ -7,7 +7,7 @@
       "blackList": "https://clientblacklist.wire.com/staging",
       "teams": "https://teams.anta.wire.link/",
       "website": "https://example.org",
-      "title": "Anta Backend",
+      "title": "anta.wire.link",
       "passwordForUsers": "Aqa123456!",
       "domain": "anta.wire.link",
       "users": [
@@ -20,6 +20,56 @@
           "id": "33ec2e20-9bed-44b2-b891-a1dbaf3da158"
         }
       ]
+    },
+    {
+      "api": "https://nginz-https.bella.wire.link",
+      "webSocket": "https://nginz-ssl.bella.wire.link",
+      "blackList": "https://clientblacklist.wire.com/staging",
+      "teams": "https://teams.bella.wire.link",
+      "accounts": "https://account.bella.wire.link",
+      "website": "https://wire.com",
+      "title": "bella.wire.link",
+      "passwordForUsers": "Aqa123456!",
+      "domain": "bella.wire.link",
+      "users": [
+        {
+          "id": "37a27c29-398b-44c6-9334-79cf84024f74",
+          "email": "[email protected]"
+        },
+        {
+          "id": "56f06ac1-69cb-43ab-96f1-816f6ddd6edb",
+          "email": "[email protected]"
+        },
+        {
+          "id": "691e3c09-45e1-4845-89f4-121fef480ec8",
+          "email": "[email protected]"
+        }
+      ]
+    },
+    {
+      "api": "https://nginz-https.chala.wire.link",
+      "webSocket": "https://nginz-ssl.chala.wire.link",
+      "blackList": "https://clientblacklist.wire.com/staging",
+      "teams": "https://teams.chala.wire.link",
+      "accounts": "https://account.chala.wire.link",
+      "website": "https://wire.com",
+      "passwordForUsers": "Aqa123456!",
+      "domain": "chala.wire.link",
+      "title": "chala.wire.link",
+      "users": [
+        {
+          "email": "[email protected]",
+          "id": "9f5a276d-f408-433e-81f6-812e69eb11bb"
+        },
+        {
+          "email": "[email protected]",
+          "id": "47fb193a-540f-4fec-969a-868377b02d8e"
+        },
+        {
+          "email": "[email protected]",
+          "id": "2342152d-3df8-4b50-81d8-950ca9105ab1"
+        }
+      ]
     }
   ]
 }