Merge pull request #89 from Parsely/coroutines

Migrate threading model to Coroutines - long running branch

Author: wzieba

Diff for: .github/workflows/readme.yml

@@ -72,4 +72,6 @@ jobs:
         if: always()
         with:
           name: artifact
-          path: ./parsely/build/reports/*
+          path: |
+            ./parsely/build/reports/*
+            ./parsely/build/outputs/androidTest-results

Diff for: example/src/main/java/com/example/MainActivity.java

@@ -30,33 +30,18 @@ protected void onCreate(Bundle savedInstanceState) {
         // Set debugging to true so we don't actually send things to Parse.ly
         ParselyTracker.sharedInstance().setDebug(true);
 
-        final TextView queueView = (TextView) findViewById(R.id.queue_size);
-        queueView.setText(String.format("Queued events: %d", ParselyTracker.sharedInstance().queueSize()));
-
-        final TextView storedView = (TextView) findViewById(R.id.stored_size);
-        storedView.setText(String.format("Stored events: %d", ParselyTracker.sharedInstance().storedEventsCount()));
-
         final TextView intervalView = (TextView) findViewById(R.id.interval);
-        storedView.setText(String.format("Flush interval: %d", ParselyTracker.sharedInstance().getFlushInterval()));
 
         updateEngagementStrings();
 
-        final TextView views[] = new TextView[3];
-        views[0] = queueView;
-        views[1] = storedView;
-        views[2] = intervalView;
+        final TextView views[] = new TextView[1];
+        views[0] = intervalView;
 
         final Handler mHandler = new Handler() {
             @Override
             public void handleMessage(Message msg) {
                 TextView[] v = (TextView[]) msg.obj;
-                TextView qView = v[0];
-                qView.setText(String.format("Queued events: %d", ParselyTracker.sharedInstance().queueSize()));
-
-                TextView sView = v[1];
-                sView.setText(String.format("Stored events: %d", ParselyTracker.sharedInstance().storedEventsCount()));
-
-                TextView iView = v[2];
+                TextView iView = v[0];
                 if (ParselyTracker.sharedInstance().flushTimerIsActive()) {
                     iView.setText(String.format("Flush Interval: %d", ParselyTracker.sharedInstance().getFlushInterval()));
                 } else {

Diff for: example/src/main/res/layout/activity_main.xml

@@ -68,26 +68,11 @@
         android:onClick="trackReset"
         android:text="@string/button_reset_video" />
 
-    <TextView
-        android:id="@+id/queue_size"
-        android:layout_width="wrap_content"
-        android:layout_height="wrap_content"
-        android:layout_below="@+id/reset_video_button"
-        android:layout_centerHorizontal="true"
-        android:text="Queued events: 0" />
-
-    <TextView android:id="@+id/stored_size"
-        android:layout_width="wrap_content"
-        android:layout_height="wrap_content"
-        android:layout_centerHorizontal="true"
-        android:layout_below="@id/queue_size" 
-        android:text="Stored events: 0"/>
-        
     <TextView android:id="@+id/interval"
         android:layout_width="wrap_content"
         android:layout_height="wrap_content"
         android:layout_centerHorizontal="true"
-        android:layout_below="@id/stored_size" 
+        android:layout_below="@id/reset_video_button"
         android:text="Flush timer inactive"/>
 
     <TextView
@@ -107,4 +92,4 @@
         android:text="Video is inactive." />
 
 
-</RelativeLayout>
+</RelativeLayout>

Diff for: parsely/build.gradle

@@ -6,6 +6,7 @@ plugins {
 
 ext {
     assertJVersion = '3.24.2'
+    coroutinesVersion = '1.7.3'
     mockWebServerVersion = '4.12.0'
 }
 
@@ -63,12 +64,14 @@ dependencies {
     implementation 'com.fasterxml.jackson.core:jackson-databind:2.13.3'
     implementation 'com.google.android.gms:play-services-ads-identifier:18.0.1'
     implementation 'androidx.lifecycle:lifecycle-process:2.6.2'
+    implementation "org.jetbrains.kotlinx:kotlinx-coroutines-android:$coroutinesVersion"
 
     testImplementation 'org.robolectric:robolectric:4.10.3'
     testImplementation 'androidx.test:core:1.5.0'
     testImplementation "org.assertj:assertj-core:$assertJVersion"
     testImplementation 'junit:junit:4.13.2'
     testImplementation "com.squareup.okhttp3:mockwebserver:$mockWebServerVersion"
+    testImplementation "org.jetbrains.kotlinx:kotlinx-coroutines-test:$coroutinesVersion"
 
     androidTestImplementation 'androidx.test.ext:junit:1.1.5'
     androidTestImplementation 'androidx.test:rules:1.5.0'

Diff for: parsely/src/androidTest/java/com/parsely/parselyandroid/FunctionalTests.kt

@@ -28,6 +28,7 @@ import okhttp3.mockwebserver.MockResponse
 import okhttp3.mockwebserver.MockWebServer
 import okhttp3.mockwebserver.RecordedRequest
 import org.assertj.core.api.Assertions.assertThat
+import org.assertj.core.api.Assertions.within
 import org.junit.Test
 import org.junit.runner.RunWith
 
@@ -49,12 +50,12 @@ class FunctionalTests {
     }
 
     /**
-     * In this scenario, the consumer application tracks more than 50 events-threshold during a flush interval.
+     * In this scenario, the consumer application tracks 51 events-threshold during a flush interval.
      * The SDK will save the events to disk and send them in the next flush interval.
      * At the end, when all events are sent, the SDK will delete the content of local storage file.
      */
     @Test
