feat(core+frontend): enhance SyncEvent model and integrate sync handling in BewerbViewModel
- **Core Updates:** - Expanded `SyncEvent` model with additional fields (`eventId`, `sequenceNumber`, `originNodeId`, `createdAt`, `checksum`, `schemaVersion`) for improved event tracking and validation. - Updated event classes (`PingEvent`, `PongEvent`, `DataChangedEvent`, `DataRequestEvent`) to align with the extended `SyncEvent`. - **Frontend Enhancements:** - Enhanced `BewerbViewModel` to handle sync events (`PingEvent`, `DataChangedEvent`) and observe connected peers using `SyncManager`. - Added support for
This commit is contained in:
+36
-15
@@ -7,8 +7,12 @@ import kotlinx.serialization.Serializable
|
||||
*/
|
||||
@Serializable
|
||||
sealed interface SyncEvent {
|
||||
val timestamp: Long
|
||||
val senderId: String
|
||||
val eventId: String
|
||||
val sequenceNumber: Long
|
||||
val originNodeId: String
|
||||
val createdAt: Long
|
||||
val checksum: String
|
||||
val schemaVersion: Int
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -16,8 +20,12 @@ sealed interface SyncEvent {
|
||||
*/
|
||||
@Serializable
|
||||
data class PingEvent(
|
||||
override val timestamp: Long,
|
||||
override val senderId: String
|
||||
override val eventId: String,
|
||||
override val sequenceNumber: Long,
|
||||
override val originNodeId: String,
|
||||
override val createdAt: Long,
|
||||
override val checksum: String = "",
|
||||
override val schemaVersion: Int = 1
|
||||
) : SyncEvent
|
||||
|
||||
/**
|
||||
@@ -25,8 +33,12 @@ data class PingEvent(
|
||||
*/
|
||||
@Serializable
|
||||
data class PongEvent(
|
||||
override val timestamp: Long,
|
||||
override val senderId: String
|
||||
override val eventId: String,
|
||||
override val sequenceNumber: Long,
|
||||
override val originNodeId: String,
|
||||
override val createdAt: Long,
|
||||
override val checksum: String = "",
|
||||
override val schemaVersion: Int = 1
|
||||
) : SyncEvent
|
||||
|
||||
/**
|
||||
@@ -34,11 +46,16 @@ data class PongEvent(
|
||||
*/
|
||||
@Serializable
|
||||
data class DataChangedEvent(
|
||||
override val timestamp: Long,
|
||||
override val senderId: String,
|
||||
val entityType: String,
|
||||
val entityId: String,
|
||||
val operation: String // "CREATED", "UPDATED", "DELETED"
|
||||
override val eventId: String,
|
||||
override val sequenceNumber: Long,
|
||||
override val originNodeId: String,
|
||||
override val createdAt: Long,
|
||||
override val checksum: String = "",
|
||||
override val schemaVersion: Int = 1,
|
||||
val aggregateType: String,
|
||||
val aggregateId: String,
|
||||
val eventType: String, // "CREATED", "UPDATED", "DELETED"
|
||||
val payload: String // Base64 oder JSON String
|
||||
) : SyncEvent
|
||||
|
||||
/**
|
||||
@@ -46,8 +63,12 @@ data class DataChangedEvent(
|
||||
*/
|
||||
@Serializable
|
||||
data class DataRequestEvent(
|
||||
override val timestamp: Long,
|
||||
override val senderId: String,
|
||||
val entityType: String,
|
||||
val entityId: String
|
||||
override val eventId: String,
|
||||
override val sequenceNumber: Long,
|
||||
override val originNodeId: String,
|
||||
override val createdAt: Long,
|
||||
override val checksum: String = "",
|
||||
override val schemaVersion: Int = 1,
|
||||
val aggregateType: String,
|
||||
val aggregateId: String
|
||||
) : SyncEvent
|
||||
|
||||
+11
-3
@@ -28,9 +28,7 @@ class SyncManager(
|
||||
discovered.forEach { service ->
|
||||
val peerKey = "${service.host}:${service.port}"
|
||||
if (!knownPeers.contains(peerKey)) {
|
||||
// Prüfen, ob wir es nicht selbst sind (einfacher Check über Port,
|
||||
// in Realität über eine Node-ID im Metadata)
|
||||
// TODO: Node-ID Vergleich
|
||||
// TODO: Node-ID Vergleich (Selbst-Verbindung vermeiden)
|
||||
println("[SyncManager] Neuer Peer entdeckt: $peerKey. Verbinde...")
|
||||
syncService.connectToPeer(service.host, service.port)
|
||||
knownPeers.add(peerKey)
|
||||
@@ -41,6 +39,16 @@ class SyncManager(
|
||||
}
|
||||
}
|
||||
|
||||
fun getConnectedPeers() = syncService.connectedPeers
|
||||
|
||||
fun broadcastEvent(event: SyncEvent) {
|
||||
scope.launch {
|
||||
syncService.broadcastEvent(event)
|
||||
}
|
||||
}
|
||||
|
||||
fun getIncomingEvents() = syncService.incomingEvents
|
||||
|
||||
fun stop() {
|
||||
scope.cancel()
|
||||
discoveryService.stopDiscovery()
|
||||
|
||||
+32
-2
@@ -2,6 +2,7 @@ package at.mocode.turnier.feature.presentation
|
||||
|
||||
import at.mocode.frontend.core.network.discovery.DiscoveredService
|
||||
import at.mocode.frontend.core.network.sync.SyncManager
|
||||
import at.mocode.frontend.core.network.sync.*
|
||||
import at.mocode.turnier.feature.domain.Bewerb
|
||||
import at.mocode.turnier.feature.domain.BewerbRepository
|
||||
import at.mocode.turnier.feature.domain.StartlistenRepository
|
||||
@@ -93,8 +94,30 @@ class BewerbViewModel(
|
||||
|
||||
private fun observeSyncEvents() {
|
||||
syncManager?.let { manager ->
|
||||
// In einer realen App würde das P2pSyncService.incomingEvents Flow genutzt
|
||||
// Hier als Demo-Verknüpfung
|
||||
scope.launch {
|
||||
manager.getIncomingEvents().collect { event ->
|
||||
when (event) {
|
||||
is DataChangedEvent -> {
|
||||
if (event.aggregateType == "Bewerb" || event.aggregateType == "Startliste") {
|
||||
load() // Bei relevanten Änderungen neu laden
|
||||
}
|
||||
}
|
||||
is PingEvent -> {
|
||||
// Optional: Heartbeat loggen oder Status anzeigen
|
||||
}
|
||||
else -> {}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Auch verbundene Peers beobachten
|
||||
scope.launch {
|
||||
manager.getConnectedPeers().collect { peers ->
|
||||
reduce { it.copy(discoveredNodes = peers.map { p ->
|
||||
at.mocode.frontend.core.network.discovery.DiscoveredService("P2P", p, 0)
|
||||
}) }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -143,6 +166,13 @@ class BewerbViewModel(
|
||||
private fun startScan() {
|
||||
syncManager?.start(8080)
|
||||
_state.update { it.copy(isScanning = true) }
|
||||
// Nach dem Start des Servers ein Ping-Event broadcasten um Präsenz zu zeigen
|
||||
syncManager?.broadcastEvent(PingEvent(
|
||||
eventId = turnierId.toString(),
|
||||
sequenceNumber = 0,
|
||||
originNodeId = "Client-${(1000..9999).random()}",
|
||||
createdAt = 0 // In commonMain ohne Clock-Lib erst mal 0
|
||||
))
|
||||
refreshNodes()
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user