fix: handle more network errors, and large refactors

composeApp/src/commonMain/kotlin/moe/lava/banksia/ui/BanksiaViewModel.kt

@@ -12,7 +12,6 @@ import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.flow.takeWhile
 import kotlinx.coroutines.flow.update
-import kotlinx.coroutines.flow.updateAndGet
 import kotlinx.coroutines.launch
 import kotlinx.datetime.Clock
 import kotlinx.datetime.Instant
@@ -31,6 +30,7 @@ import moe.lava.banksia.ui.state.MapState
 import moe.lava.banksia.ui.state.SearchState
 import moe.lava.banksia.util.BoxedValue
 import moe.lava.banksia.util.BoxedValue.Companion.box
+import moe.lava.banksia.util.LoopFlow.Companion.waitUntilSubscribed
 
 sealed class BanksiaEvent {
     data object DismissState : BanksiaEvent()
@@ -72,7 +72,7 @@ class BanksiaViewModel : ViewModel() {
     private val iSearchState = MutableStateFlow(SearchState())
     val searchState = iSearchState.asStateFlow()
 
-    private val ptvService = PtvService()
+    private val ptvService = PtvService(viewModelScope)
     private var locationTrackerJob: Job? = null
     private var lastKnownLocation: Point? = null
 
@@ -84,9 +84,9 @@ class BanksiaViewModel : ViewModel() {
         viewModelScope.launch {
             when (event) {
                 is BanksiaEvent.DismissState -> dismissState()
-                is BanksiaEvent.SelectRoute -> switchRoute(event.id)
+                is BanksiaEvent.SelectRoute -> state = InternalState(route = event.id)
-                is BanksiaEvent.SelectRun -> switchRun(event.ref)
+                is BanksiaEvent.SelectRun -> state = state.copy(run = event.ref)
-                is BanksiaEvent.SelectStop -> switchStop(event.typeAndId)
+                is BanksiaEvent.SelectStop -> state = state.copy(stop = event.typeAndId)
                 is BanksiaEvent.SearchUpdate -> searchUpdate(event.text)
             }
         }
@@ -167,13 +167,14 @@ class BanksiaViewModel : ViewModel() {
             return
         }
 
-        var lastState = iInfoState.value
+        val lastState = state.run
         var routeName: String? = null
         ptvService.runFlow(ref, firstWithCache = true)
-            .takeWhile { lastState == iInfoState.value }
+            .waitUntilSubscribed(iInfoState)
+            .takeWhile { lastState == state.run }
             .onEach { run ->
                 if (routeName == null) {
-                    lastState = iInfoState.updateAndGet {
+                    iInfoState.update {
                         InfoPanelState.Run(
                             direction = run.destinationName,
                             type = run.routeType,
@@ -182,7 +183,7 @@ class BanksiaViewModel : ViewModel() {
                     routeName = ptvService.route(run.routeId).routeName
                 }
 
-                lastState = iInfoState.updateAndGet {
+                iInfoState.update {
                     InfoPanelState.Run(
                         direction = run.destinationName,
                         type = run.routeType,
@@ -220,7 +221,7 @@ class BanksiaViewModel : ViewModel() {
         val timetable = HashMap<Pair<Int, Int>, Pair<String, MutableList<String>>>()
         res.departures.forEach { dep ->
             val key = Pair(dep.directionId, dep.routeId)
-            val direction = ptvService.cache.direction(dep.directionId, dep.routeId) ?: return@forEach
+            val direction = ptvService.direction(dep.directionId, dep.routeId)
             val route = res.routes[dep.routeId.toString()]
             val prefix = route?.let { if (it.routeNumber == "") "" else "${it.routeNumber} - " } ?: ""
             val element = timetable.getOrPut(key) { Pair(prefix + direction.directionName, mutableListOf()) }.second
@@ -282,12 +283,11 @@ class BanksiaViewModel : ViewModel() {
         newCameraPosition?.let { iCameraChangeEmitter.emit(it.box()) }
     }
 
-    var runsRouteKey: Int? = null
     private fun buildRuns(route: PtvRoute) {
-        runsRouteKey = route.routeId
         ptvService
             .runsFlow(route.routeId)
-            .takeWhile { route.routeId == runsRouteKey }
+            .waitUntilSubscribed(iInfoState)
+            .takeWhile { state.route == route.routeId }
             .onEach { runs ->
                 val markers = runs
                     .filter { it.vehiclePosition != null }

shared/src/androidMain/kotlin/moe/lava/banksia/Logging.android.kt

@@ -5,3 +5,8 @@ import android.util.Log
 actual fun log(tag: String, msg: String) {
     Log.i(tag, msg)
 }
+
+actual fun error(tag: String, msg: String, throwable: Throwable?) {
+    Log.e(tag, msg)
+    throwable?.let { Log.e(tag, it.stackTraceToString()) }
+}

shared/src/commonMain/kotlin/moe/lava/banksia/Logging.kt

@@ -1,3 +1,5 @@
 package moe.lava.banksia
 
 expect fun log(tag: String, msg: String)
+fun error(tag: String, throwable: Throwable) = error(tag, "", throwable)
+expect fun error(tag: String, msg: String, throwable: Throwable? = null)

shared/src/commonMain/kotlin/moe/lava/banksia/api/ptv/PtvService.kt

@@ -6,13 +6,15 @@ import io.ktor.client.plugins.HttpSend
 import io.ktor.client.plugins.contentnegotiation.ContentNegotiation
 import io.ktor.client.plugins.defaultRequest
 import io.ktor.client.plugins.plugin
+import io.ktor.client.request.HttpRequestBuilder
 import io.ktor.client.request.get
 import io.ktor.client.request.parameter
+import io.ktor.client.request.url
+import io.ktor.client.statement.HttpResponse
 import io.ktor.http.appendPathSegments
 import io.ktor.serialization.kotlinx.json.json
+import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.delay
-import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.flow
 import kotlinx.serialization.Serializable
 import kotlinx.serialization.json.Json
 import moe.lava.banksia.Constants
@@ -22,7 +24,11 @@ import moe.lava.banksia.api.ptv.structures.PtvRoute
 import moe.lava.banksia.api.ptv.structures.PtvRouteType
 import moe.lava.banksia.api.ptv.structures.PtvRun
 import moe.lava.banksia.api.ptv.structures.PtvStop
+import moe.lava.banksia.error
 import moe.lava.banksia.log
+import moe.lava.banksia.util.CacheMap
+import moe.lava.banksia.util.LoopFlow.Companion.initWith
+import moe.lava.banksia.util.loopFlow
 import okio.ByteString.Companion.encodeUtf8
 import kotlin.random.Random
 
@@ -47,52 +53,22 @@ object Responses {
     data class PtvDirectionsResponse(val directions: List<PtvDirection>)
 }
 
-class PtvService {
+suspend inline fun <K, V> MutableMap<K, V>.getOrPutSuspend(key: K, defaultValue: suspend () -> V): V {
+    if (!containsKey(key))
+        this[key] = defaultValue()
+    return this[key]!!
+}
+
+class PtvService(coroutineScope: CoroutineScope) {
     class PtvCache(
-        private val service: PtvService,
+        coroutineScope: CoroutineScope,
-        private val directions: HashMap<Pair<Int, Int>, PtvDirection> = HashMap(),
+        val directions: CacheMap<Pair<Int, Int>, PtvDirection> = CacheMap(coroutineScope),
-        private val routes: HashMap<Int, PtvRoute> = HashMap(),
+        val routes: CacheMap<Int, PtvRoute> = CacheMap(coroutineScope),
-        private val runs: HashMap<String, PtvRun> = HashMap(),
+        val runs: CacheMap<String, PtvRun> = CacheMap(coroutineScope),
-        private val stops: HashMap<Int, PtvStop> = HashMap(),
+        val stops: CacheMap<Int, PtvStop> = CacheMap(coroutineScope),
-    ) {
+    )
-        suspend fun direction(directionID: Int, routeID: Int): PtvDirection? {
-            val ret = directions[Pair(directionID, routeID)]
-            if (ret == null) {
-                val res = service.directionsByRoute(routeID)
-                for (dir in res)
-                    directions[Pair(dir.directionId, dir.routeId)] = dir
-            }
 
-            return ret ?: directions[Pair(directionID, routeID)]
+    val cache = PtvCache(coroutineScope)
-        }
-
-        fun setRoutes(routes: Iterable<PtvRoute>) {
-            routes.forEach {
-                this.routes[it.routeId] = it
-            }
-        }
-
-        fun getRoute(routeId: Int) = routes[routeId]
-        fun getRoutes() = routes.values.toList()
-
-        fun addStops(stops: Iterable<PtvStop>) {
-            stops.forEach {
-                this.stops[it.stopId] = it
-            }
-        }
-
-        fun getStop(stopId: Int) = stops[stopId]
-
-        fun addRuns(runs: Iterable<PtvRun>) {
-            runs.forEach {
-                this.runs[it.runRef] = it
-            }
-        }
-
-        fun getRun(runRef: String) = runs[runRef]
-    }
-
-    val cache = PtvCache(this)
 
     private val client = HttpClient() {
         install(ContentNegotiation) {
@@ -105,7 +81,7 @@ class PtvService {
         }
     }
 
-    constructor() {
+    init {
         client.plugin(HttpSend).intercept { req ->
             req.parameter("devid", Constants.devid)
             @OptIn(ExperimentalStdlibApi::class)
@@ -118,134 +94,146 @@ class PtvService {
         }
     }
 
+    suspend fun HttpClient.safeGet(
+        urlString: String? = null,
+        retries: Int = 1,
+        block: (HttpRequestBuilder.() -> Unit)? = null
+    ): HttpResponse =
+        runCatching {
+            get {
+                urlString?.let { url(it) }
+                block?.invoke(this)
+            }
+        }.getOrElse { e ->
+            error("PtvService", "Fetch error occurred (attempt $retries / 3), retrying in 5000ms...", e)
+            if (retries >= 3)
+                throw e
+            delay(5000)
+            safeGet(urlString, retries + 1, block)
+        }
+
     suspend fun route(id: Int, includeGeopath: Boolean = false): PtvRoute {
-        val cached = cache.getRoute(id)
+        val cached = cache.routes[id]
         // TODO: im braindead so clean this up later
         if (cached != null && (!includeGeopath || (includeGeopath && cached.geopath.isNotEmpty())))
             return cached
 
-        val response: Responses.PtvRouteResponse = client.get("routes") {
+        return client
+            .safeGet("routes") {
                 url {
                     appendPathSegments(id.toString())
                     parameters.append("include_geopath", if (includeGeopath) "true" else "false")
                 }
-        }.body()
+            }
-        cache.setRoutes(listOf(response.route))
+            .body<Responses.PtvRouteResponse>()
-        return response.route
+            .route
+            .also { cache.routes[it.routeId] = it }
     }
 
     suspend fun routes(): List<PtvRoute> {
-        val cached = cache.getRoutes()
+        val cached = cache.routes
-        if (cached.isNotEmpty())
+        if (cached.isEmpty()) {
-            return cached
+            client
-
+                .safeGet("routes")
-        val response: Responses.PtvRoutesResponse = client.get("routes").body()
+                .body<Responses.PtvRoutesResponse>()
-        cache.setRoutes(response.routes)
+                .routes
-        return response.routes
+                .forEach { route ->
+                    cached[route.routeId] = route
+                }
+        }
+        return cached.values.toList()
     }
 
-    fun runFlow(ref: String, firstWithCache: Boolean = false, intervalMillis: Long = 5000): Flow<PtvRun> = flow {
+    fun runFlow(ref: String, firstWithCache: Boolean = false) =
-        val cached = cache.getRun(ref)
+        loopFlow {
-        if (firstWithCache && cached != null)
+            client
-            emit(cached)
+                .safeGet {
-
-        while (true) {
-            val response: Responses.PtvRunsResponse = client.get {
                     url {
-                    appendPathSegments(
+                        appendPathSegments("runs", ref)
-                        "runs",
-                        ref,
-                    )
                     }
-            }.body()
+                }
-            cache.addRuns(response.runs)
+                .body<Responses.PtvRunsResponse>()
-            emit(response.runs[0])
+                .runs
-            delay(intervalMillis)
+                .also { it.forEach { run -> cache.runs[run.runRef] = run } }
+                .let { emit(it[0]) }
+        }.initWith {
+            cache.runs[ref]?.let {
+                if (firstWithCache)
+                    emit(it)
             }
         }
 
-    fun runsFlow(routeId: Int, intervalMillis: Long = 5000): Flow<List<PtvRun>> = flow {
+    fun runsFlow(routeId: Int) =
-        while (true) {
+        loopFlow {
-            val response: Responses.PtvRunsResponse = client.get {
+            client
+                .safeGet {
                     url {
-                    appendPathSegments(
+                        appendPathSegments("runs", "route", routeId.toString())
-                        "runs",
-                        "route",
-                        routeId.toString(),
-                    )
                         parameter("expand", "VehiclePosition")
                     }
-            }.body()
-            cache.addRuns(response.runs)
-            emit(response.runs)
-            delay(intervalMillis)
                 }
+                .body<Responses.PtvRunsResponse>()
+                .runs
+                .also { it.forEach { run -> cache.runs[run.runRef] = run } }
+                .let { emit(it) }
         }
 
-    suspend fun stopsByRoute(routeId: Int, routeType: PtvRouteType): List<PtvStop> {
+    suspend fun stopsByRoute(routeId: Int, routeType: PtvRouteType): List<PtvStop> =
-        val response: Responses.PtvStopsResponse = client.get("stops") {
+        client
+            .safeGet("stops") {
                 url {
                     appendPathSegments(
-                    "route",
+                        "route", routeId.toString(),
-                    routeId.toString(),
+                        "route_type", routeType.ordinal.toString(),
-                    "route_type",
-                    routeType.ordinal.toString()
                     )
                 }
-        }.body()
-        val stops = response.stops
-        cache.addStops(stops)
-        return stops
             }
+            .body<Responses.PtvStopsResponse>()
+            .stops
+            .also { it.forEach { stop -> cache.stops[stop.stopId] = stop } }
 
-    suspend fun stop(routeType: PtvRouteType, stopId: Int): PtvStop {
+    suspend fun stop(routeType: PtvRouteType, stopId: Int): PtvStop =
-        val cached = cache.getStop(stopId)
+        cache.stops.getOrPutSuspend(stopId) {
-        if (cached != null)
+            client
-            return cached
+                .safeGet {
-
-        val response: Responses.PtvStopResponse = client.get() {
                     url {
                         appendPathSegments(
-                    "stops",
+                            "stops", stopId.toString(),
-                    stopId.toString(),
+                            "route_type", routeType.ordinal.toString(),
-                    "route_type",
-                    routeType.ordinal.toString(),
                         )
                     }
-        }.body()
+                }
-        val stop = response.stop
+                .body<Responses.PtvStopResponse>()
-        cache.addStops(listOf(stop))
+                .stop
-        return stop
         }
 
-    suspend fun directionsByRoute(routeId: Int): List<PtvDirection> {
+    suspend fun directionsByRoute(routeId: Int): List<PtvDirection> =
-        val response: Responses.PtvDirectionsResponse = client.get("directions") {
+        client
+            .safeGet("directions") {
                 url {
                     appendPathSegments("route", routeId.toString())
                 }
-        }.body()
+            }
-        return response.directions
+            .body<Responses.PtvDirectionsResponse>()
+            .directions
+
+    suspend fun direction(directionId: Int, routeId: Int): PtvDirection {
+        if (!cache.directions.containsKey(directionId to routeId)) {
+            val directions = directionsByRoute(routeId)
+            for (direction in directions)
+                cache.directions[direction.directionId to direction.routeId] = direction
         }
 
-    suspend fun direction(id: Int, routeType: PtvRouteType?): List<PtvDirection> {
+        return cache.directions[directionId to routeId]!!
-        val response: Responses.PtvDirectionsResponse = client.get("directions") {
-            url {
-                appendPathSegments(id.toString())
-                if (routeType != null)
-                    appendPathSegments("route_type", routeType.ordinal.toString())
-            }
-        }.body()
-        return response.directions
     }
 
     suspend fun departures(routeType: PtvRouteType, stopId: Int): Responses.PtvDeparturesResponse =
-        client.get("departures") {
+        client
+            .safeGet ("departures") {
                 url {
                     appendPathSegments(
-                    "route_type",
+                        "route_type", routeType.ordinal.toString(),
-                    routeType.ordinal.toString(),
+                        "stop", stopId.toString(),
-                    "stop",
-                    stopId.toString()
                     )
                     parameter("expand", "Route")
                     parameter("expand", "Direction")

shared/src/commonMain/kotlin/moe/lava/banksia/util/CacheMap.kt

@@ -0,0 +1,48 @@
+package moe.lava.banksia.util
+
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+import moe.lava.banksia.error
+
+class CacheMap<K, V>(
+    coroutineScope: CoroutineScope,
+    val expiryMinutes: Int = 5,
+    private val innerMap: MutableMap<K, V> = mutableMapOf()
+) : MutableMap<K, V> by innerMap {
+    val keyExpiries = mutableMapOf<K, Int>()
+    var counter = 0
+
+    init {
+        coroutineScope.launch {
+            while (true) {
+                delay(60000)
+                counter += 1
+                keyExpiries
+                    .filterValues { expiry -> expiry >= counter }
+                    .keys
+                    .forEach { key ->
+                        innerMap.remove(key)
+                        keyExpiries.remove(key)
+                    }
+            }
+        }
+    }
+
+    override fun put(key: K, value: V): V? {
+        keyExpiries.put(key, counter + expiryMinutes + 1)
+        return innerMap.put(key, value)
+    }
+
+    override fun putAll(from: Map<out K, V>) {
+        keyExpiries.putAll(from.map { it.key to (counter + expiryMinutes + 1) })
+        innerMap.putAll(from)
+    }
+
+    override val entries: MutableSet<MutableMap.MutableEntry<K, V>>
+        get() {
+            error("CacheMap", ".entries accessed, cloning..", IllegalStateException())
+            return this.entries.toMutableSet()
+        }
+
+}

shared/src/commonMain/kotlin/moe/lava/banksia/util/LoopFlow.kt

@@ -0,0 +1,65 @@
+package moe.lava.banksia.util
+
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.AbstractFlow
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.FlowCollector
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.first
+import moe.lava.banksia.log
+import kotlin.experimental.ExperimentalTypeInference
+
+@OptIn(ExperimentalCoroutinesApi::class)
+class LoopFlow<T>(private val block: suspend FlowCollector<T>.() -> Unit) : AbstractFlow<T>() {
+    private var delayMs: Long = 5000
+    private var init: (suspend FlowCollector<T>.() -> Unit)? = null
+    private var waiter: (suspend () -> Unit)? = null
+
+    override suspend fun collectSafely(collector: FlowCollector<T>) {
+        init?.invoke(collector)
+        while (true) {
+            waiter?.invoke()
+            collector.block()
+            delay(delayMs)
+        }
+    }
+
+    companion object {
+        fun <T> Flow<T>.delayFor(delay: Long) = apply {
+            @Suppress("UnusedFlow")
+            if (this is LoopFlow)
+                this.delayMs = delay;
+            else
+                throw IllegalStateException()
+        }
+
+        fun <T> Flow<T>.initWith(block: suspend FlowCollector<T>.() -> Unit) = apply {
+            @Suppress("UnusedFlow")
+            if (this is LoopFlow)
+                this.init = block;
+            else
+                throw IllegalStateException()
+        }
+
+        fun <T> Flow<T>.waitFor(waiter: suspend () -> Unit) = apply {
+            @Suppress("UnusedFlow")
+            if (this is LoopFlow)
+                this.waiter = waiter;
+            else
+                throw IllegalStateException()
+        }
+
+        fun <T> Flow<T>.waitUntilSubscribed(other: MutableStateFlow<*>) = waitFor {
+            val blocked = other.subscriptionCount.value == 0
+            if (blocked)
+                log("LoopFlow", "blocking flow")
+            other.subscriptionCount.first { it > 0 }
+            if (blocked)
+                log("LoopFlow", "unblocking flow")
+        }
+    }
+}
+
+@OptIn(ExperimentalTypeInference::class)
+fun <T> loopFlow(@BuilderInference block: suspend FlowCollector<T>.() -> Unit) = LoopFlow(block)

shared/src/iosMain/kotlin/moe/lava/banksia/Logging.ios.kt

@@ -3,3 +3,7 @@ package moe.lava.banksia
 actual fun log(tag: String, msg: String) {
     TODO("Not yet implemented")
 }
+
+actual fun error(tag: String, msg: String, throwable: Throwable?) {
+    TODO("Not yet implemented")
+}

shared/src/jvmMain/kotlin/moe/lava/banksia/Logging.jvm.kt

@@ -3,3 +3,8 @@ package moe.lava.banksia
 actual fun log(tag: String, msg: String) {
     println("[$tag] $msg")
 }
+
+actual fun error(tag: String, msg: String, throwable: Throwable?) {
+    println("[$tag] $msg")
+    throwable?.let { println(it.stackTraceToString()) }
+}