remove timer, run subscriptions again after insert on database

app/release/output-metadata.json

@@ -0,0 +1,97 @@
+{
+  "version": 3,
+  "artifactType": {
+    "type": "APK",
+    "kind": "Directory"
+  },
+  "applicationId": "com.greenart7c3.citrine",
+  "variantName": "release",
+  "elements": [
+    {
+      "type": "UNIVERSAL",
+      "filters": [],
+      "attributes": [],
+      "versionCode": 12,
+      "versionName": "0.1.1",
+      "outputFile": "app-universal-release.apk"
+    },
+    {
+      "type": "ONE_OF_MANY",
+      "filters": [
+        {
+          "filterType": "ABI",
+          "value": "arm64-v8a"
+        }
+      ],
+      "attributes": [],
+      "versionCode": 12,
+      "versionName": "0.1.1",
+      "outputFile": "app-arm64-v8a-release.apk"
+    },
+    {
+      "type": "ONE_OF_MANY",
+      "filters": [
+        {
+          "filterType": "ABI",
+          "value": "x86_64"
+        }
+      ],
+      "attributes": [],
+      "versionCode": 12,
+      "versionName": "0.1.1",
+      "outputFile": "app-x86_64-release.apk"
+    },
+    {
+      "type": "ONE_OF_MANY",
+      "filters": [
+        {
+          "filterType": "ABI",
+          "value": "armeabi-v7a"
+        }
+      ],
+      "attributes": [],
+      "versionCode": 12,
+      "versionName": "0.1.1",
+      "outputFile": "app-armeabi-v7a-release.apk"
+    },
+    {
+      "type": "ONE_OF_MANY",
+      "filters": [
+        {
+          "filterType": "ABI",
+          "value": "x86"
+        }
+      ],
+      "attributes": [],
+      "versionCode": 12,
+      "versionName": "0.1.1",
+      "outputFile": "app-x86-release.apk"
+    }
+  ],
+  "elementType": "File",
+  "baselineProfiles": [
+    {
+      "minApi": 28,
+      "maxApi": 30,
+      "baselineProfiles": [
+        "baselineProfiles/1/app-universal-release.dm",
+        "baselineProfiles/1/app-arm64-v8a-release.dm",
+        "baselineProfiles/1/app-x86_64-release.dm",
+        "baselineProfiles/1/app-armeabi-v7a-release.dm",
+        "baselineProfiles/1/app-x86-release.dm"
+      ]
+    },
+    {
+      "minApi": 31,
+      "maxApi": 2147483647,
+      "baselineProfiles": [
+        "baselineProfiles/0/app-universal-release.dm",
+        "baselineProfiles/0/app-arm64-v8a-release.dm",
+        "baselineProfiles/0/app-x86_64-release.dm",
+        "baselineProfiles/0/app-armeabi-v7a-release.dm",
+        "baselineProfiles/0/app-x86-release.dm"
+      ]
+    }
+  ],
+  "minSdkVersionForDexing": 26
+}

app/src/main/java/com/greenart7c3/citrine/database/EventDao.kt

@@ -5,6 +5,7 @@ import androidx.room.Insert
 import androidx.room.OnConflictStrategy
 import androidx.room.Query
 import androidx.room.Transaction
+import com.greenart7c3.citrine.server.EventSubscription
 
 @Dao
 interface EventDao {
@@ -79,6 +80,7 @@ interface EventDao {
             }
 
             insertTags(dbEvent.tags)
+            EventSubscription.executeAll()
         }
     }
 }

app/src/main/java/com/greenart7c3/citrine/server/EventRepository.kt

