Implementing real time updates in the frontend using Flows.

src/backend/src/main/kotlin/pt/ulisboa/ist/pharmacist/repository/medicines/MedicinesRepositoryMem.kt

@@ -29,7 +29,7 @@ class MedicinesRepositoryMem(private val dataSource: MemDataSource) : MedicinesR
             MedicineWithClosestPharmacyDto(
                 medicine = MedicineDto(medicine),
                 closestPharmacy = dataSource.pharmacies.values.filter { pharmacy ->
-                    pharmacy.medicines.map { it.medicine.medicineId }.contains(medicine.medicineId)
+                    pharmacy.medicines.any { it.medicine.medicineId == medicine.medicineId }
                 }.minByOrNull { pharmacy ->
                     if (location == null) return@minByOrNull 0.0
 

src/frontend/app/src/main/kotlin/pt/ulisboa/ist/pharmacist/DependenciesContainer.kt

@@ -3,6 +3,7 @@ package pt.ulisboa.ist.pharmacist
 import com.google.gson.Gson
 import okhttp3.OkHttpClient
 import pt.ulisboa.ist.pharmacist.service.http.PharmacistService
+import pt.ulisboa.ist.pharmacist.service.http.services.medicines.RealTimeUpdatesService
 import pt.ulisboa.ist.pharmacist.session.SessionManager
 
 /**
@@ -15,6 +16,7 @@ import pt.ulisboa.ist.pharmacist.session.SessionManager
 interface DependenciesContainer {
     val jsonEncoder: Gson
     val sessionManager: SessionManager
+    val realTimeUpdatesService: RealTimeUpdatesService
     val pharmacistService: PharmacistService
     val httpClient: OkHttpClient
 }

src/frontend/app/src/main/kotlin/pt/ulisboa/ist/pharmacist/PharmacistApplication.kt

@@ -11,6 +11,7 @@ import com.google.gson.Gson
 import com.google.gson.GsonBuilder
 import okhttp3.OkHttpClient
 import pt.ulisboa.ist.pharmacist.service.http.PharmacistService
+import pt.ulisboa.ist.pharmacist.service.http.services.medicines.RealTimeUpdatesService
 import pt.ulisboa.ist.pharmacist.service.real_time_updates.RealTimeUpdatesBackgroundService
 import pt.ulisboa.ist.pharmacist.session.SessionManager
 import pt.ulisboa.ist.pharmacist.session.SessionManagerSharedPrefs
@@ -41,6 +42,11 @@ class PharmacistApplication : DependenciesContainer, Application() {
         sessionManager = sessionManager
     )
 
+    override val realTimeUpdatesService = RealTimeUpdatesService(
+        apiEndpoint = API_ENDPOINT,
+        sessionManager = sessionManager,
+        httpClient = httpClient
+    )
 
     override fun onCreate() {
         super.onCreate()

src/frontend/app/src/main/kotlin/pt/ulisboa/ist/pharmacist/service/http/services/medicines/RealTimeUpdatesService.kt

@@ -4,7 +4,10 @@ import android.util.Log
 import com.google.gson.Gson
 import kotlinx.coroutines.channels.awaitClose
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.callbackFlow
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
 import okhttp3.OkHttpClient
 import okhttp3.Request
 import okhttp3.Response
@@ -13,7 +16,15 @@ import okhttp3.WebSocketListener
 import okio.ByteString
 import pt.ulisboa.ist.pharmacist.service.http.utils.Uris
 import pt.ulisboa.ist.pharmacist.service.http.utils.fromJson
-import pt.ulisboa.ist.pharmacist.service.real_time_updates.RealTimeUpdate
+import pt.ulisboa.ist.pharmacist.service.real_time_updates.RTU
+import pt.ulisboa.ist.pharmacist.service.real_time_updates.RealTimeUpdateMedicineNotificationPublishingData
+import pt.ulisboa.ist.pharmacist.service.real_time_updates.RealTimeUpdateNewPharmacyPublishingData
+import pt.ulisboa.ist.pharmacist.service.real_time_updates.RealTimeUpdatePharmacyGlobalRatingPublishingData
+import pt.ulisboa.ist.pharmacist.service.real_time_updates.RealTimeUpdatePharmacyMedicineStockPublishingData
+import pt.ulisboa.ist.pharmacist.service.real_time_updates.RealTimeUpdatePharmacyUserFavoritedPublishingData
+import pt.ulisboa.ist.pharmacist.service.real_time_updates.RealTimeUpdatePharmacyUserFlaggedPublishingData
+import pt.ulisboa.ist.pharmacist.service.real_time_updates.RealTimeUpdatePharmacyUserRatingPublishingData
+import pt.ulisboa.ist.pharmacist.service.real_time_updates.RealTimeUpdatePublishingDto
 import pt.ulisboa.ist.pharmacist.service.real_time_updates.RealTimeUpdatesBackgroundService.Companion.TAG
 import pt.ulisboa.ist.pharmacist.session.SessionManager
 import java.net.HttpURLConnection
@@ -25,8 +36,25 @@ class RealTimeUpdatesService(
 ) {
     private var webSocket: WebSocket? = null
 
+    private val updateFlow = MutableSharedFlow<RealTimeUpdatePublishingDto>()
 
-    fun getUpdateFlow(): Flow<RealTimeUpdate> = callbackFlow {
+    init {
+        runBlocking {
+            while (true) {
+                if (sessionManager.isLoggedIn()) {
+                    getWebSocketListenerFlow()
+                }
+                // TODO Synchronization issue?
+                sessionManager.logInFlow.collect { loggedIn ->
+                    if (loggedIn && sessionManager.isLoggedIn()) {
+                        getWebSocketListenerFlow()
+                    }
+                }
+            }
+        }
+    }
+
+    private fun getWebSocketListenerFlow(): Flow<RealTimeUpdatePublishingDto> = callbackFlow {
         val listener = object : WebSocketListener() {
             override fun onOpen(webSocket: WebSocket, response: Response) {
                 super.onOpen(webSocket, response)
@@ -39,7 +67,7 @@ class RealTimeUpdatesService(
                     close()
                     return
                 }
-                val message = Gson().fromJson<RealTimeUpdate>(text)
+                val message = Gson().fromJson<RealTimeUpdatePublishingDto>(text)
                 trySend(message)
                 Log.d(TAG, "WebSocket message received")
             }
@@ -77,17 +105,76 @@ class RealTimeUpdatesService(
 
         webSocket = httpClient.newWebSocket(request, listener)
 
+        launch {
+            sessionManager.logInFlow.collect { loggedIn ->
+                if (!loggedIn) {
+                    close()
+                }
+            }
+        }
+
         awaitClose {
             webSocket?.close(1000, null)
             Log.d(TAG, "WebSocket closed")
         }
     }
-}
 
-object RealTimeUpdateTypes {
-    const val PHARMACY = "pharmacy"
-    const val PHARMACY_MEDICINE_STOCK = "pharmacy-medicine-stock"
-    const val MEDICINE_NOTIFICATION = "medicine-notification"
+
+    suspend fun listenForRealTimeUpdates(
+        onNewPharmacy: (RealTimeUpdateNewPharmacyPublishingData) -> Unit = {},
+        onPharmacyUserRating: (RealTimeUpdatePharmacyUserRatingPublishingData) -> Unit = {},
+        onPharmacyGlobalRating: (RealTimeUpdatePharmacyGlobalRatingPublishingData) -> Unit = {},
+        onPharmacyUserFlagged: (RealTimeUpdatePharmacyUserFlaggedPublishingData) -> Unit = {},
+        onPharmacyUserFavorited: (RealTimeUpdatePharmacyUserFavoritedPublishingData) -> Unit = {},
+        onMedicineStock: (RealTimeUpdatePharmacyMedicineStockPublishingData) -> Unit = {},
+        onMedicineNotification: (RealTimeUpdateMedicineNotificationPublishingData) -> Unit = {}
+    ) {
+        updateFlow.collect { realTimeUpdateDto ->
+            when (val realTimeUpdateClass = RTU.getType(realTimeUpdateDto.type)) {
+                RTU.NEW_PHARMACY -> onNewPharmacy(realTimeUpdateClass.parseJson(realTimeUpdateDto.data))
+                RTU.PHARMACY_USER_RATING -> onPharmacyUserRating(
+                    realTimeUpdateClass.parseJson(
+                        realTimeUpdateDto.data
+                    )
+                )
+
+                RTU.PHARMACY_GLOBAL_RATING -> onPharmacyGlobalRating(
+                    realTimeUpdateClass.parseJson(
+                        realTimeUpdateDto.data
+                    )
+                )
+
+                RTU.PHARMACY_USER_FLAGGED -> onPharmacyUserFlagged(
+                    realTimeUpdateClass.parseJson(
+                        realTimeUpdateDto.data
+                    )
+                )
+
+                RTU.PHARMACY_USER_FAVORITED -> onPharmacyUserFavorited(
+                    realTimeUpdateClass.parseJson(
+                        realTimeUpdateDto.data
+                    )
+                )
+
+                RTU.PHARMACY_MEDICINE_STOCK -> onMedicineStock(
+                    realTimeUpdateClass.parseJson(
+                        realTimeUpdateDto.data
+                    )
+                )
+
+                RTU.MEDICINE_NOTIFICATION -> onMedicineNotification(
+                    realTimeUpdateClass.parseJson(
+                        realTimeUpdateDto.data
+                    )
+                )
+
+                null -> Log.e(
+                    "RealTimeUpdatesService",
+                    "Unknown real time update type: ${realTimeUpdateDto.type}"
+                )
+            }
+        }
+    }
 }
 
 class JsonWebSocket(private val webSocket: WebSocket) {

src/frontend/app/src/main/kotlin/pt/ulisboa/ist/pharmacist/service/real_time_updates/RealTimeUpdatePublishSubscribeData.kt

@@ -1,26 +1,66 @@
 package pt.ulisboa.ist.pharmacist.service.real_time_updates
 
+import com.google.gson.Gson
+import org.jetbrains.annotations.Contract
 import pt.ulisboa.ist.pharmacist.domain.medicines.Medicine
-import pt.ulisboa.ist.pharmacist.service.real_time_updates.RealTimeUpdateTypes.MEDICINE_NOTIFICATION
-import pt.ulisboa.ist.pharmacist.service.real_time_updates.RealTimeUpdateTypes.PHARMACY
-import pt.ulisboa.ist.pharmacist.service.real_time_updates.RealTimeUpdateTypes.PHARMACY_MEDICINE_STOCK
+import pt.ulisboa.ist.pharmacist.domain.pharmacies.Location
+import pt.ulisboa.ist.pharmacist.service.http.connection.APIResult
+import pt.ulisboa.ist.pharmacist.service.http.connection.isSuccess
+import kotlin.contracts.ExperimentalContracts
+import kotlin.contracts.contract
 
 /*
  * Types
  */
 
-object RealTimeUpdateTypes {
-    const val PHARMACY = "pharmacy"
-    const val PHARMACY_MEDICINE_STOCK = "pharmacy-medicine-stock"
-    const val MEDICINE_NOTIFICATION = "medicine-notification"
+enum class RTU(val type: String, val dataClass: Class<out RealTimeUpdatePublishingData>) {
+    NEW_PHARMACY(
+        "new-pharmacy",
+        RealTimeUpdateNewPharmacyPublishingData::class.java
+    ),
+    PHARMACY_USER_RATING(
+        "pharmacy-user-rating",
+        RealTimeUpdatePharmacyUserRatingPublishingData::class.java
+    ),
+    PHARMACY_GLOBAL_RATING(
+        "pharmacy-global-rating",
+        RealTimeUpdatePharmacyGlobalRatingPublishingData::class.java
+    ),
+    PHARMACY_USER_FLAGGED(
+        "pharmacy-user-flagged",
+        RealTimeUpdatePharmacyUserFlaggedPublishingData::class.java
+    ),
+    PHARMACY_USER_FAVORITED(
+        "pharmacy-user-favorited",
+        RealTimeUpdatePharmacyUserFavoritedPublishingData::class.java
+    ),
+    PHARMACY_MEDICINE_STOCK(
+        "pharmacy-medicine-stock",
+        RealTimeUpdatePharmacyMedicineStockPublishingData::class.java
+    ),
+    MEDICINE_NOTIFICATION(
+        "medicine-notification",
+        RealTimeUpdateMedicineNotificationPublishingData::class.java
+    );
+
+    companion object {
+        fun getType(type: String): RTU? {
+            return entries.find { it.type == type }
+        }
+    }
+
+    inline fun <reified T: RealTimeUpdatePublishingData> parseJson(data: String): T {
+        return Gson().fromJson(
+            data,
+            dataClass
+        ) as T
+    }
 }
 
 /*
  * Publishes
  */
 
-typealias RealTimeUpdate = RealTimeUpdatePublishingDto
-
 data class RealTimeUpdatePublishingDto(
     val type: String,
     val data: String
@@ -30,19 +70,69 @@ data class RealTimeUpdatePublishing(
     val type: String,
     val data: RealTimeUpdatePublishingData
 ) {
+    constructor(rtu: RTU, data: RealTimeUpdatePublishingData) : this(rtu.type, data)
+
     companion object {
-        fun pharmacy(pharmacyId: Long, globalRatingSum: Double, numberOfRatings: List<Int>) =
+
+        fun newPharmacy(
+            pharmacyId: Long,
+            name: String,
+            location: Location,
+            pictureUrl: String,
+            globalRating: Double?,
+            numberOfRatings: List<Int>
+        ) =
+            RealTimeUpdatePublishing(
+                RTU.NEW_PHARMACY, RealTimeUpdateNewPharmacyPublishingData(
+                    pharmacyId = pharmacyId,
+                    name = name,
+                    location = location,
+                    pictureUrl = pictureUrl,
+                    globalRating = globalRating,
+                    numberOfRatings = numberOfRatings
+                )
+            )
+
+        fun pharmacyUserRating(pharmacyId: Long, userRating: Int) =
+            RealTimeUpdatePublishing(
+                RTU.PHARMACY_USER_RATING, RealTimeUpdatePharmacyUserRatingPublishingData(
+                    pharmacyId = pharmacyId,
+                    userRating = userRating
+                )
+            )
+
+        fun pharmacyGlobalRating(
+            pharmacyId: Long,
+            globalRatingSum: Double,
+            numberOfRatings: List<Int>
+        ) =
             RealTimeUpdatePublishing(
-                PHARMACY, RealTimeUpdatePharmacyPublishingData(
+                RTU.PHARMACY_GLOBAL_RATING, RealTimeUpdatePharmacyGlobalRatingPublishingData(
                     pharmacyId = pharmacyId,
                     globalRatingSum = globalRatingSum,
                     numberOfRatings = numberOfRatings
                 )
             )
 
+        fun pharmacyUserFlagged(pharmacyId: Long, flagged: Boolean) =
+            RealTimeUpdatePublishing(
+                RTU.PHARMACY_USER_FLAGGED, RealTimeUpdatePharmacyUserFlaggedPublishingData(
+                    pharmacyId = pharmacyId,
+                    flagged = flagged
+                )
+            )
+
+        fun pharmacyUserFavorited(pharmacyId: Long, favorited: Boolean) =
+            RealTimeUpdatePublishing(
+                RTU.PHARMACY_USER_FAVORITED, RealTimeUpdatePharmacyUserFavoritedPublishingData(
+                    pharmacyId = pharmacyId,
+                    favorited = favorited
+                )
+            )
+
         fun pharmacyMedicineStock(pharmacyId: Long, medicineId: Long, stock: Long) =
             RealTimeUpdatePublishing(
-                PHARMACY_MEDICINE_STOCK, RealTimeUpdatePharmacyMedicineStockPublishingData(
+                RTU.PHARMACY_MEDICINE_STOCK, RealTimeUpdatePharmacyMedicineStockPublishingData(
                     pharmacyId = pharmacyId,
                     medicineId = medicineId,
                     stock = stock
@@ -54,7 +144,7 @@ data class RealTimeUpdatePublishing(
             medicineStock: RealTimeUpdateMedicineNotificationPublishingData.MedicineStock
         ) =
             RealTimeUpdatePublishing(
-                MEDICINE_NOTIFICATION, RealTimeUpdateMedicineNotificationPublishingData(
+                RTU.MEDICINE_NOTIFICATION, RealTimeUpdateMedicineNotificationPublishingData(
                     pharmacy = pharmacy,
                     medicineStock = medicineStock
                 )
@@ -64,12 +154,36 @@ data class RealTimeUpdatePublishing(
 
 interface RealTimeUpdatePublishingData
 
-data class RealTimeUpdatePharmacyPublishingData(
+data class RealTimeUpdateNewPharmacyPublishingData(
+    var pharmacyId: Long,
+    val name: String,
+    val location: Location,
+    val pictureUrl: String,
+    val globalRating: Double?,
+    val numberOfRatings: List<Int>
+) : RealTimeUpdatePublishingData
+
+data class RealTimeUpdatePharmacyUserRatingPublishingData(
+    val pharmacyId: Long,
+    val userRating: Int
+) : RealTimeUpdatePublishingData
+
+data class RealTimeUpdatePharmacyGlobalRatingPublishingData(
     val pharmacyId: Long,
     val globalRatingSum: Double,
     val numberOfRatings: List<Int>
 ) : RealTimeUpdatePublishingData
 
+data class RealTimeUpdatePharmacyUserFlaggedPublishingData(
+    var pharmacyId: Long,
+    val flagged: Boolean
+) : RealTimeUpdatePublishingData
+
+data class RealTimeUpdatePharmacyUserFavoritedPublishingData(
+    var pharmacyId: Long,
+    val favorited: Boolean
+) : RealTimeUpdatePublishingData
+
 data class RealTimeUpdatePharmacyMedicineStockPublishingData(
     val pharmacyId: Long,
     val medicineId: Long,