Refactor FlowRedux (#52)

* refactor FlowReduxStore

* refactor FlowReduxStore

* refactor FlowReduxStore

* refactor FlowReduxStore

* refactor FlowReduxStore

* refactor FlowReduxStore

* refactor FlowReduxStore

* refactor FlowReduxStore

* refactor FlowReduxStore

* Update flowredux/src/commonMain/kotlin/com/hoc081098/flowredux/FlowReduxStore.kt

* Update GithubSearchViewModel.kt

* Update buildSrc/src/main/kotlin/deps.kt [skip ci]

* Update buildSrc/src/main/kotlin/deps.kt  [skip ci]

* Update buildSrc/src/main/kotlin/deps.kt

* Apply suggestions from code review

* refactor FlowReduxStore

* flip

Author: hoc081098

buildSrc/buildSrc/gradle/wrapper/gradle-wrapper.jar

@@ -1,30 +1,33 @@
 package com.hoc081098.flowredux
 
+import kotlin.coroutines.CoroutineContext
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.async
 import kotlinx.coroutines.awaitAll
+import kotlinx.coroutines.cancel
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.coroutineScope
-import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
-import kotlinx.coroutines.flow.SharedFlow
 import kotlinx.coroutines.flow.StateFlow
-import kotlinx.coroutines.flow.asSharedFlow
 import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.flow.buffer
 import kotlinx.coroutines.flow.consumeAsFlow
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.merge
 import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.job
 
 internal class DefaultFlowReduxStore<Action, State>(
-  override val coroutineScope: CoroutineScope,
+  coroutineContext: CoroutineContext,
   initialState: State,
-  sideEffects: List<SideEffect<State, Action>>,
-  reducer: Reducer<State, Action>
+  sideEffects: List<SideEffect<Action, State>>,
+  reducer: Reducer<Action, State>
 ) : FlowReduxStore<Action, State> {
+  private val coroutineScope = CoroutineScope(coroutineContext + Job())
+
   private val _stateFlow = MutableStateFlow(initialState)
   private val _actionChannel = Channel<Action>(Channel.UNLIMITED)
-  private val _actionSharedFlow = MutableSharedFlow<Action>(Channel.UNLIMITED)
 
   override val stateFlow: StateFlow<State> = _stateFlow.asStateFlow()
 
@@ -42,19 +45,25 @@ internal class DefaultFlowReduxStore<Action, State>(
         )
       }
       add(_actionChannel.consumeAsFlow())
-    }.merge()
+    }
+      .merge()
+      .buffer(Channel.UNLIMITED) // buffer all actions, we don't want to miss any action.
 
     actionFlow
       .onEach { action ->
+        // update state
         _stateFlow.value = reducer(_stateFlow.value, action)
 
+        // send action to loopbacks
         loopbacks.sendAll(action)
-        check(_actionSharedFlow.tryEmit(action)) { "Cannot send $action" }
       }
       .launchIn(coroutineScope)
   }
 
-  override val actionSharedFlow: SharedFlow<Action> = _actionSharedFlow.asSharedFlow()
+  override fun close() = coroutineScope.cancel()
+
+  override fun isClosed() = coroutineScope.coroutineContext.job.isCancelled
+
   override fun dispatch(action: Action): Boolean = _actionChannel
     .trySend(action)
     .isSuccess

buildSrc/gradle/wrapper/gradle-wrapper.jar

@@ -1,35 +1,112 @@
 package com.hoc081098.flowredux
 
+import kotlin.coroutines.CoroutineContext
 import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.ReceiveChannel
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.SharedFlow
 import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.emptyFlow
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.mapNotNull
+import kotlinx.coroutines.flow.onCompletion
+import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.job
 
+// TODO: Consider using AutoCloseable since Kotlin 1.8.20
 public sealed interface FlowReduxStore<Action, State> {
-  public val coroutineScope: CoroutineScope
-
+  /**
+   * The state of this store.
+   */
   public val stateFlow: StateFlow<State>
 
-  /** Get streams of actions.
-   *
-   * This [Flow] includes dispatched [Action]s (via [dispatch] function)
-   * and [Action]s returned from [SideEffect]s.
+  /**
+   * @return false if cannot dispatch action (this store was closed).
    */
-  public val actionSharedFlow: SharedFlow<Action>
+  public fun dispatch(action: Action): Boolean
 
   /**
-   * @return false if cannot dispatch action ([coroutineScope] was cancelled).
+   * Call this method to close this store.
+   * A closed store will not accept any action anymore, thus state will not change anymore.
+   * All [SideEffect]s will be cancelled.
    */
-  public fun dispatch(action: Action): Boolean
+  public fun close()
+
+  /**
+   * After calling [close] method, this function will return true.
+   *
+   * @return true if this store was closed.
+   */
+  public fun isClosed(): Boolean
 }
 
