Copy
fun addAssetEvent(event: Event)
  ├─> assetHitProcessor.accumulateEvent(event)      // Add to memory
  ├─> persistEventImmediately(event, queue)         // Write to disk
  └─> checkAndFlushIfNeeded()                       // Check thresholds

suspend fun performFlush()
  ├─> val events = assetHitProcessor.processAccumulatedEvents()
  └─> [Orchestrator processes events → dispatches to Edge]
      └─> Edge guarantees delivery from here
Copied to your clipboard
fun addAssetEvent(event: Event)
  ├─> assetHitProcessor.accumulateEvent(event)      // Add to memory
  ├─> persistEventImmediately(event, queue)         // Write to disk
  └─> checkAndFlushIfNeeded()                       // Check thresholds

suspend fun performFlush()
  ├─> val events = assetHitProcessor.processAccumulatedEvents()
  └─> [Orchestrator processes events → dispatches to Edge]
      └─> Edge guarantees delivery from here

Copy
func addAssetEvent(_ event: Event)
  ├─> assetHitProcessor.accumulateEvent(event)      // Add to memory
  ├─> persistEventImmediately(event, to: queue)    // Write to disk
  └─> checkAndFlushIfNeeded()                       // Check thresholds

func performFlush()
  ├─> let events = assetHitProcessor.processAccumulatedEvents()
  └─> [Orchestrator processes events → dispatches to Edge]
      └─> Edge guarantees delivery from here
Copied to your clipboard
func addAssetEvent(_ event: Event)
  ├─> assetHitProcessor.accumulateEvent(event)      // Add to memory
  ├─> persistEventImmediately(event, to: queue)    // Write to disk
  └─> checkAndFlushIfNeeded()                       // Check thresholds

func performFlush()
  ├─> let events = assetHitProcessor.processAccumulatedEvents()
  └─> [Orchestrator processes events → dispatches to Edge]
      └─> Edge guarantees delivery from here

Copy
override suspend fun processHit(entity: DataEntity): Boolean
  ├─> Decode event from disk
  ├─> Accumulate in memory (if not already present)
  └─> return true → clear from disk (event now in memory)
Copied to your clipboard
override suspend fun processHit(entity: DataEntity): Boolean
  ├─> Decode event from disk
  ├─> Accumulate in memory (if not already present)
  └─> return true → clear from disk (event now in memory)

Copy
func processHit(entity: DataEntity, completion: (Bool) -> Void)
  ├─> Decode event from disk
  ├─> Accumulate in memory (if not already present)
  └─> completion(true) → clear from disk (event now in memory)
Copied to your clipboard
func processHit(entity: DataEntity, completion: (Bool) -> Void)
  ├─> Decode event from disk
  ├─> Accumulate in memory (if not already present)
  └─> completion(true) → clear from disk (event now in memory)

Copy
// On flush (ContentAnalyticsOrchestrator.kt)
private fun buildAssetMetricsCollection(events: List<Event>): AssetMetricsCollection {
    val groupedEvents = events.groupBy { it.assetKey ?: "" }
    val metricsMap = mutableMapOf<String, AssetMetrics>()
    
    for ((key, events) in groupedEvents) {
        val views = events.count { it.interactionType == InteractionType.VIEW }
        val clicks = events.count { it.interactionType == InteractionType.CLICK }
        metricsMap[key] = AssetMetrics(viewCount = views, clickCount = clicks, ...)
    }
    
    return AssetMetricsCollection(metricsMap)
}
Copied to your clipboard
// On flush (ContentAnalyticsOrchestrator.kt)
private fun buildAssetMetricsCollection(events: List<Event>): AssetMetricsCollection {
    val groupedEvents = events.groupBy { it.assetKey ?: "" }
    val metricsMap = mutableMapOf<String, AssetMetrics>()
    
    for ((key, events) in groupedEvents) {
        val views = events.count { it.interactionType == InteractionType.VIEW }
        val clicks = events.count { it.interactionType == InteractionType.CLICK }
        metricsMap[key] = AssetMetrics(viewCount = views, clickCount = clicks, ...)
    }
    
    return AssetMetricsCollection(metricsMap)
}

Copy
// On flush (ContentAnalyticsOrchestrator.swift)
func buildAssetMetricsCollection(from events: [Event]) -> AssetMetricsCollection {
    let groupedEvents = Dictionary(grouping: events) { $0.assetKey ?? "" }
    var metricsMap: [String: AssetMetrics] = [:]
    
    for (key, events) in groupedEvents {
        let views = events.filter { $0.interactionType == .view }.count
        let clicks = events.filter { $0.interactionType == .click }.count
        metricsMap[key] = AssetMetrics(viewCount: views, clickCount: clicks, ...)
    }
    
    return AssetMetricsCollection(metrics: metricsMap)
}
Copied to your clipboard
// On flush (ContentAnalyticsOrchestrator.swift)
func buildAssetMetricsCollection(from events: [Event]) -> AssetMetricsCollection {
    let groupedEvents = Dictionary(grouping: events) { $0.assetKey ?? "" }
    var metricsMap: [String: AssetMetrics] = [:]
    
    for (key, events) in groupedEvents {
        let views = events.filter { $0.interactionType == .view }.count
        let clicks = events.filter { $0.interactionType == .click }.count
        metricsMap[key] = AssetMetrics(viewCount: views, clickCount: clicks, ...)
    }
    
    return AssetMetricsCollection(metrics: metricsMap)
}

Copy
// BatchCoordinator
private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())
private val stateMutex = kotlinx.coroutines.sync.Mutex()

// DirectHitProcessor
private val mutex = Mutex()

// All state mutations wrapped in mutex.withLock { }
Copied to your clipboard
// BatchCoordinator
private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())
private val stateMutex = kotlinx.coroutines.sync.Mutex()

// DirectHitProcessor
private val mutex = Mutex()

// All state mutations wrapped in mutex.withLock { }

Last updated 3/3/2026

Was this helpful?

Yes

No