@@ -9,104 +9,85 @@ package com.datadog.android.rum.internal.timeseries
99import androidx.annotation.WorkerThread
1010import com.datadog.android.api.InternalLogger
1111import com.datadog.android.core.internal.utils.scheduleSafe
12+ import com.datadog.android.rum.internal.domain.RumContext
1213import com.datadog.android.rum.internal.domain.scope.RumViewType
1314import java.util.concurrent.ScheduledExecutorService
15+ import java.util.concurrent.ScheduledFuture
1416import java.util.concurrent.TimeUnit
15- import java.util.concurrent.atomic.AtomicInteger
16- import java.util.concurrent.atomic.AtomicReference
17-
18- /* *
19- * Per-session timeseries collector.
20- *
21- * Lifecycle invariants (enforced by [onSessionStart]/[onSessionStop] idempotency):
22- * - Each instance is single-use: exactly one [onSessionStart] followed by exactly one [onSessionStop].
23- * - Multiple [onSessionStart] / [onSessionStop] calls are safe — duplicate calls are no-ops.
24- * - After [onSessionStop], the instance must not be restarted; create a new instance.
25- *
26- * Background sampling:
27- * - When [collectInBackground] is `false`, [onViewTypeUpdate] pauses sampling on leaving foreground
28- * and resumes it on returning to foreground.
29- *
30- * Threading:
31- * - [onSessionStart] / [onSessionStop] / [onViewTypeUpdate] are called from the RUM event-handler thread.
32- * - Sampling tasks run on a [scheduledExecutorService] shared with other RUM components
33- * (owned by the SDK core); the instance neither owns nor shuts it down.
34- * - [Pipeline] is responsible for its own thread safety via internal synchronization.
35- * - Duplicate-chain prevention: each sampling chain carries a generation number.
36- * When [startSampling] starts a new generation, any in-flight or queued ticks from the
37- * previous generation self-terminate on their first check.
38- */
17+
3918internal class DefaultTimeseriesCollector (
4019 private val internalLogger : InternalLogger ,
4120 internal val pipelines : List <Pipeline <* >>,
42- private val collectInBackground : Boolean ,
43- internal val scheduledExecutorService : ScheduledExecutorService
21+ internal val scheduledExecutorService : ScheduledExecutorService ,
22+ @Volatile private var rumContext : RumContext
4423) : TimeseriesCollector {
24+ private val state = State ()
4525
46- private enum class State { IDLE , RUNNING , SUSPENDED , STOPPED }
47- private val state = AtomicReference ( State . IDLE )
48-
49- // Incremented on every start/resume. Ticks carrying a stale generation self-terminate.
50- private val currentGeneration = AtomicInteger ( 0 )
26+ // Batches are attributed to the view they are sent from, but a flush can happen once the app
27+ // already left the foreground, where there is no view to attribute to. Keep the last foreground
28+ // context so those batches stay attached to the view they were sent from.
29+ @Volatile
30+ private var lastForegroundRumContext : RumContext = rumContext
5131
5232 @Volatile
53- private var currentViewType : RumViewType ? = null
33+ private var recentSuspension : ScheduledFuture < * > ? = null
5434
35+ // region PUBLIC API
5536 @WorkerThread
5637 override fun onSessionStart () {
57- if (state.compareAndSet(State .IDLE , State .RUNNING )) {
58- startSampling()
59- }
38+ state.set(isActive = rumContext.viewType.isForeground)?.let { generation -> scheduleSampling(generation) }
6039 }
6140
6241 @WorkerThread
6342 override fun onSessionStop () {
64- if (state.getAndSet(State .STOPPED ) != State .STOPPED ) {
65- pipelines.forEach { pipeline ->
66- try {
67- synchronized(pipeline, pipeline::flush)
68- } catch (@Suppress(" TooGenericExceptionCaught" : Throwable ) {
69- internalLogger.log(
70- level = InternalLogger .Level .ERROR ,
71- targets = listOf (InternalLogger .Target .MAINTAINER , InternalLogger .Target .TELEMETRY ),
72- messageBuilder = { ERROR_FLUSH_FAILED },
73- throwable = t
74- )
75- }
76- }
43+ cancelSuspension()
44+ if (state.set(false ) != null ) {
45+ flushPipelines(lastForegroundRumContext)
7746 }
7847 }
7948
8049 @WorkerThread
81- override fun onViewTypeUpdate (newViewType : RumViewType ) {
82- if (newViewType == currentViewType) return
83- val isEnterForeground = ! currentViewType.isForeground && newViewType.isForeground
84- currentViewType = newViewType
85- if (! collectInBackground) {
86- if (isEnterForeground && state.compareAndSet(State .SUSPENDED , State .RUNNING )) {
87- startSampling()
88- } else if (! newViewType.isForeground) {
89- state.compareAndSet(State .RUNNING , State .SUSPENDED )
90- }
50+ override fun onRumContextUpdate (newRumContext : RumContext ) {
51+ val oldViewType = rumContext.viewType
52+ val newViewType = newRumContext.viewType
53+
54+ val isEnterForeground = ! oldViewType.isForeground && newViewType.isForeground
55+ val isLeaveForeground = oldViewType.isForeground && ! newViewType.isForeground
56+
57+ rumContext = newRumContext
58+ if (newViewType.isForeground) lastForegroundRumContext = newRumContext
59+
60+ if (isLeaveForeground) {
61+ scheduleStop(state.currentGeneration)
62+ } else if (isEnterForeground) {
63+ scheduleSampling(generation = state.startGeneration())
9164 }
9265 }
9366
94- private fun startSampling () {
95- val generation = currentGeneration.incrementAndGet()
96- pipelines.forEach { schedulePipeline(it, generation) }
67+ // endregion
68+
69+ // region SAMPLING
70+
71+ private fun scheduleSampling (generation : Int ) {
72+ cancelSuspension()
73+ pipelines.forEach { pipeline -> scheduledExecutorService.schedulePipeline(pipeline, generation) }
9774 }
9875
99- private fun schedulePipeline (pipeline : Pipeline <* >, generation : Int ) {
100- scheduledExecutorService.scheduleSafe(
101- OPERATION_NAME ,
76+ private fun ScheduledExecutorService.schedulePipeline (
77+ pipeline : Pipeline <* >,
78+ generation : Int
79+ ) {
80+ scheduleSafe(
81+ TIMESERIES_OPERATION_NAME ,
10282 pipeline.intervalMs,
10383 TimeUnit .MILLISECONDS ,
10484 internalLogger
10585 ) {
106- if (! isActive (generation)) return @scheduleSafe
86+ if (! state.isGenerationActive (generation)) return @scheduleSafe
10787 try {
108- synchronized(pipeline) {
109- if (isActive(generation)) pipeline.execute()
88+ val currentRumContext = rumContext
89+ if (currentRumContext.viewType.isForeground) {
90+ pipeline.execute(currentRumContext)
11091 }
11192 } catch (@Suppress(" TooGenericExceptionCaught" : Throwable ) {
11293 internalLogger.log(
@@ -116,24 +97,109 @@ internal class DefaultTimeseriesCollector(
11697 throwable = t
11798 )
11899 } finally {
119- if (isActive(generation)) schedulePipeline(pipeline, generation)
100+ if (state.isGenerationActive(generation)) schedulePipeline(pipeline, generation)
101+ }
102+ }
103+ }
104+
105+ @WorkerThread
106+ private fun flushPipelines (rumContext : RumContext ) {
107+ pipelines.forEach { pipeline ->
108+ try {
109+ pipeline.flush(rumContext)
110+ } catch (@Suppress(" TooGenericExceptionCaught" : Throwable ) {
111+ internalLogger.log(
112+ level = InternalLogger .Level .ERROR ,
113+ targets = listOf (InternalLogger .Target .MAINTAINER , InternalLogger .Target .TELEMETRY ),
114+ messageBuilder = { ERROR_FLUSH_FAILED },
115+ throwable = t
116+ )
120117 }
121118 }
122119 }
123120
124- private fun isActive (generation : Int ): Boolean =
125- state.get() == State .RUNNING && currentGeneration.get() == generation
121+ // endregion
122+
123+ // region SUSPENSION
124+
125+ private fun scheduleStop (stopRequestGeneration : Int ) {
126+ recentSuspension = scheduledExecutorService.scheduleSafe(
127+ SUSPEND_OPERATION_NAME ,
128+ SUSPEND_DELAY_MS ,
129+ TimeUnit .MILLISECONDS ,
130+ internalLogger
131+ ) {
132+ if (! rumContext.viewType.isForeground && state.stopGeneration(stopRequestGeneration)) {
133+ flushPipelines(lastForegroundRumContext)
134+ }
135+ }
136+ }
137+
138+ private fun cancelSuspension () {
139+ recentSuspension?.cancel(false )
140+ recentSuspension = null
141+ }
142+
143+ // endregion
126144
127145 internal companion object {
128- const val OPERATION_NAME = " Timeseries sampling"
146+ const val TIMESERIES_OPERATION_NAME = " Timeseries sampling"
147+ const val SUSPEND_OPERATION_NAME = " Timeseries suspend"
129148 const val ERROR_SAMPLING_FAILED = " Timeseries sampling iteration failed; rescheduling next sample."
130- const val ERROR_FLUSH_FAILED = " Timeseries flush on session stop failed."
149+ const val ERROR_FLUSH_FAILED = " Timeseries flush failed."
150+
151+ // Matches ActivityViewTrackingStrategy.STOP_VIEW_DELAY_MS, which guards the same race:
152+ // an Activity-to-Activity transition leaves no active view for a moment when the tracking
153+ // strategy stops the view on pause rather than on stop.
154+ const val SUSPEND_DELAY_MS = 200L
155+
131156 val RumViewType ?.isForeground: Boolean
132157 get() = when (this ) {
133158 RumViewType .FOREGROUND , RumViewType .APPLICATION_LAUNCH -> true
134159 RumViewType .BACKGROUND -> false
135160 RumViewType .NONE -> false
136161 null -> false
137162 }
163+
164+ private class State {
165+ // Written under the monitor only, but read outside of it, hence @Volatile: a stale
166+ // generation would make the guards below reject a live sampling chain or flush.
167+ @Volatile
168+ var currentGeneration: Int = 0
169+ private set
170+
171+ private var active: Boolean = false
172+
173+ fun isGenerationActive (generation : Int ): Boolean = synchronized(this ) {
174+ currentGeneration == generation && active
175+ }
176+
177+ /* *
178+ * Always starts a fresh generation, so that a suspension pending on the previous one
179+ * can no longer stop the sampling chain. Returns the new generation.
180+ */
181+ fun startGeneration (): Int = synchronized(this ) {
182+ active = true
183+ ++ currentGeneration
184+ }
185+
186+ /* *
187+ * Deactivates [generation] if it is still the current one.
188+ * Returns true when this call is the one that deactivated it.
189+ */
190+ fun stopGeneration (generation : Int ): Boolean = synchronized(this ) {
191+ currentGeneration == generation && set(false ) != null
192+ }
193+
194+ /* * Returns the new generation if this call changed the state, null otherwise. */
195+ fun set (isActive : Boolean ): Int? = synchronized(this ) {
196+ if (isActive == active) {
197+ null
198+ } else {
199+ active = isActive
200+ ++ currentGeneration
201+ }
202+ }
203+ }
138204 }
139205}
0 commit comments