-    fun appTracksEventsAboveQueueSizeLimit() {
+    fun appTracksEventsDuringTheFlushInterval() {
         ActivityScenario.launch(SampleActivity::class.java).use { scenario ->
             scenario.onActivity { activity: Activity ->
                 beforeEach(activity)
@@ -76,6 +77,47 @@ class FunctionalTests {
         }
     }
 
+    /**
+     * In this scenario, the consumer app tracks 2 events during the first flush interval.
+     * Then, we validate, that after flush interval passed the SDK sends the events
+     * to Parse.ly servers.
+     *
+     * Then, the consumer app tracks another event and we validate that the SDK sends the event
+     * to Parse.ly servers as well.
+     */
+    @Test
+    fun appFlushesEventsAfterFlushInterval() {
+        ActivityScenario.launch(SampleActivity::class.java).use { scenario ->
+            scenario.onActivity { activity: Activity ->
+                beforeEach(activity)
+                server.enqueue(MockResponse().setResponseCode(200))
+                parselyTracker = initializeTracker(activity)
+
+                parselyTracker.trackPageview("url", null, null, null)
+            }
+
+            Thread.sleep((defaultFlushInterval / 2).inWholeMilliseconds)
+
+            scenario.onActivity {
+                parselyTracker.trackPageview("url", null, null, null)
+            }
+
+            Thread.sleep((defaultFlushInterval / 2).inWholeMilliseconds)
+
+            val firstRequestPayload = server.takeRequest(2000, TimeUnit.MILLISECONDS)?.toMap()
+            assertThat(firstRequestPayload!!["events"]).hasSize(2)
+
+            scenario.onActivity {
+                parselyTracker.trackPageview("url", null, null, null)
+            }
+
+            Thread.sleep(defaultFlushInterval.inWholeMilliseconds)
+
+            val secondRequestPayload = server.takeRequest(2000, TimeUnit.MILLISECONDS)?.toMap()
+            assertThat(secondRequestPayload!!["events"]).hasSize(1)
+        }
+    }
+
     /**
      * In this scenario, the consumer application:
      * 1. Goes to the background
@@ -113,6 +155,132 @@ class FunctionalTests {
         }
     }
 
+    /**
+     * In this scenario we "stress test" the concurrency model to see if we have any conflict during
+     *
+     * - Unexpectedly high number of recorded events in small intervals (I/O locking)
+     * - Scenario in which a request is sent at the same time as new events are recorded
+     */
+    @Test
+    fun stressTest() {
+        val eventsToSend = 500
+
+        ActivityScenario.launch(SampleActivity::class.java).use { scenario ->
+            scenario.onActivity { activity: Activity ->
+                beforeEach(activity)
+                server.enqueue(MockResponse().setResponseCode(200))
+                parselyTracker = initializeTracker(activity)
+
+                repeat(eventsToSend) {
+                    parselyTracker.trackPageview("url", null, null, null)
+                }
+            }
+
+            // Wait some time to give events chance to be saved in local data storage
+            Thread.sleep((defaultFlushInterval * 2).inWholeMilliseconds)
+
+            // Catch up to 10 requests. We don't know how many requests the device we test on will
+            // perform. It's probably more like 1-2, but we're on safe (not flaky) side here.
+            val requests = (1..10).mapNotNull {
+                runCatching { server.takeRequest(100, TimeUnit.MILLISECONDS) }.getOrNull()
+            }.flatMap {
+                it.toMap()["events"]!!
+            }
+
+            assertThat(requests).hasSize(eventsToSend)
+        }
+    }
+
+    /**
+     * In this scenario consumer app starts an engagement session and after 27150 ms,
+     * it stops the session.
+     *
+     * Intervals:
+     * With current implementation of `HeartbeatIntervalCalculator`, the next intervals are:
+     * - 10500ms for the first interval
+     * - 13650ms for the second interval
+     *
+     * So after ~27,2s we should observe
+     *  - 2 `heartbeat` events from `startEngagement` + 1 `heartbeat` event caused by `stopEngagement` which is triggered during engagement interval
+     *
+     * Time off-differences in assertions are acceptable, because it's a time-sensitive test
+     */
+    @Test
+    fun engagementManagerTest() {
+        val engagementUrl = "engagementUrl"
+        var startTimestamp = Duration.ZERO
+        val firstInterval = 10500.milliseconds
+        val secondInterval = 13650.milliseconds
+        val pauseInterval = 3.seconds
+        ActivityScenario.launch(SampleActivity::class.java).use { scenario ->
+            // given
+            scenario.onActivity { activity: Activity ->
+                beforeEach(activity)
+                server.enqueue(MockResponse().setResponseCode(200))
+                parselyTracker = initializeTracker(activity, flushInterval = 30.seconds)
+
+                // when
+                startTimestamp = System.currentTimeMillis().milliseconds
+                parselyTracker.startEngagement(engagementUrl, null)
+            }
+
+            Thread.sleep((firstInterval + secondInterval + pauseInterval).inWholeMilliseconds)
+            parselyTracker.stopEngagement()
+
+            // then
+            val request = server.takeRequest(35, TimeUnit.SECONDS)!!.toMap()["events"]!!
+
+            assertThat(
+                request.sortedBy { it.data.timestamp }
+                    .filter { it.action == "heartbeat" }
+            ).hasSize(3)
+                .satisfies({
+                    val firstEvent = it[0]
+                    val secondEvent = it[1]
+                    val thirdEvent = it[2]
+
+                    assertThat(firstEvent.data.timestamp).isCloseTo(
+                        (startTimestamp + firstInterval).inWholeMilliseconds,
+                        within(1.seconds.inWholeMilliseconds)
+                    )
+                    assertThat(firstEvent.totalTime).isCloseTo(
+                        firstInterval.inWholeMilliseconds,
+                        within(100L)
+                    )
+                    assertThat(firstEvent.incremental).isCloseTo(
+                        firstInterval.inWholeSeconds,
+                        within(1L)
+                    )
+
+                    assertThat(secondEvent.data.timestamp).isCloseTo(
+                        (startTimestamp + firstInterval + secondInterval).inWholeMilliseconds,
+                        within(1.seconds.inWholeMilliseconds)
+                    )
+                    assertThat(secondEvent.totalTime).isCloseTo(
+                        (firstInterval + secondInterval).inWholeMilliseconds,
+                        within(100L)
+                    )
+                    assertThat(secondEvent.incremental).isCloseTo(
+                        secondInterval.inWholeSeconds,
+                        within(1L)
+                    )
+
+                    assertThat(thirdEvent.data.timestamp).isCloseTo(
+                        (startTimestamp + firstInterval + secondInterval + pauseInterval).inWholeMilliseconds,
+                        within(1.seconds.inWholeMilliseconds)
+                    )
+                    assertThat(thirdEvent.totalTime).isCloseTo(
+                        (firstInterval + secondInterval + pauseInterval).inWholeMilliseconds,
+                        within(100L)
+                    )
+                    assertThat(thirdEvent.incremental).isCloseTo(
+                        (pauseInterval).inWholeSeconds,
+                        within(1L)
+                    )
+                })
+        }
+    }
+
     private fun RecordedRequest.toMap(): Map<String, List<Event>> {
         val listType: TypeReference<Map<String, List<Event>>> =
             object : TypeReference<Map<String, List<Event>>>() {}
@@ -123,6 +291,15 @@ class FunctionalTests {
     @JsonIgnoreProperties(ignoreUnknown = true)
     data class Event(
         @JsonProperty("idsite") var idsite: String,
+        @JsonProperty("action") var action: String,
+        @JsonProperty("data") var data: ExtraData,
+        @JsonProperty("tt") var totalTime: Long,
+        @JsonProperty("inc") var incremental: Long,
+    )
+
+    @JsonIgnoreProperties(ignoreUnknown = true)
+    data class ExtraData(
+        @JsonProperty("ts") var timestamp: Long,
     )
 
     private val locallyStoredEvents
@@ -137,19 +314,18 @@ class FunctionalTests {
         activity: Activity,
         flushInterval: Duration = defaultFlushInterval
     ): ParselyTracker {
+        val field: Field = ParselyTracker::class.java.getDeclaredField("ROOT_URL")
+        field.isAccessible = true
+        field.set(this, url)
         return ParselyTracker.sharedInstance(
             siteId, flushInterval.inWholeSeconds.toInt(), activity.application
-        ).apply {
-            val f: Field = this::class.java.getDeclaredField("ROOT_URL")
-            f.isAccessible = true
-            f.set(this, url)
-        }
+        )
     }
 
     private companion object {
         const val siteId = "123"
         const val localStorageFileName = "parsely-events.ser"
-        val defaultFlushInterval = 10.seconds
+        val defaultFlushInterval = 5.seconds
     }
 
     class SampleActivity : Activity()

Diff for: parsely/src/main/java/com/parsely/parselyandroid/AdvertisementIdProvider.kt

@@ -0,0 +1,50 @@
+package com.parsely.parselyandroid
+
+import android.content.Context
+import android.provider.Settings
+import com.google.android.gms.ads.identifier.AdvertisingIdClient
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.launch
+
+internal class AdvertisementIdProvider(
+    private val context: Context,
+    coroutineScope: CoroutineScope
+) : IdProvider {
+
+    private var adKey: String? = null
+
+    init {
+        coroutineScope.launch {
+            try {
+                adKey = AdvertisingIdClient.getAdvertisingIdInfo(context).id
+            } catch (e: Exception) {
+                ParselyTracker.PLog("No Google play services or error!")
+            }
+        }
+    }
+
+    /**
+     * @return advertisement id if the coroutine in the constructor finished executing AdvertisingIdClient#getAdvertisingIdInfo
+     * null otherwise
+     */
+    override fun provide(): String? = adKey
+}
+
+internal class AndroidIdProvider(private val context: Context) : IdProvider {
+    override fun provide(): String? {
+        val uuid = try {
+            Settings.Secure.getString(
+                context.applicationContext.contentResolver,
+                Settings.Secure.ANDROID_ID
+            )
+        } catch (ex: Exception) {
+            null
+        }
+        ParselyTracker.PLog(String.format("Android ID: %s", uuid))
+        return uuid
+    }
+}
+
+internal fun interface IdProvider {
+    fun provide(): String?
+}