remove connection listener, move repo into connectionmgr

app/src/main/java/io/heckel/ntfy/MainActivity.kt

@@ -18,13 +18,13 @@ import io.heckel.ntfy.data.*
 import io.heckel.ntfy.detail.DetailActivity
 import kotlin.random.Random
 
-const val TOPIC_ID = "topic_id"
+const val SUBSCRIPTION_ID = "topic_id"
 const val TOPIC_NAME = "topic_name"
-const val TOPIC_BASE_URL = "base_url"
+const val SERVICE_BASE_URL = "base_url"
 
 class MainActivity : AppCompatActivity() {
-    private val newTopicActivityRequestCode = 1
+    private val newSubscriptionActivityRequestCode = 1
-    private val topicsViewModel by viewModels<SubscriptionViewModel> {
+    private val subscriptionViewModel by viewModels<SubscriptionsViewModel> {
         SubscriptionsViewModelFactory()
     }
 
@@ -39,11 +39,11 @@ class MainActivity : AppCompatActivity() {
         }
 
         // Update main list based on topicsViewModel (& its datasource/livedata)
-        val adapter = TopicsAdapter { topic -> topicOnClick(topic) }
+        val adapter = TopicsAdapter { topic -> subscriptionOnClick(topic) }
         val recyclerView: RecyclerView = findViewById(R.id.recycler_view)
         recyclerView.adapter = adapter
 
-        topicsViewModel.list().observe(this) {
+        subscriptionViewModel.list().observe(this) {
             it?.let {
                 println("new data arrived: $it")
                 adapter.submitList(it as MutableList<Subscription>)
@@ -52,36 +52,31 @@ class MainActivity : AppCompatActivity() {
 
         // Set up notification channel
         createNotificationChannel()
-        topicsViewModel.setListener(object : NotificationListener {
+        subscriptionViewModel.setListener { n -> displayNotification(n) }
-            override fun onNotification(subscriptionId: Long, notification: Notification) {
-                displayNotification(notification)
-            }
-        })
     }
 
-    /* Opens TopicDetailActivity when RecyclerView item is clicked. */
+    /* Opens detail view when list item is clicked. */
-    private fun topicOnClick(topic: Subscription) {
+    private fun subscriptionOnClick(subscription: Subscription) {
         val intent = Intent(this, DetailActivity()::class.java)
-        intent.putExtra(TOPIC_ID, topic.id)
+        intent.putExtra(SUBSCRIPTION_ID, subscription.id)
         startActivity(intent)
     }
 
     /* Adds topic to topicList when FAB is clicked. */
     private fun fabOnClick() {
         val intent = Intent(this, AddTopicActivity::class.java)
-        startActivityForResult(intent, newTopicActivityRequestCode)
+        startActivityForResult(intent, newSubscriptionActivityRequestCode)
     }
 
     override fun onActivityResult(requestCode: Int, resultCode: Int, intentData: Intent?) {
         super.onActivityResult(requestCode, resultCode, intentData)
 
-        if (requestCode == newTopicActivityRequestCode && resultCode == Activity.RESULT_OK) {
+        if (requestCode == newSubscriptionActivityRequestCode && resultCode == Activity.RESULT_OK) {
             intentData?.let { data ->
                 val name = data.getStringExtra(TOPIC_NAME) ?: return
-                val baseUrl = data.getStringExtra(TOPIC_BASE_URL) ?: return
+                val baseUrl = data.getStringExtra(SERVICE_BASE_URL) ?: return
-                val topic = Subscription(Random.nextLong(), name, baseUrl, Status.CONNECTING, 0)
+                val subscription = Subscription(Random.nextLong(), name, baseUrl, Status.CONNECTING, 0)
-
+                subscriptionViewModel.add(subscription)
-                topicsViewModel.add(topic)
             }
         }
     }

app/src/main/java/io/heckel/ntfy/SubscriptionViewModel.kt

@@ -1,67 +0,0 @@
-package io.heckel.ntfy
-
-import androidx.lifecycle.LiveData
-import androidx.lifecycle.ViewModel
-import androidx.lifecycle.ViewModelProvider
-import androidx.lifecycle.viewModelScope
-import io.heckel.ntfy.data.*
-import kotlin.collections.List
-
-
-class SubscriptionViewModel(private val repository: Repository, private val connectionManager: ConnectionManager) : ViewModel() {
-    fun add(topic: Subscription) {
-        repository.add(topic)
-        connectionManager.start(topic, viewModelScope)
-    }
-
-    fun get(id: Long) : Subscription? {
-        return repository.get(id)
-    }
-
-    fun list(): LiveData<List<Subscription>> {
-        return repository.list()
-    }
-
-    fun remove(topic: Subscription) {
-        repository.remove(topic)
-        connectionManager.stop(topic)
-    }
-
-    fun setListener(listener: NotificationListener) {
-        connectionManager.setListener(object : ConnectionListener {
-            override fun onStatusChanged(subcriptionId: Long, status: Status) {
-                println("onStatusChanged($subcriptionId, $status)")
-                val topic = repository.get(subcriptionId)
-                if (topic != null) {
-                    println("-> old topic: $topic")
-                    repository.update(topic.copy(status = status))
-                }
-            }
-
-            override fun onNotification(subscriptionId: Long, notification: Notification) {
-                println("onNotification($subscriptionId, $notification)")
-                val topic = repository.get(subscriptionId)
-                if (topic != null) {
-                    println("-> old topic: $topic")
-                    repository.update(topic.copy(messages = topic.messages + 1))
-                }
-                listener.onNotification(subscriptionId, notification) // Forward downstream
-            }
-        })
-    }
-}
-
-class SubscriptionsViewModelFactory : ViewModelProvider.Factory {
-    @Suppress("UNCHECKED_CAST")
-    override fun <T : ViewModel?> create(modelClass: Class<T>) =
-        with(modelClass){
-            when {
-                isAssignableFrom(SubscriptionViewModel::class.java) -> {
-                    val repository = Repository.getInstance()
-                    val connectionManager = ConnectionManager.getInstance()
-                    SubscriptionViewModel(repository, connectionManager) as T
-                }
-                else -> throw IllegalArgumentException("Unknown viewModel class $modelClass")
-            }
-        }
-}

app/src/main/java/io/heckel/ntfy/SubscriptionsViewModel.kt

@@ -0,0 +1,46 @@
+package io.heckel.ntfy
+
+import androidx.lifecycle.LiveData
+import androidx.lifecycle.ViewModel
+import androidx.lifecycle.ViewModelProvider
+import io.heckel.ntfy.data.*
+import kotlin.collections.List
+
+class SubscriptionsViewModel(private val repository: Repository, private val connectionManager: ConnectionManager) : ViewModel() {
+    fun add(topic: Subscription) {
+        repository.add(topic)
+        connectionManager.start(topic)
+    }
+
+    fun get(id: Long) : Subscription? {
+        return repository.get(id)
+    }
+
+    fun list(): LiveData<List<Subscription>> {
+        return repository.list()
+    }
+
+    fun remove(topic: Subscription) {
+        repository.remove(topic)
+        connectionManager.stop(topic)
+    }
+
+    fun setListener(listener: NotificationListener) {
+        connectionManager.setListener(listener)
+    }
+}
+
+class SubscriptionsViewModelFactory : ViewModelProvider.Factory {
+    @Suppress("UNCHECKED_CAST")
+    override fun <T : ViewModel?> create(modelClass: Class<T>) =
+        with(modelClass){
+            when {
+                isAssignableFrom(SubscriptionsViewModel::class.java) -> {
+                    val repository = Repository.getInstance()
+                    val connectionManager = ConnectionManager.getInstance(repository)
+                    SubscriptionsViewModel(repository, connectionManager) as T
+                }
+                else -> throw IllegalArgumentException("Unknown viewModel class $modelClass")
+            }
+        }
+}

app/src/main/java/io/heckel/ntfy/add/AddActivity.kt

@@ -7,7 +7,7 @@ import android.widget.Button
 import androidx.appcompat.app.AppCompatActivity
 import com.google.android.material.textfield.TextInputEditText
 import io.heckel.ntfy.R
-import io.heckel.ntfy.TOPIC_BASE_URL
+import io.heckel.ntfy.SERVICE_BASE_URL
 import io.heckel.ntfy.TOPIC_NAME
 
 class AddTopicActivity : AppCompatActivity() {
@@ -39,7 +39,7 @@ class AddTopicActivity : AppCompatActivity() {
             setResult(Activity.RESULT_CANCELED, resultIntent)
         } else {
             resultIntent.putExtra(TOPIC_NAME, topicName.text.toString())
-            resultIntent.putExtra(TOPIC_BASE_URL, baseUrl.text.toString())
+            resultIntent.putExtra(SERVICE_BASE_URL, baseUrl.text.toString())
             setResult(Activity.RESULT_OK, resultIntent)
         }
         finish()

app/src/main/java/io/heckel/ntfy/data/ConnectionManager.kt

@@ -2,80 +2,88 @@ package io.heckel.ntfy.data
 
 import com.google.gson.GsonBuilder
 import com.google.gson.JsonObject
-import com.google.gson.JsonSyntaxException
 import kotlinx.coroutines.*
-import java.io.IOException
 import java.net.HttpURLConnection
 import java.net.URL
 
 const val READ_TIMEOUT = 60_000 // Keep alive every 30s assumed
 
-class ConnectionManager {
+class ConnectionManager(private val repository: Repository) {
     private val jobs = mutableMapOf<Long, Job>()
     private val gson = GsonBuilder().create()
-    private var listener: ConnectionListener? = null;
+    private var listener: NotificationListener? = null;
 
-    fun start(subscription: Subscription, scope: CoroutineScope) {
+    fun start(s: Subscription) {
-        jobs[subscription.id] = launchConnection(subscription, scope)
+        jobs[s.id] = launchConnection(s.id, topicJsonUrl(s))
     }
 
-    fun stop(subscription: Subscription) {
+    fun stop(s: Subscription) {
-        jobs.remove(subscription.id)?.cancel() // Cancel coroutine and remove
+        jobs.remove(s.id)?.cancel() // Cancel coroutine and remove
     }
 
-    fun setListener(listener: ConnectionListener) {
+    fun setListener(l: NotificationListener) {
-        this.listener = listener
+        this.listener = l
     }
 
-    private fun launchConnection(subscription: Subscription, scope: CoroutineScope): Job {
+    private fun launchConnection(subscriptionId: Long, topicUrl: String): Job {
-        return scope.launch(Dispatchers.IO) {
+        return GlobalScope.launch(Dispatchers.IO) {
             while (isActive) {
-                openConnection(this, subscription)
+                openConnection(subscriptionId, topicUrl)
                 delay(5000) // TODO exponential back-off
             }
         }
     }
 
-    private fun openConnection(scope: CoroutineScope, subscription: Subscription) {
+    private fun openConnection(subscriptionId: Long, topicUrl: String) {
-        val url = "${subscription.baseUrl}/${subscription.topic}/json"
+        println("Connecting to $topicUrl ...")
-        println("Connecting to $url ...")
+        val conn = (URL(topicUrl).openConnection() as HttpURLConnection).also {
-        val conn = (URL(url).openConnection() as HttpURLConnection).also {
             it.doInput = true
             it.readTimeout = READ_TIMEOUT
         }
         try {
-            listener?.onStatusChanged(subscription.id, Status.CONNECTED)
+            updateStatus(subscriptionId, Status.CONNECTED)
             val input = conn.inputStream.bufferedReader()
-            while (scope.isActive) {
+            while (GlobalScope.isActive) {
                 val line = input.readLine() ?: break // Break if EOF is reached, i.e. readLine is null
-                if (!scope.isActive) {
+                if (!GlobalScope.isActive) {
                     break // Break if scope is not active anymore; readLine blocks for a while, so we want to be sure
                 }
-                try {
                 val json = gson.fromJson(line, JsonObject::class.java) ?: break // Break on unexpected line
-                    if (!json.isJsonNull && !json.has("event") && json.has("message")) {
+                val validNotification = !json.isJsonNull
-                        val message = json.get("message").asString
+                        && !json.has("event") // No keepalive or open messages
-                        listener?.onNotification(subscription.id, Notification(subscription, message))
+                        && json.has("message")
-                    }
+                if (validNotification) {
-                } catch (e: JsonSyntaxException) {
+                    notify(subscriptionId, json.get("message").asString)
-                    break // Break on unexpected line
                 }
             }
-        } catch (e: IOException) {
+        } catch (e: Exception) {
-            println("Connection error: " + e.message)
+            println("Connection error: " + e)
         } finally {
             conn.disconnect()
         }
-        listener?.onStatusChanged(subscription.id, Status.CONNECTING)
+        updateStatus(subscriptionId, Status.CONNECTING)
-        println("Connection terminated: $url")
+        println("Connection terminated: $topicUrl")
+    }
+
+    private fun updateStatus(subscriptionId: Long, status: Status) {
+        val subscription = repository.get(subscriptionId)
+        repository.update(subscription?.copy(status = status))
+    }
+
+    private fun notify(subscriptionId: Long, message: String) {
+        val subscription = repository.get(subscriptionId)
+        if (subscription != null) {
+            listener?.let { it(Notification(subscription, message)) }
+            repository.update(subscription.copy(messages = subscription.messages + 1))
+        }
     }
 
     companion object {
         private var instance: ConnectionManager? = null
 
-        fun getInstance(): ConnectionManager {
+        fun getInstance(repository: Repository): ConnectionManager {
             return synchronized(ConnectionManager::class) {
-                val newInstance = instance ?: ConnectionManager()
+                val newInstance = instance ?: ConnectionManager(repository)
                 instance = newInstance
                 newInstance
             }

app/src/main/java/io/heckel/ntfy/data/Models.kt

@@ -17,13 +17,8 @@ data class Notification(
     val message: String
 )
 
-interface NotificationListener {
+typealias NotificationListener = (notification: Notification) -> Unit
-    fun onNotification(subscriptionId: Long, notification: Notification)
-}
-
-interface ConnectionListener : NotificationListener {
-    fun onStatusChanged(subcriptionId: Long, status: Status)
-}
 
 fun topicUrl(s: Subscription) = "${s.baseUrl}/${s.topic}"
+fun topicJsonUrl(s: Subscription) = "${s.baseUrl}/${s.topic}/json"
 fun topicShortUrl(s: Subscription) = topicUrl(s).replace("http://", "").replace("https://", "")

app/src/main/java/io/heckel/ntfy/data/Repository.kt

@@ -14,7 +14,10 @@ class Repository {
         }
     }
 
-    fun update(subscription: Subscription) {
+    fun update(subscription: Subscription?) {
+        if (subscription == null) {
+            return
+        }
         synchronized(subscriptions) {
             val index = subscriptions.indexOfFirst { it.id == subscription.id } // Find index by Topic ID
             if (index == -1) return

app/src/main/java/io/heckel/ntfy/detail/DetailActivity.kt

@@ -22,13 +22,13 @@ import android.widget.TextView
 import androidx.activity.viewModels
 import androidx.appcompat.app.AppCompatActivity
 import io.heckel.ntfy.R
-import io.heckel.ntfy.TOPIC_ID
+import io.heckel.ntfy.SUBSCRIPTION_ID
-import io.heckel.ntfy.SubscriptionViewModel
+import io.heckel.ntfy.SubscriptionsViewModel
 import io.heckel.ntfy.SubscriptionsViewModelFactory
 import io.heckel.ntfy.data.topicShortUrl
 
 class DetailActivity : AppCompatActivity() {
-    private val subscriptionsViewModel by viewModels<SubscriptionViewModel> {
+    private val subscriptionsViewModel by viewModels<SubscriptionsViewModel> {
         SubscriptionsViewModelFactory()
     }
 
@@ -44,7 +44,7 @@ class DetailActivity : AppCompatActivity() {
 
         val bundle: Bundle? = intent.extras
         if (bundle != null) {
-            subscriptionId = bundle.getLong(TOPIC_ID)
+            subscriptionId = bundle.getLong(SUBSCRIPTION_ID)
         }
 
         // TODO This should probably fail hard if topicId is null