Implement sync fully

libquassel-client/src/main/kotlin/de/justjanne/libquassel/client/session/BaseInitHandler.kt

+/*
+ * libquassel
+ * Copyright (c) 2021 Janne Mareike Koschinski
+ *
+ * This Source Code Form is subject to the terms of the Mozilla Public License,
+ * v. 2.0. If a copy of the MPL was not distributed with this file, You can
+ * obtain one at https://mozilla.org/MPL/2.0/.
+ */
+
+package de.justjanne.libquassel.client.session
+
+import de.justjanne.libquassel.client.util.CoroutineQueue
+import de.justjanne.libquassel.protocol.syncables.ObjectIdentifier
+import de.justjanne.libquassel.protocol.syncables.SyncableStub
+import de.justjanne.libquassel.protocol.util.update
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.MutableStateFlow
+
+class BaseInitHandler(
+  private val session: ClientSession
+) {
+  private val coroutineQueue = CoroutineQueue<Unit>()
+
+  fun sync(stub: SyncableStub) {
+    if (!stub.initialized) {
+      state.update {
+        copy(started = true, total = total + 1, waiting = waiting + ObjectIdentifier(stub))
+      }
+    }
+    session.proxy.synchronize(stub)
+  }
+
+  suspend fun initialized(identifier: ObjectIdentifier) {
+    state.update {
+      copy(waiting = waiting - identifier)
+    }
+    if (initDone()) {
+      coroutineQueue.resume(Unit)
+    }
+  }
+
+  fun initDone() = state().started && state().waiting.isEmpty()
+
+  suspend fun waitForInitDone() = if (!initDone()) {
+    coroutineQueue.wait()
+  } else Unit
+
+  @Suppress("NOTHING_TO_INLINE")
+  inline fun state(): BaseInitHandlerState = state.value
+
+  @Suppress("NOTHING_TO_INLINE")
+  inline fun flow(): Flow<BaseInitHandlerState> = state
+
+  @PublishedApi
+  internal val state = MutableStateFlow(BaseInitHandlerState())
+}

libquassel-client/src/main/kotlin/de/justjanne/libquassel/client/session/BaseInitHandlerState.kt

+/*
+ * libquassel
+ * Copyright (c) 2021 Janne Mareike Koschinski
+ *
+ * This Source Code Form is subject to the terms of the Mozilla Public License,
+ * v. 2.0. If a copy of the MPL was not distributed with this file, You can
+ * obtain one at https://mozilla.org/MPL/2.0/.
+ */
+
+package de.justjanne.libquassel.client.session
+
+import de.justjanne.libquassel.protocol.syncables.ObjectIdentifier
+
+data class BaseInitHandlerState(
+  val started: Boolean = false,
+  val total: Int = 0,
+  val waiting: Set<ObjectIdentifier> = setOf()
+)

libquassel-client/src/main/kotlin/de/justjanne/libquassel/client/session/ClientConnectionHandler.kt

@@ -24,6 +24,10 @@ abstract class ClientConnectionHandler : ConnectionHandler {
     return false
   }
 