@@ -62,7 +62,7 @@ object EventRepository {
         val predicatesSql = whereClause.joinToString(" AND ", prefix = "WHERE ")
 
         var query = """
-            SELECT EventEntity.id
+            SELECT DISTINCT EventEntity.id
               FROM EventEntity EventEntity
               LEFT JOIN TagEntity TagEntity ON EventEntity.id = TagEntity.pkEvent
               $predicatesSql 
@@ -74,10 +74,6 @@ object EventRepository {
         }
 
         val cursor = subscription.appDatabase.query(query, arrayOf())
-        if (filter.kinds.contains(31234)) {
-            Log.d("draft", "${filter.lastExecuted}")
-            Log.d("draft", "${cursor.count}")
-        }
         if (cursor.count > 0) {
             filter.lastExecuted = TimeUtils.now()
         }
@@ -87,6 +83,7 @@ object EventRepository {
                 val eventEntity = subscription.appDatabase.eventDao().getById(item.getString(0))
                 eventEntity?.let {
                     val event = it.toEvent()
+                    Log.d("SEND", "sending event ${event.toJsonObject()}")
                     if (!event.isExpired()) {
                         runBlocking {
                             subscription.connection.session.send(

app/src/main/java/com/greenart7c3/citrine/server/EventSubscription.kt

@@ -4,6 +4,10 @@ import android.util.Log
 import android.util.LruCache
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.greenart7c3.citrine.database.AppDatabase
+import kotlinx.coroutines.DelicateCoroutinesApi
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.launch
 
 data class Subscription(
     val id: String,
@@ -16,6 +20,16 @@ data class Subscription(
 object EventSubscription {
     private val subscriptions = LruCache<String, SubscriptionManager>(30)
 
+    @OptIn(DelicateCoroutinesApi::class)
+    fun executeAll() {
+        Log.d("executeAll", "executeAll")
+        GlobalScope.launch(Dispatchers.IO) {
+            subscriptions.snapshot().values.forEach {
+                it.execute()
+            }
+        }
+    }
+
     fun closeAll(connectionName: String) {
         Log.d("connection", "finalizing subscriptions from $connectionName")
         subscriptions.snapshot().keys.forEach {
@@ -33,7 +47,6 @@ object EventSubscription {
     }
 
     fun close(subscriptionId: String) {
-        subscriptions[subscriptionId]?.finalize()
         subscriptions.remove(subscriptionId)
         Log.d("subscriptions", subscriptions.size().toString())
     }

app/src/main/java/com/greenart7c3/citrine/server/SubscriptionManager.kt

@@ -6,58 +6,44 @@ import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.DelicateCoroutinesApi
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.GlobalScope
-import kotlinx.coroutines.Job
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
-import java.util.Timer
-import kotlin.concurrent.schedule
 
 @OptIn(DelicateCoroutinesApi::class)
 class SubscriptionManager(val subscription: Subscription) {
-    private val timer: Timer = Timer()
-    private var currentJob: Job? = null
-
-    fun finalize() {
-        timer.cancel()
-        Log.d("timer", "finalize id: ${subscription.id}")
-    }
-
-    init {
-        timer.schedule(
-            0,
-            15000
-        ) {
-            Log.d("timer", "executed timer id: ${subscription.id}")
-            Log.d("timer", "isClosedForSend: ${subscription.connection.session.outgoing.isClosedForSend}")
-            currentJob?.cancel()
-
-            if (subscription.connection.session.outgoing.isClosedForSend) {
-                EventSubscription.close(subscription.id)
-                Log.d("timer", "cancelling subscription isClosedForSend: ${subscription.id}")
-                return@schedule
-            }
-
-            currentJob = GlobalScope.launch(Dispatchers.IO) {
-                for (filter in subscription.filters) {
-                    try {
-                        runBlocking {
-                            EventRepository.subscribe(
-                                subscription,
-                                filter
-                            )
-                        }
-                    } catch (e: Exception) {
-                        if (e is CancellationException) throw e
+    suspend fun execute() {
+        if (subscription.connection.session.outgoing.isClosedForSend) {
+            EventSubscription.close(subscription.id)
+            Log.d("timer", "cancelling subscription isClosedForSend: ${subscription.id}")
+            return
+        }
 
-                        Log.d("error", "Error reading data from database", e)
-                        subscription.connection.session.send(
-                            NoticeResult.invalid("Error reading data from database").toJson()
+        val currentJob = GlobalScope.launch(Dispatchers.IO) {
+            for (filter in subscription.filters) {
+                try {
+                    runBlocking {
+                        EventRepository.subscribe(
+                            subscription,
+                            filter
                         )
                     }
+                } catch (e: Exception) {
+                    if (e is CancellationException) throw e
+
+                    Log.d("error", "Error reading data from database", e)
+                    subscription.connection.session.send(
+                        NoticeResult.invalid("Error reading data from database").toJson()
+                    )
                 }
-                subscription.connection.session.send(EOSE(subscription.id).toJson())
             }
-            runBlocking { currentJob?.join() }
+        }
+        runBlocking { currentJob.join() }
+    }
+
+    init {
+        runBlocking {
+            execute()
+            subscription.connection.session.send(EOSE(subscription.id).toJson())
         }
     }
 }