cancel jobs of completed states in DefaultStatesMachine

0.9.14 (upfixes)

CHANGELOG.md

@@ -1,5 +1,20 @@
 # Changelog
 
+## 0.9.15
+
+* `FSM`:
+    * Rename `DefaultUpdatableStatesMachine#compare` to `DefaultUpdatableStatesMachine#shouldReplaceJob`
+    * `DefaultStatesManager` now is extendable
+    * `DefaultStatesMachine` will stop all jobs of states which was removed from `statesManager`
+
+## 0.9.14
+
+* `Versions`:
+    * `Klock`: `2.6.2` -> `2.6.3`
+    * `Ktor`: `1.6.7` -> `1.6.8`
+* `Ktor`:
+    * Add temporal files uploading functionality (for clients to upload and for server to receive)
+
 ## 0.9.13
 
 * `Versions`:

fsm/common/src/commonMain/kotlin/dev/inmo/micro_utils/fsm/common/StatesMachine.kt

@@ -105,6 +105,16 @@ open class DefaultStatesMachine <T: State>(
         statesManager.onChainStateUpdated.subscribeSafelyWithoutExceptions(this) {
             launch { performStateUpdate(Optional.presented(it.first), it.second, scope.LinkedSupervisorScope()) }
         }
+        statesManager.onEndChain.subscribeSafelyWithoutExceptions(this) { removedState ->
+            launch {
+                statesJobsMutex.withLock {
+                    val stateInMap = statesJobs.keys.firstOrNull { stateInMap -> stateInMap == removedState }
+                    if (stateInMap === removedState) {
+                        statesJobs[stateInMap] ?.cancel()
+                    }
+                }
+            }
+        }
 
         statesManager.getActiveStates().forEach {
             launch { performStateUpdate(Optional.absent(), it, scope.LinkedSupervisorScope()) }

fsm/common/src/commonMain/kotlin/dev/inmo/micro_utils/fsm/common/UpdatableStatesMachine.kt

@@ -26,6 +26,12 @@ open class DefaultUpdatableStatesMachine<T : State>(
 ), UpdatableStatesMachine<T> {
     protected val jobsStates = mutableMapOf<Job, T>()
 
+    /**
+     * Realization of this update will use the [Job] of [previousState] in [statesJobs] and [jobsStates] if
+     * [previousState] is [Optional.presented] and [shouldReplaceJob] has returned true for [previousState] and [actualState]. In
+     * other words, [Job] of [previousState] WILL NOT be replaced with the new one if they are "equal". Equality of
+     * states is solved in [shouldReplaceJob] and can be rewritten in subclasses
+     */
     override suspend fun performStateUpdate(previousState: Optional<T>, actualState: T, scope: CoroutineScope) {
         statesJobsMutex.withLock {
             if (compare(previousState, actualState)) {
@@ -52,7 +58,13 @@ open class DefaultUpdatableStatesMachine<T : State>(
         }
     }
 
-    protected open suspend fun compare(previous: Optional<T>, new: T): Boolean = previous.dataOrNull() != new
+    /**
+     * Compare if [previous] potentially lead to the same behaviour with [new]
+     */
+    protected open suspend fun shouldReplaceJob(previous: Optional<T>, new: T): Boolean = previous.dataOrNull() != new
+
+    @Deprecated("Overwrite shouldReplaceJob instead")
+    protected open suspend fun compare(previous: Optional<T>, new: T): Boolean = shouldReplaceJob(previous, new)
 
     override suspend fun updateChain(currentState: T, newState: T) {
         statesManager.update(currentState, newState)

fsm/common/src/commonMain/kotlin/dev/inmo/micro_utils/fsm/common/managers/DefaultStatesManager.kt

@@ -37,30 +37,31 @@ interface DefaultStatesManagerRepo<T : State> {
 
 /**
  * @param repo This repo will be used as repository for storing states. All operations with this repo will happen BEFORE
- * any event will be sent to [onChainStateUpdated], [onStartChain] or [onEndChain]. By default will be used
+ * any event will be sent to [onChainStateUpdated], [onStartChain] or [onEndChain]. By default, will be used
  * [InMemoryDefaultStatesManagerRepo] or you may create custom [DefaultStatesManagerRepo] and pass as [repo] parameter
  * @param onContextsConflictResolver Receive old [State], new one and the state currently placed on new [State.context]
  * key. In case when this callback will returns true, the state placed on [State.context] of new will be replaced by
  * new state by using [endChain] with that state
  */
-class DefaultStatesManager<T : State>(
-    private val repo: DefaultStatesManagerRepo<T> = InMemoryDefaultStatesManagerRepo(),
-    private val onContextsConflictResolver: suspend (old: T, new: T, currentNew: T) -> Boolean = { _, _, _ -> true }
+open class DefaultStatesManager<T : State>(
+    protected val repo: DefaultStatesManagerRepo<T> = InMemoryDefaultStatesManagerRepo(),
+    protected val onContextsConflictResolver: suspend (old: T, new: T, currentNew: T) -> Boolean = { _, _, _ -> true }
 ) : StatesManager<T> {
-    private val _onChainStateUpdated = MutableSharedFlow<Pair<T, T>>(0)
+    protected val _onChainStateUpdated = MutableSharedFlow<Pair<T, T>>(0)
     override val onChainStateUpdated: Flow<Pair<T, T>> = _onChainStateUpdated.asSharedFlow()
-    private val _onStartChain = MutableSharedFlow<T>(0)
+    protected val _onStartChain = MutableSharedFlow<T>(0)
     override val onStartChain: Flow<T> = _onStartChain.asSharedFlow()
-    private val _onEndChain = MutableSharedFlow<T>(0)
+    protected val _onEndChain = MutableSharedFlow<T>(0)
     override val onEndChain: Flow<T> = _onEndChain.asSharedFlow()
 
-    private val mapMutex = Mutex()
+    protected val mapMutex = Mutex()
 
     override suspend fun update(old: T, new: T) = mapMutex.withLock {
         val stateByOldContext: T? = repo.getContextState(old.context)
         when {
             stateByOldContext != old -> return@withLock
             stateByOldContext == null || old.context == new.context -> {
+                repo.removeState(old)
                 repo.set(new)
                 _onChainStateUpdated.emit(old to new)
             }
@@ -83,7 +84,7 @@ class DefaultStatesManager<T : State>(
         }
     }
 
-    private suspend fun endChainWithoutLock(state: T) {
+    protected open suspend fun endChainWithoutLock(state: T) {
         if (repo.getContextState(state.context) == state) {
             repo.removeState(state)
             _onEndChain.emit(state)

gradle.properties

@@ -14,5 +14,5 @@ crypto_js_version=4.1.1
 # Project data
 
 group=dev.inmo
-version=0.9.13
-android_code_version=103
+version=0.9.15
+android_code_version=105

gradle/libs.versions.toml

@@ -8,10 +8,10 @@ jb-compose = "1.1.1"
 jb-exposed = "0.37.3"
 jb-dokka = "1.6.10"
 
-klock = "2.6.2"
+klock = "2.6.3"
 uuid = "0.4.0"
 
-ktor = "1.6.7"
+ktor = "1.6.8"
 
 gh-release = "2.2.12"
 

ktor/client/src/commonMain/kotlin/dev/inmo/micro_utils/ktor/client/TemporalUpload.kt

@@ -0,0 +1,19 @@
+package dev.inmo.micro_utils.ktor.client
+
+import dev.inmo.micro_utils.common.MPPFile
+import dev.inmo.micro_utils.ktor.common.*
+import io.ktor.client.HttpClient
+
+expect suspend fun HttpClient.tempUpload(
+    fullTempUploadDraftPath: String,
+    file: MPPFile,
+    onUpload: (uploaded: Long, count: Long) -> Unit = { _, _ -> }
+): TemporalFileId
+
+suspend fun UnifiedRequester.tempUpload(
+    fullTempUploadDraftPath: String,
+    file: MPPFile,
+    onUpload: (uploaded: Long, count: Long) -> Unit = { _, _ -> }
+): TemporalFileId = client.tempUpload(
+    fullTempUploadDraftPath, file, onUpload
+)

ktor/client/src/jsMain/kotlin/dev/inmo/micro_utils/ktor/client/ActualTemporalUpload.kt

@@ -0,0 +1,58 @@
+package dev.inmo.micro_utils.ktor.client
+
+import dev.inmo.micro_utils.common.MPPFile
+import dev.inmo.micro_utils.ktor.common.TemporalFileId
+import io.ktor.client.HttpClient
+import kotlinx.coroutines.*
+import org.w3c.xhr.*
+
+suspend fun tempUpload(
+    fullTempUploadDraftPath: String,
+    file: MPPFile,
+    onUpload: (Long, Long) -> Unit
+): TemporalFileId {
+    val formData = FormData()
+    val answer = CompletableDeferred<TemporalFileId>()
+
+    formData.append(
+        "data",
+        file
+    )
+
+    val request = XMLHttpRequest()
+    request.responseType = XMLHttpRequestResponseType.TEXT
+    request.upload.onprogress = {
+        onUpload(it.loaded.toLong(), it.total.toLong())
+    }
+    request.onload = {
+        if (request.status == 200.toShort()) {
+            answer.complete(TemporalFileId(request.responseText))
+        } else {
+            answer.completeExceptionally(Exception("Something went wrong: $it"))
+        }
+    }
+    request.onerror = {
+        answer.completeExceptionally(Exception("Something went wrong: $it"))
+    }
+    request.open("POST", fullTempUploadDraftPath, true)
+    request.send(formData)
+
+    val handle = currentCoroutineContext().job.invokeOnCompletion {
+        runCatching {
+            request.abort()
+        }
+    }
+
+    return runCatching {
+        answer.await()
+    }.also {
+        handle.dispose()
+    }.getOrThrow()
+}
+
+
+actual suspend fun HttpClient.tempUpload(
+    fullTempUploadDraftPath: String,
+    file: MPPFile,
+    onUpload: (uploaded: Long, count: Long) -> Unit
+): TemporalFileId = dev.inmo.micro_utils.ktor.client.tempUpload(fullTempUploadDraftPath, file, onUpload)

ktor/client/src/jvmMain/kotlin/dev/inmo/micro_utils/ktor/client/ActualTemporalUpload.kt

@@ -0,0 +1,39 @@
+package dev.inmo.micro_utils.ktor.client
+
+import dev.inmo.micro_utils.common.MPPFile
+import dev.inmo.micro_utils.common.filename
+import dev.inmo.micro_utils.ktor.common.TemporalFileId
+import io.ktor.client.HttpClient
+import io.ktor.client.features.onUpload
+import io.ktor.client.request.forms.formData
+import io.ktor.client.request.forms.submitFormWithBinaryData
+import io.ktor.http.Headers
+import io.ktor.http.HttpHeaders
+import java.net.URLConnection
+
+internal val MPPFile.mimeType: String
+    get() = URLConnection.getFileNameMap().getContentTypeFor(filename.name) ?: "*/*"
+
+actual suspend fun HttpClient.tempUpload(
+    fullTempUploadDraftPath: String,
+    file: MPPFile,
+    onUpload: (Long, Long) -> Unit
+): TemporalFileId {
+    val inputProvider = file.inputProvider()
+    val fileId = submitFormWithBinaryData<String>(
+        fullTempUploadDraftPath,
+        formData = formData {
+            append(
+                "data",
+                inputProvider,
+                Headers.build {
+                    append(HttpHeaders.ContentType, file.mimeType)
+                    append(HttpHeaders.ContentDisposition, "filename=\"${file.filename.string}\"")
+                }
+            )
+        }
+    ) {
+        onUpload(onUpload)
+    }
+    return TemporalFileId(fileId)
+}

ktor/common/build.gradle

@@ -13,6 +13,7 @@ kotlin {
                 api internalProject("micro_utils.common")
                 api libs.kt.serialization.cbor
                 api libs.klock
+                api libs.uuid
             }
         }
     }

ktor/common/src/commonMain/kotlin/dev/inmo/micro_utils/ktor/common/TemporalFiles.kt

@@ -0,0 +1,10 @@
+package dev.inmo.micro_utils.ktor.common
+
+import kotlin.jvm.JvmInline
+import kotlinx.serialization.Serializable
+
+const val DefaultTemporalFilesSubPath = "temp_upload"
+
+@Serializable
+@JvmInline
+value class TemporalFileId(val string: String)

ktor/server/src/jvmMain/kotlin/dev/inmo/micro_utils/ktor/server/ServerRoutingShortcuts.kt

@@ -92,6 +92,11 @@ class UnifiedRouter(
             call.respond(HttpStatusCode.BadRequest, "Request query parameters must contains $field")
         }
     }
+
+    companion object {
+        val default
+            get() = defaultUnifiedRouter
+    }
 }
 
 val defaultUnifiedRouter = UnifiedRouter()

ktor/server/src/jvmMain/kotlin/dev/inmo/micro_utils/ktor/server/TemporalFilesRoutingConfigurator.kt

@@ -0,0 +1,132 @@
+package dev.inmo.micro_utils.ktor.server
+
+import com.benasher44.uuid.uuid4
+import dev.inmo.micro_utils.common.FileName
+import dev.inmo.micro_utils.common.MPPFile
+import dev.inmo.micro_utils.coroutines.launchSafelyWithoutExceptions
+import dev.inmo.micro_utils.ktor.common.DefaultTemporalFilesSubPath
+import dev.inmo.micro_utils.ktor.common.TemporalFileId
+import dev.inmo.micro_utils.ktor.server.configurators.ApplicationRoutingConfigurator
+import io.ktor.application.call
+import io.ktor.http.HttpStatusCode
+import io.ktor.http.content.PartData
+import io.ktor.http.content.streamProvider
+import io.ktor.request.receiveMultipart
+import io.ktor.response.respond
+import io.ktor.routing.Route
+import io.ktor.routing.post
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+import java.io.File
+import java.nio.file.Files
+import java.nio.file.attribute.FileTime
+
+class TemporalFilesRoutingConfigurator(
+    private val subpath: String = DefaultTemporalFilesSubPath,
+    private val unifiedRouter: UnifiedRouter = UnifiedRouter.default,
+    private val temporalFilesUtilizer: TemporalFilesUtilizer = TemporalFilesUtilizer
+) : ApplicationRoutingConfigurator.Element {
+    interface TemporalFilesUtilizer {
+        fun start(filesMap: MutableMap<TemporalFileId, MPPFile>, filesMutex: Mutex, onNewFileFlow: Flow<TemporalFileId>): Job
+
+        companion object : TemporalFilesUtilizer {
+            class ByTimerUtilizer(
+                private val removeMillis: Long,
+                private val scope: CoroutineScope
+            ) : TemporalFilesUtilizer {
+                override fun start(
+                    filesMap: MutableMap<TemporalFileId, MPPFile>,
+                    filesMutex: Mutex,
+                    onNewFileFlow: Flow<TemporalFileId>
+                ): Job = scope.launchSafelyWithoutExceptions {
+                    while (isActive) {
+                        val filesWithCreationInfo = filesMap.mapNotNull { (fileId, file) ->
+                            fileId to ((Files.getAttribute(file.toPath(), "creationTime") as? FileTime) ?.toMillis() ?: return@mapNotNull null)
+                        }
+                        if (filesWithCreationInfo.isEmpty()) {
+                            delay(removeMillis)
+                            continue
+                        }
+                        var min = filesWithCreationInfo.first()
+                        for (fileWithCreationInfo in filesWithCreationInfo) {
+                            if (fileWithCreationInfo.second < min.second) {
+                                min = fileWithCreationInfo
+                            }
+                        }
+                        delay(System.currentTimeMillis() - (min.second + removeMillis))
+                        filesMutex.withLock {
+                            filesMap.remove(min.first)
+                        } ?.delete()
+                    }
+
+                }
+            }
+
+            override fun start(
+                filesMap: MutableMap<TemporalFileId, MPPFile>,
+                filesMutex: Mutex,
+                onNewFileFlow: Flow<TemporalFileId>
+            ): Job = Job()
+        }
+    }
+
+    private val temporalFilesMap = mutableMapOf<TemporalFileId, MPPFile>()
+    private val temporalFilesMutex = Mutex()
+    private val filesFlow = MutableSharedFlow<TemporalFileId>()
+    val utilizerJob = temporalFilesUtilizer.start(temporalFilesMap, temporalFilesMutex, filesFlow.asSharedFlow())
+
+    override fun Route.invoke() {
+        post(subpath) {
+            unifiedRouter.apply {
+                val multipart = call.receiveMultipart()
+
+                var fileInfo: Pair<TemporalFileId, MPPFile>? = null
+                var part = multipart.readPart()
+
+                while (part != null) {
+                    if (part is PartData.FileItem) {
+                        break
+                    }
+                    part = multipart.readPart()
+                }
+
+                part ?.let {
+                    if (it is PartData.FileItem) {
+                        val fileId = TemporalFileId(uuid4().toString())
+                        val fileName = it.originalFileName ?.let { FileName(it) } ?: return@let
+                        fileInfo = fileId to File.createTempFile(fileId.string, ".${fileName.extension}").apply {
+                            outputStream().use { outputStream ->
+                                it.streamProvider().use {
+                                    it.copyTo(outputStream)
+                                }
+                            }
+                            deleteOnExit()
+                        }
+                    }
+                }
+
+                fileInfo ?.also { (fileId, file) ->
+                    temporalFilesMutex.withLock {
+                        temporalFilesMap[fileId] = file
+                    }
+                    call.respond(fileId.string)
+                    launchSafelyWithoutExceptions { filesFlow.emit(fileId) }
+                } ?: call.respond(HttpStatusCode.BadRequest)
+            }
+        }
+    }
+
+    suspend fun removeTemporalFile(temporalFileId: TemporalFileId) {
+        temporalFilesMutex.withLock {
+            temporalFilesMap.remove(temporalFileId)
+        }
+    }
+
+    fun getTemporalFile(temporalFileId: TemporalFileId) = temporalFilesMap[temporalFileId]
+
+    suspend fun getAndRemoveTemporalFile(temporalFileId: TemporalFileId) = temporalFilesMutex.withLock {
+        temporalFilesMap.remove(temporalFileId)
+    }
+}