+  override suspend fun done() {
+    this.channel = null
+  }
+
   suspend fun emit(message: SignalProxyMessage) {
     if (channel == null) {
       readyQueue.wait()

libquassel-client/src/main/kotlin/de/justjanne/libquassel/client/session/ClientHandshakeHandler.kt

@@ -20,13 +20,21 @@ import de.justjanne.libquassel.protocol.session.MessageChannelReadThread
 import de.justjanne.libquassel.protocol.session.Session
 import de.justjanne.libquassel.protocol.util.log.trace
 import de.justjanne.libquassel.protocol.variant.QVariantMap
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.runInterruptible
+import kotlinx.coroutines.withContext
 import org.slf4j.LoggerFactory
+import sun.security.util.DisabledAlgorithmConstraints
 import java.nio.ByteBuffer
 
 class ClientHandshakeHandler(
   val session: Session
 ) : HandshakeHandler, ClientConnectionHandler() {
   private val messageQueue = CoroutineKeyedQueue<Class<out HandshakeMessage>, HandshakeMessage>()
+  private var sessionInit: HandshakeMessage.SessionInit? = null
 
   override suspend fun read(buffer: ByteBuffer): Boolean {
     return dispatch(HandshakeMessageSerializer.deserialize(buffer, channel!!.negotiatedFeatures))
@@ -36,11 +44,18 @@ class ClientHandshakeHandler(
     logger.trace { "Read handshake message $message" }
     messageQueue.resume(message.javaClass, message)
     if (message is HandshakeMessage.SessionInit) {
-      session.init(message.identities, message.bufferInfos, message.networkIds)
+      sessionInit = message
       return true
-    } else {
+    }
     return false
   }
+
+  override suspend fun done() {
+    super.done()
+    val message = sessionInit
+    if (message != null) {
+      session.init(message.identities, message.bufferInfos, message.networkIds)
+    }
   }
 
   override suspend fun init(

libquassel-client/src/main/kotlin/de/justjanne/libquassel/client/session/ClientMagicHandler.kt

@@ -60,6 +60,8 @@ class ClientMagicHandler(
 
   override suspend fun read(buffer: ByteBuffer) = true
 
+  override suspend fun done() = Unit
+
   companion object {
     private val logger = LoggerFactory.getLogger(ClientMagicHandler::class.java)
   }

libquassel-client/src/main/kotlin/de/justjanne/libquassel/client/session/ClientProxyMessageHandler.kt

@@ -15,18 +15,19 @@ import de.justjanne.libquassel.protocol.models.SignalProxyMessage
 import de.justjanne.libquassel.protocol.serializers.SignalProxyMessageSerializer
 import de.justjanne.libquassel.protocol.session.ProxyMessageHandler
 import de.justjanne.libquassel.protocol.syncables.HeartBeatHandler
+import de.justjanne.libquassel.protocol.syncables.ObjectIdentifier
 import de.justjanne.libquassel.protocol.syncables.ObjectRepository
 import de.justjanne.libquassel.protocol.syncables.common.RpcHandler
 import de.justjanne.libquassel.protocol.syncables.invoker.Invokers
 import de.justjanne.libquassel.protocol.util.log.trace
 import org.slf4j.LoggerFactory
-import java.lang.Exception
 import java.nio.ByteBuffer
 
 class ClientProxyMessageHandler(
   private val heartBeatHandler: HeartBeatHandler,
   private val objectRepository: ObjectRepository,
-  private val rpcHandler: RpcHandler
+  private val rpcHandler: RpcHandler,
+  private val baseInitHandler: BaseInitHandler
 ) : ProxyMessageHandler, ClientConnectionHandler() {
 
   override suspend fun read(buffer: ByteBuffer): Boolean {
@@ -36,14 +37,16 @@ class ClientProxyMessageHandler(
 
   override suspend fun dispatch(message: SignalProxyMessage) {
     logger.trace { "Read signal proxy message $message" }
-    try {
     when (message) {
       is SignalProxyMessage.HeartBeat -> emit(SignalProxyMessage.HeartBeatReply(message.timestamp))
       is SignalProxyMessage.HeartBeatReply -> heartBeatHandler.recomputeLatency(message.timestamp, force = true)
-        is SignalProxyMessage.InitData -> objectRepository.init(
+      is SignalProxyMessage.InitData -> {
+        objectRepository.init(
           objectRepository.find(message.className, message.objectName) ?: return,
           message.initData
         )
+        baseInitHandler.initialized(ObjectIdentifier(message.className, message.objectName))
+      }
       is SignalProxyMessage.InitRequest -> {
         // Ignore incoming requests, we’re a client, we shouldn’t ever receive these
       }
@@ -60,9 +63,6 @@ class ClientProxyMessageHandler(
         invoker.invoke(syncable, message.slotName, message.params)
       }
     }
-    } catch (e: Exception) {
-      println(e)
-    }
   }
 
   companion object {

libquassel-client/src/main/kotlin/de/justjanne/libquassel/client/session/ClientRpcHandler.kt

@@ -14,10 +14,12 @@ import de.justjanne.libquassel.protocol.models.StatusMessage
 import de.justjanne.libquassel.protocol.serializers.qt.StringSerializerUtf8
 import de.justjanne.libquassel.protocol.session.Session
 import de.justjanne.libquassel.protocol.syncables.common.RpcHandler
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.runBlocking
 import java.nio.ByteBuffer
 
 class ClientRpcHandler(session: Session) : RpcHandler(session) {
@@ -30,11 +32,15 @@ class ClientRpcHandler(session: Session) : RpcHandler(session) {
   }
 
   override fun displayMsg(message: Message) {
-    messages.tryEmit(message)
+    runBlocking(Dispatchers.Default) {
+      messages.emit(message)
+    }
   }
 
   override fun displayStatusMsg(net: String?, msg: String?) {
-    statusMessage.tryEmit(StatusMessage(net, msg ?: return))
+    runBlocking(Dispatchers.Default) {
+      statusMessage.emit(StatusMessage(net, msg ?: ""))
+    }
   }
 
   @Suppress("NOTHING_TO_INLINE")

libquassel-client/src/main/kotlin/de/justjanne/libquassel/client/session/ClientSession.kt

@@ -12,6 +12,7 @@ import de.justjanne.libquassel.annotations.ProtocolSide
 import de.justjanne.libquassel.client.syncables.ClientBacklogManager
 import de.justjanne.libquassel.protocol.connection.ProtocolFeatures
 import de.justjanne.libquassel.protocol.connection.ProtocolMeta
+import de.justjanne.libquassel.protocol.features.QuasselFeature
 import de.justjanne.libquassel.protocol.io.CoroutineChannel
 import de.justjanne.libquassel.protocol.models.BufferInfo
 import de.justjanne.libquassel.protocol.models.ids.IdentityId
@@ -26,6 +27,7 @@ import de.justjanne.libquassel.protocol.syncables.ObjectRepository
 import de.justjanne.libquassel.protocol.syncables.common.AliasManager
 import de.justjanne.libquassel.protocol.syncables.common.BufferSyncer
 import de.justjanne.libquassel.protocol.syncables.common.BufferViewManager
+import de.justjanne.libquassel.protocol.syncables.common.CertManager
 import de.justjanne.libquassel.protocol.syncables.common.CoreInfo
 import de.justjanne.libquassel.protocol.syncables.common.DccConfig
 import de.justjanne.libquassel.protocol.syncables.common.HighlightRuleManager
@@ -34,6 +36,7 @@ import de.justjanne.libquassel.protocol.syncables.common.IgnoreListManager
 import de.justjanne.libquassel.protocol.syncables.common.IrcListHelper
 import de.justjanne.libquassel.protocol.syncables.common.Network
 import de.justjanne.libquassel.protocol.syncables.common.NetworkConfig
+import de.justjanne.libquassel.protocol.syncables.state.CertManagerState
 import de.justjanne.libquassel.protocol.syncables.state.NetworkState
 import de.justjanne.libquassel.protocol.util.log.info
 import de.justjanne.libquassel.protocol.util.update
@@ -51,21 +54,23 @@ class ClientSession(
 ) : Session {
   override val side = ProtocolSide.CLIENT
 
-  private val rpcHandler = ClientRpcHandler(this)
-  private val heartBeatHandler = HeartBeatHandler()
+  override val rpcHandler = ClientRpcHandler(this)
+  override val heartBeatHandler = HeartBeatHandler()
   override val objectRepository = ObjectRepository()
   val handshakeHandler = ClientHandshakeHandler(this)
+  val baseInitHandler = BaseInitHandler(this)
   private val proxyMessageHandler = ClientProxyMessageHandler(
     heartBeatHandler,
     objectRepository,
-    rpcHandler
+    rpcHandler,
+    baseInitHandler
   )
-  private val magicHandler = ClientMagicHandler(protocolFeatures, protocols, sslContext)
   override val proxy = CommonSyncProxy(
     ProtocolSide.CLIENT,
     objectRepository,
     proxyMessageHandler
   )
+  private val magicHandler = ClientMagicHandler(protocolFeatures, protocols, sslContext)
   private val messageChannel = MessageChannel(connection)
 
   init {
@@ -76,15 +81,53 @@ class ClientSession(
   }
 
   override fun init(
-    identities: List<QVariantMap>,
+    identityInfo: List<QVariantMap>,
     bufferInfos: List<BufferInfo>,
     networkIds: List<NetworkId>
   ) {
     logger.info {
-      "Client session initialized: networks = $networkIds, buffers = $bufferInfos, identities = $identities"
+      "Client session initialized: networks = $networkIds, buffers = $bufferInfos, identities = $identityInfo"
+    }
+
+    bufferSyncer.initializeBufferInfos(bufferInfos)
+
+    val networks = networkIds.map {
+      Network(this@ClientSession, NetworkState(networkId = it))
+    }
+    val identities = identityInfo.map {
+      Identity(this@ClientSession).apply {
+        fromVariantMap(it)
+      }
+    }
+
+    state.update {
+      copy(
+        networks = networks.associateBy(Network::networkId),
+        identities = identities.associateBy(Identity::id),
+      )
+    }
+
+    for (network in networks) {
+      baseInitHandler.sync(network)
+    }
+    for (identity in identities) {
+      baseInitHandler.sync(identity)
+      baseInitHandler.sync(CertManager(this, CertManagerState(identityId = identity.id())))
+    }
+    baseInitHandler.sync(state().aliasManager)
+    baseInitHandler.sync(state().bufferSyncer)
+    baseInitHandler.sync(state().bufferViewManager)
+    baseInitHandler.sync(state().coreInfo)
+    if (messageChannel.negotiatedFeatures.hasFeature(QuasselFeature.DccFileTransfer)) {
+      baseInitHandler.sync(state().dccConfig)
     }
-    objectRepository.add(state().coreInfo)
-    objectRepository.add(state().backlogManager)
+    baseInitHandler.sync(state().ignoreListManager)
+    if (messageChannel.negotiatedFeatures.hasFeature(QuasselFeature.CoreSideHighlights)) {
+      baseInitHandler.sync(state().highlightRuleManager)
+    }
+    baseInitHandler.sync(state().ircListHelper)
+    baseInitHandler.sync(state().networkConfig)
+    baseInitHandler.sync(state().backlogManager)
   }
 
   override fun network(id: NetworkId) = state().networks[id]
@@ -104,6 +147,8 @@ class ClientSession(
     }
   }
 
+  override fun networks() = state().networks.values.toSet()
+
   override fun identity(id: IdentityId) = state().identities[id]
 
   override fun addIdentity(properties: QVariantMap) {
@@ -123,6 +168,12 @@ class ClientSession(
     }
   }
 
+  override fun identities() = state().identities.values.toSet()
+
+  override fun certManager(id: IdentityId) = state().certManagers[id]
+
+  override fun certManagers() = state().certManagers.values.toSet()
+
   override fun rename(className: String, oldName: String, newName: String) {
     rpcHandler.objectRenamed(
       StringSerializerUtf8.serializeRaw(className),
@@ -162,6 +213,7 @@ class ClientSession(
     ClientSessionState(
       networks = mapOf(),
       identities = mapOf(),
+      certManagers = mapOf(),
       aliasManager = AliasManager(this),
       backlogManager = ClientBacklogManager(this),
       bufferSyncer = BufferSyncer(this),

libquassel-client/src/main/kotlin/de/justjanne/libquassel/client/session/ClientSessionState.kt

@@ -15,6 +15,7 @@ import de.justjanne.libquassel.protocol.models.ids.NetworkId
 import de.justjanne.libquassel.protocol.syncables.common.AliasManager
 import de.justjanne.libquassel.protocol.syncables.common.BufferSyncer
 import de.justjanne.libquassel.protocol.syncables.common.BufferViewManager
+import de.justjanne.libquassel.protocol.syncables.common.CertManager
 import de.justjanne.libquassel.protocol.syncables.common.CoreInfo
 import de.justjanne.libquassel.protocol.syncables.common.DccConfig
 import de.justjanne.libquassel.protocol.syncables.common.HighlightRuleManager
@@ -27,6 +28,7 @@ import de.justjanne.libquassel.protocol.syncables.common.NetworkConfig
 data class ClientSessionState(
   val networks: Map<NetworkId, Network>,
   val identities: Map<IdentityId, Identity>,
+  val certManagers: Map<IdentityId, CertManager>,
   val aliasManager: AliasManager,
   val backlogManager: ClientBacklogManager,
   val bufferSyncer: BufferSyncer,

libquassel-client/src/test/kotlin/de/justjanne/libquassel/client/ClientTest.kt

@@ -23,11 +23,11 @@ import de.justjanne.testcontainersci.api.providedContainer
 import de.justjanne.testcontainersci.extension.CiContainers
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.runBlocking
+import org.junit.jupiter.api.Assertions.assertTrue
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertThrows
 import java.net.InetSocketAddress
 import javax.net.ssl.SSLContext
-import kotlin.test.assertTrue
 
 @ExperimentalCoroutinesApi
 @CiContainers
@@ -36,17 +36,13 @@ class ClientTest {
     QuasselCoreContainer()
   }
 
-  private val sslContext = SSLContext.getInstance("TLSv1.3").apply {
-    init(null, arrayOf(TestX509TrustManager), null)
-  }
-
-  private val channel = CoroutineChannel()
-
-  private val username = "AzureDiamond"
-  private val password = "hunter2"
+  private val username = "kuschku"
+  private val password = "goalielecturetrawl"
 
   @Test
   fun testConnect(): Unit = runBlocking {
+    val channel = CoroutineChannel()
+
     channel.connect(
       InetSocketAddress(
         quassel.address,
@@ -62,7 +58,9 @@ class ClientTest {
           0x0000u
         )
       ),
-      sslContext
+      SSLContext.getInstance("TLSv1.3").apply {
+        init(null, arrayOf(TestX509TrustManager), null)
+      }
     )
     val coreState: CoreState = session.handshakeHandler.init(
       "Quasseltest v0.1",
@@ -92,6 +90,7 @@ class ClientTest {
       session.handshakeHandler.login("acidburn", "ineverweardresses")
     }
     session.handshakeHandler.login(username, password)
+    session.baseInitHandler.waitForInitDone()
     channel.close()
   }
 }

libquassel-client/src/test/resources/simplelogger.properties

@@ -7,4 +7,4 @@
 # obtain one at https://mozilla.org/MPL/2.0/.
 #
 
-org.slf4j.simpleLogger.defaultLogLevel=debug
+org.slf4j.simpleLogger.defaultLogLevel=trace

libquassel-protocol/src/main/kotlin/de/justjanne/libquassel/protocol/models/ConnectedClient.kt

@@ -19,6 +19,7 @@ import de.justjanne.libquassel.protocol.variant.QVariantMap
 import de.justjanne.libquassel.protocol.variant.into
 import de.justjanne.libquassel.protocol.variant.qVariant
 import org.threeten.bp.Instant
+import org.threeten.bp.ZoneOffset
 
 data class ConnectedClient(
   val id: Int,
@@ -51,7 +52,7 @@ data class ConnectedClient(
       versionDate = properties["clientVersionDate"].into("")
         .toLongOrNull()
         ?.let(Instant::ofEpochSecond),
-      connectedSince = properties["connectedSince"].into(Instant.EPOCH),
+      connectedSince = properties["connectedSince"].into(Instant.EPOCH.atOffset(ZoneOffset.UTC)).toInstant(),
       secure = properties["secure"].into(false),
       features = FeatureSet.build(
         LegacyFeature.of(properties["features"].into()),

libquassel-protocol/src/main/kotlin/de/justjanne/libquassel/protocol/serializers/qt/QDateTimeSerializer.kt

@@ -16,6 +16,7 @@ import de.justjanne.libquassel.protocol.serializers.PrimitiveSerializer
 import org.threeten.bp.Instant
 import org.threeten.bp.LocalDateTime
 import org.threeten.bp.OffsetDateTime
+import org.threeten.bp.ZoneId
 import org.threeten.bp.ZoneOffset
 import org.threeten.bp.ZonedDateTime
 import org.threeten.bp.temporal.Temporal
@@ -51,7 +52,7 @@ object QDateTimeSerializer : PrimitiveSerializer<Temporal> {
     }
   }
 
-  override fun deserialize(buffer: ByteBuffer, featureSet: FeatureSet): Temporal {
+  override fun deserialize(buffer: ByteBuffer, featureSet: FeatureSet): OffsetDateTime {
     val julianDay = QDateSerializer.deserialize(buffer, featureSet)
     val localTime = QTimeSerializer.deserialize(buffer, featureSet)
     val localDateTime = LocalDateTime.of(julianDay, localTime)
@@ -61,7 +62,7 @@ object QDateTimeSerializer : PrimitiveSerializer<Temporal> {
       TimeSpec.LocalStandard,
       TimeSpec.LocalUnknown,
       TimeSpec.LocalDST ->
-        localDateTime
+        localDateTime.atZone(ZoneId.systemDefault()).toOffsetDateTime()
       TimeSpec.OffsetFromUTC ->
         localDateTime
           .atOffset(
@@ -72,7 +73,6 @@ object QDateTimeSerializer : PrimitiveSerializer<Temporal> {
       TimeSpec.UTC ->
         localDateTime
           .atOffset(ZoneOffset.UTC)
-          .toInstant()
     }
   }
 }

libquassel-protocol/src/main/kotlin/de/justjanne/libquassel/protocol/serializers/qt/QTimeSerializer.kt

@@ -28,6 +28,10 @@ object QTimeSerializer : PrimitiveSerializer<LocalTime> {
 
   override fun deserialize(buffer: ByteBuffer, featureSet: FeatureSet): LocalTime {
     val millisecondOfDay = IntSerializer.deserialize(buffer, featureSet).toLong()
+    try {
       return LocalTime.ofNanoOfDay(millisecondOfDay * 1_000_000)
+    } catch (e: Exception) {
+      return LocalTime.MAX
+    }
   }
 }

libquassel-protocol/src/main/kotlin/de/justjanne/libquassel/protocol/serializers/signalproxy/HeartBeatSerializer.kt

@@ -16,6 +16,7 @@ import de.justjanne.libquassel.protocol.variant.QVariantList
 import de.justjanne.libquassel.protocol.variant.into
 import de.justjanne.libquassel.protocol.variant.qVariant
 import org.threeten.bp.Instant
+import org.threeten.bp.ZoneOffset
 
 /**
  * Serializer for [SignalProxyMessage.HeartBeat]
@@ -29,6 +30,6 @@ object HeartBeatSerializer : SignalProxySerializer<SignalProxyMessage.HeartBeat>
   )
 
   override fun deserialize(data: QVariantList) = SignalProxyMessage.HeartBeat(
-    data.getOrNull(1).into(Instant.EPOCH)
+    data.getOrNull(1).into(Instant.EPOCH.atOffset(ZoneOffset.UTC)).toInstant()
   )
 }

libquassel-protocol/src/main/kotlin/de/justjanne/libquassel/protocol/serializers/signalproxy/InitDataSerializer.kt

@@ -35,6 +35,6 @@ object InitDataSerializer : SignalProxySerializer<SignalProxyMessage.InitData> {
   override fun deserialize(data: QVariantList) = SignalProxyMessage.InitData(
     StringSerializerUtf8.deserializeRaw(data.getOrNull(1).into<ByteBuffer>()),
     StringSerializerUtf8.deserializeRaw(data.getOrNull(2).into<ByteBuffer>()),
-    data.drop(3).toVariantMap()
+    data.drop(3).toVariantMap(byteBuffer = true)
   )
 }

libquassel-protocol/src/main/kotlin/de/justjanne/libquassel/protocol/session/CommonSyncProxy.kt

@@ -14,6 +14,7 @@ import de.justjanne.libquassel.protocol.models.SignalProxyMessage
 import de.justjanne.libquassel.protocol.syncables.ObjectRepository
 import de.justjanne.libquassel.protocol.syncables.SyncableStub
 import de.justjanne.libquassel.protocol.variant.QVariantList
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.runBlocking
 
 class CommonSyncProxy(
@@ -22,9 +23,9 @@ class CommonSyncProxy(
   private val proxyMessageHandler: ProxyMessageHandler
 ) : SyncProxy {
   override fun synchronize(syncable: SyncableStub) {
-    if (objectRepository.add(syncable)) {
-      runBlocking {
-        proxyMessageHandler.dispatch(SignalProxyMessage.InitRequest(syncable.className, syncable.objectName))
+    if (objectRepository.add(syncable) && !syncable.initialized) {
+      runBlocking(context = Dispatchers.IO) {
+        proxyMessageHandler.emit(SignalProxyMessage.InitRequest(syncable.className, syncable.objectName))
       }
     }
   }

libquassel-protocol/src/main/kotlin/de/justjanne/libquassel/protocol/session/ConnectionHandler.kt

@@ -13,5 +13,6 @@ import java.nio.ByteBuffer
 
 interface ConnectionHandler {
   suspend fun init(channel: MessageChannel): Boolean
+  suspend fun done()
   suspend fun read(buffer: ByteBuffer): Boolean
 }

libquassel-protocol/src/main/kotlin/de/justjanne/libquassel/protocol/session/MessageChannel.kt

@@ -55,7 +55,8 @@ class MessageChannel(
     dispatch(messageBuffer)
   }
 
-  private suspend fun setupHandlers() {
+  private suspend fun setupHandlers(): List<ConnectionHandler> {
+    val removed = mutableListOf<ConnectionHandler>()
     while (true) {
       val handler = handlers.firstOrNull()
       logger.trace { "Setting up handler $handler" }
@@ -63,18 +64,27 @@ class MessageChannel(
         break
       }
       logger.trace { "Handler $handler is done" }
-      handlers.removeFirst()
+      removed.add(handlers.removeFirst())
     }
     if (handlers.isEmpty()) {
       logger.trace { "All handlers done" }
       channel.close()
     }
+    return removed
   }
 
   private suspend fun dispatch(message: ByteBuffer) {
-    if (handlers.first().read(message)) {
-      handlers.removeFirst()
-      setupHandlers()
+    val handlerDone = try {
+      handlers.first().read(message)
+    } catch (e: Exception) {
+      logger.warn("Error while handling message: ", e)
+      false
+    }
+    if (handlerDone) {
+      val removed = listOf(handlers.removeFirst()) + setupHandlers()
+      for (handler in removed) {
+        handler.done()
+      }
     }
   }
 

libquassel-protocol/src/main/kotlin/de/justjanne/libquassel/protocol/session/Session.kt

@@ -13,11 +13,13 @@ import de.justjanne.libquassel.annotations.ProtocolSide
 import de.justjanne.libquassel.protocol.models.BufferInfo
 import de.justjanne.libquassel.protocol.models.ids.IdentityId
 import de.justjanne.libquassel.protocol.models.ids.NetworkId
+import de.justjanne.libquassel.protocol.syncables.HeartBeatHandler
 import de.justjanne.libquassel.protocol.syncables.ObjectRepository
 import de.justjanne.libquassel.protocol.syncables.common.AliasManager
 import de.justjanne.libquassel.protocol.syncables.common.BacklogManager
 import de.justjanne.libquassel.protocol.syncables.common.BufferSyncer
 import de.justjanne.libquassel.protocol.syncables.common.BufferViewManager
+import de.justjanne.libquassel.protocol.syncables.common.CertManager
 import de.justjanne.libquassel.protocol.syncables.common.CoreInfo
 import de.justjanne.libquassel.protocol.syncables.common.DccConfig
 import de.justjanne.libquassel.protocol.syncables.common.HighlightRuleManager
@@ -26,6 +28,7 @@ import de.justjanne.libquassel.protocol.syncables.common.IgnoreListManager
 import de.justjanne.libquassel.protocol.syncables.common.IrcListHelper
 import de.justjanne.libquassel.protocol.syncables.common.Network
 import de.justjanne.libquassel.protocol.syncables.common.NetworkConfig
+import de.justjanne.libquassel.protocol.syncables.common.RpcHandler
 import de.justjanne.libquassel.protocol.variant.QVariantMap
 
 interface Session {
@@ -34,7 +37,7 @@ interface Session {
   val objectRepository: ObjectRepository
 
   fun init(
-    identities: List<QVariantMap>,
+    identityInfo: List<QVariantMap>,
     bufferInfos: List<BufferInfo>,
     networkIds: List<NetworkId>
   )
@@ -42,13 +45,21 @@ interface Session {
   fun network(id: NetworkId): Network?
   fun addNetwork(id: NetworkId)
   fun removeNetwork(id: NetworkId)
+  fun networks(): Set<Network>
 
   fun identity(id: IdentityId): Identity?
   fun addIdentity(properties: QVariantMap)
   fun removeIdentity(id: IdentityId)
+  fun identities(): Set<Identity>
+
+  fun certManager(id: IdentityId): CertManager?
+  fun certManagers(): Set<CertManager>
 
   fun rename(className: String, oldName: String, newName: String)
 
+  val heartBeatHandler: HeartBeatHandler
+  val rpcHandler: RpcHandler
+
   val aliasManager: AliasManager
   val backlogManager: BacklogManager
   val bufferSyncer: BufferSyncer