+/**
+ * Create a [FlowReduxStore] with [sideEffects] and [reducer].
+ *
+ * The [FlowReduxStore] will be closed when the [CoroutineScope] is cancelled.
+ * That requires the [CoroutineScope] has a [Job] in its context.
+ * And you don't need to call [FlowReduxStore.close] manually.
+ *
+ * @receiver The [CoroutineScope] of the state machine. This scope must have a [Job] in its context.
+ * @param initialState The initial state of the state machine.
+ * @param sideEffects A list of [SideEffect]s.
+ * @param reducer A [Reducer] function.
+ */
 public fun <Action, State> CoroutineScope.createFlowReduxStore(
   initialState: State,
-  sideEffects: List<SideEffect<State, Action>>,
-  reducer: Reducer<State, Action>
+  sideEffects: List<SideEffect<Action, State>>,
+  reducer: Reducer<Action, State>,
+): FlowReduxStore<Action, State> {
+  val store = DefaultFlowReduxStore(
+    coroutineContext = coroutineContext,
+    initialState = initialState,
+    sideEffects = sideEffects,
+    reducer = reducer
+  )
+  coroutineContext.job.invokeOnCompletion {
+    store.close()
+  }
+  return store
+}
+
+/**
+ * Create a [SideEffect] that maps all actions to [Output]s and send them to a [Channel].
+ * The result [Channel] will be closed when the [SideEffect] is cancelled (when calling [FlowReduxStore.close]).
+ *
+ * @param capacity The capacity of the [Channel].
+ * @param transformActionToOutput A function that maps an [Action] to an [Output].
+ * If the function returns `null`, the [Action] will be ignored.
+ * Otherwise, the [Action] will be mapped to an [Output] and sent to the [Channel].
+ * @return A [Pair] of the [SideEffect] and a [Flow] of [Output]s.
+ */
+public fun <Action, State, Output> allActionsToOutputChannelSideEffect(
+  capacity: Int = Channel.UNLIMITED,
+  transformActionToOutput: (Action) -> Output?,
+): Pair<SideEffect<Action, State>, ReceiveChannel<Output>> {
+  val actionChannel = Channel<Output>(capacity)
+
+  val sideEffect = SideEffect<Action, State> { actionFlow, _, coroutineScope ->
+    actionFlow
+      .mapNotNull(transformActionToOutput)
+      .onEach(actionChannel::send)
+      .onCompletion { actionChannel.close() }
+      .launchIn(coroutineScope)
+
+    emptyFlow()
+  }
+
+  return sideEffect to actionChannel
+}
+
+@Suppress("FunctionName") // Factory function
+public fun <Action, State> FlowReduxStore(
+  coroutineContext: CoroutineContext,
+  initialState: State,
+  sideEffects: List<SideEffect<Action, State>>,
+  reducer: Reducer<Action, State>,
 ): FlowReduxStore<Action, State> = DefaultFlowReduxStore(
-  coroutineScope = this,
+  coroutineContext = coroutineContext,
   initialState = initialState,
   sideEffects = sideEffects,
   reducer = reducer

flowredux/src/commonMain/kotlin/com/hoc081098/flowredux/DefaultFlowReduxStore.kt

@@ -6,9 +6,9 @@ package com.hoc081098.flowredux
  *
  * If a reducer should not react on a Action, just return the old State.
  *
- * @param S The type of the state
- * @param A The type of the Actions
+ * @param State The type of the state
+ * @param Action The type of the Actions
  */
-public fun interface Reducer<S, A> {
-  public operator fun invoke(state: S, action: A): S
+public fun interface Reducer<Action, State> {
+  public operator fun invoke(state: State, action: Action): State
 }

flowredux/src/commonMain/kotlin/com/hoc081098/flowredux/FlowReduxStore.kt

@@ -8,17 +8,18 @@ import kotlinx.coroutines.flow.StateFlow
  * It is a function which takes a stream of actions and returns a stream of actions. Actions in, actions out
  * (concept borrowed from redux-observable.js.or - so called epics).
  */
-public fun interface SideEffect<S, A> {
+public fun interface SideEffect<Action, State> {
   /**
    * @param actionFlow Input action. Every SideEffect should be responsible to handle a single Action
    * (i.e using [kotlinx.coroutines.flow.filter] or [kotlinx.coroutines.flow.filterIsInstance] operator)
    * @param stateFlow Allow getting the latest state of the state machine or reacting to state changes.
    * @param coroutineScope The scope of [FlowReduxStore]. It can be used to start a coroutine or
    * share [actionFlow] via [kotlinx.coroutines.flow.shareIn] operator.
+   * @return A Flow of actions. It can be empty if no action should be dispatched.
    */
   public operator fun invoke(
-    actionFlow: Flow<A>,
-    stateFlow: StateFlow<S>,
+    actionFlow: Flow<Action>,
+    stateFlow: StateFlow<State>,
     coroutineScope: CoroutineScope
-  ): Flow<A>
+  ): Flow<Action>
 }