sphrak · January 6, 2023 14:03 · Jan 6, 2023 · Jan 6, 2023
diff --git a/MessageService.kt b/MessageService.kt
@@ -1,4 +1,3 @@
-
 @Serialized
 data class MessageDto(
     @SerialName(value = "id")
@@ -7,15 +6,17 @@ data class MessageDto(
     val message: String
 )
 
-class MessageService constructor(private val httpClient: HttpClient) {
+class MessageService constructor(
+    private val httpClient: HttpClient
+) {
 
     private val sendMessageFlow: Flow<MessageDto> = MutableSharedFlow()
 
     fun sendMessage(messageDto) {
         sendMessageFlow.emit(messageDto)
     }
 
-    fun messages(): Flow<List<MessageDto>> = flow {
+    fun messages(): Flow<MessageDto> = flow {
         httpClient.wss(
             method = HttpMethod.Get,
             host = getHostname(),
@@ -28,7 +29,7 @@ class MessageService constructor(private val httpClient: HttpClient) {
                     .map(::sendSerialized)
                     .launchIn(this)
                 while (true) {
-                    emit(receiveDeserialized<List<MessageDto>>())
+                    emit(receiveDeserialized<MessageDto>())
                 }
             }
         }

diff --git a/MessageService.kt b/MessageService.kt
@@ -0,0 +1,37 @@
+
+@Serialized
+data class MessageDto(
+    @SerialName(value = "id")
+    val id: String,
+    @SerialName(value = "message")
+    val message: String
+)
+
+class MessageService constructor(private val httpClient: HttpClient) {
+
+    private val sendMessageFlow: Flow<MessageDto> = MutableSharedFlow()
+
+    fun sendMessage(messageDto) {
+        sendMessageFlow.emit(messageDto)
+    }
+
+    fun messages(): Flow<List<MessageDto>> = flow {
+        httpClient.wss(
+            method = HttpMethod.Get,
+            host = getHostname(),
+            port = getPort(),
+            path = "/v1/messages"
+        ) {
+            coroutineScope {
+                sendMessageFlow
+                    .conflate()
+                    .map(::sendSerialized)
+                    .launchIn(this)
+                while (true) {
+                    emit(receiveDeserialized<List<MessageDto>>())
+                }
+            }
+        }
+    }.retryWithBackoff()
+        .flowOn(coroutineDispatcher)
+}
diff --git a/retry.kt b/retry.kt
@@ -0,0 +1,34 @@
+inline val defaultExceptions: Set<Class<out Exception>>
+    get() = setOf(
+        ClosedReceiveChannelException::class.java,
+        ConnectException::class.java,
+        UnresolvedAddressException::class.java
+    )
+/**
+ *  Retries a `Flow<T>` indefinitely with exponential backoff
+ *  
+ *  @param delay -- delay in seconds between retries
+ *  @param delayAtMost -- maximum delay in seconds after which this value is used to delay
+ */
+public fun <T> Flow<T>.retryWithBackoff(
+    delay: Int = 1,
+    delayAtMost: Int = 10,
+    exceptions: Set<Class<out Exception>> = defaultExceptions
+): Flow<T> {
+
+    var _delay = delay
+
+    return retryWhen { cause, attempt ->
+
+        val shouldRetry = exceptions.any { (cause::class.java == it) }
+
+        if (shouldRetry) {
+            delay(_delay.seconds.inWholeMilliseconds)
+            val next = _delay * 2
+            _delay = if (next < delayAtMost) {
+                next
+            } else delayAtMost
+            true
+        } else false
+    }
+}