Convenient extensions of state and action

rxpm/src/main/kotlin/me/dmdev/rxpm/Action.kt

@@ -25,9 +25,21 @@ class Action<T> internal constructor(internal val pm: PresentationModel) {
 
 /**
  * Creates the [Action].
+ * Optionally subscribes the [action chain][actionChain] to this action.
+ * This chain will be unsubscribed ON [DESTROY][PresentationModel.Lifecycle.DESTROYED].
  */
-fun <T> PresentationModel.action(): Action<T> {
-    return Action(this)
+fun <T> PresentationModel.action(
+    actionChain: (Observable<T>.() -> Observable<*>)? = null
+): Action<T> {
+    val action = Action<T>(pm = this)
+    actionChain?.let { chain ->
+        action.relay
+            .chain()
+            .retry()
+            .subscribe()
+            .untilDestroy()
+    }
+    return action
 }
 
 /**

rxpm/src/main/kotlin/me/dmdev/rxpm/State.kt

@@ -77,18 +77,36 @@ class State<T> internal constructor(
     fun hasValue() = behaviorRelay.hasValue()
 }
 
+private val UNPROCESSED_ERROR_CONSUMER = Consumer<Throwable> { throwable ->
+    throw IllegalStateException(
+        "Unprocessed error encountered in the State. " +
+                "State accepts only emitted items, so you need to process errors yourself.",
+        throwable
+    )
+}
+
 /**
  * Creates the [State].
+ * Optionally subscribes to the provided [state source][stateSource] and
+ * unsubscribes from it ON [DESTROY][PresentationModel.Lifecycle.DESTROYED].
  *
  * @param [initialValue] initial value.
  * @param [diffStrategy] diff strategy.
+ * @param [stateSource] source of the state.
  */
 @Suppress("UNCHECKED_CAST")
 fun <T> PresentationModel.state(
     initialValue: T? = null,
-    diffStrategy: DiffStrategy<T>? = DiffByEquals as DiffStrategy<T>
+    diffStrategy: DiffStrategy<T>? = DiffByEquals as DiffStrategy<T>,
+    stateSource: (() -> Observable<T>)? = null
 ): State<T> {
-    return State(this, initialValue, diffStrategy)
+    val state = State(pm = this, initialValue = initialValue, diffStrategy = diffStrategy)
+    stateSource?.let { source ->
+        source()
+            .subscribe(state.relay, UNPROCESSED_ERROR_CONSUMER)
+            .untilDestroy()
+    }
+    return state
 }
 
 /**

sample/src/main/kotlin/me/dmdev/rxpm/sample/counter/CounterPm.kt

@@ -9,35 +9,24 @@ class CounterPm : PresentationModel() {
     }
 
     val count = state(initialValue = 0)
-    val minusButtonEnabled = state(false)
-    val plusButtonEnabled = state(false)
 
-    val minusButtonClicks = action<Unit>()
-    val plusButtonClicks = action<Unit>()
-
-    override fun onCreate() {
-        super.onCreate()
-
-        count.observable
-            .map { it > 0 }
-            .subscribe(minusButtonEnabled.consumer)
-            .untilDestroy()
+    val minusButtonEnabled = state {
+        count.observable.map { it > 0 }
+    }
 
-        count.observable
-            .map { it < MAX_COUNT }
-            .subscribe(plusButtonEnabled.consumer)
-            .untilDestroy()
+    val plusButtonEnabled = state {
+        count.observable.map { it < MAX_COUNT }
+    }
 
-        minusButtonClicks.observable
-            .filter { count.value > 0 }
+    val minusButtonClicks = action<Unit> {
+        this.filter { count.value > 0 }
             .map { count.value - 1 }
-            .subscribe(count.consumer)
-            .untilDestroy()
+            .doOnNext(count.consumer)
+    }
 
-        plusButtonClicks.observable
-            .filter { count.value < MAX_COUNT }
+    val plusButtonClicks = action<Unit> {
+        this.filter { count.value < MAX_COUNT }
             .map { count.value + 1 }
-            .subscribe(count.consumer)
-            .untilDestroy()
+            .doOnNext(count.consumer)
     }
 }