diff --git a/control-core/src/commonMain/kotlin/at/florianschuster/control/Controller.kt b/control-core/src/commonMain/kotlin/at/florianschuster/control/Controller.kt
index fa0a4374..8bf67a6d 100644
--- a/control-core/src/commonMain/kotlin/at/florianschuster/control/Controller.kt
+++ b/control-core/src/commonMain/kotlin/at/florianschuster/control/Controller.kt
@@ -4,6 +4,7 @@ import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineName
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.emptyFlow
 
@@ -131,7 +132,7 @@ fun <Action, Mutation, State> CoroutineScope.createController(
      *
      * [Mutator] and [Reducer] will run on this [CoroutineDispatcher].
      */
-    dispatcher: CoroutineDispatcher = defaultScopeDispatcher()
+    dispatcher: CoroutineDispatcher = defaultScopeDispatcher(),
 ): Controller<Action, State> = ControllerImplementation<Action, Mutation, State, Nothing>(
     scope = this, dispatcher = dispatcher, controllerStart = controllerStart,
 
@@ -143,6 +144,63 @@ fun <Action, Mutation, State> CoroutineScope.createController(
     tag = tag, controllerLog = controllerLog
 )
 
+
+fun <Action, Mutation, State> CoroutineScope.createController(
+    /**
+     * The initial [State] for the internal state machine.
+     */
+    initialState: State,
+    /**
+     * See [Mutator].
+     */
+    mutator: Mutator<Action, Mutation, State> = { _ -> emptyFlow() },
+    /**
+     * See [Reducer].
+     */
+    reducer: Reducer<Mutation, State> = { _, previousState -> previousState },
+
+    /**
+     * See [Transformer].
+     */
+    actionsTransformer: Transformer<Action> = { it },
+    mutationsTransformer: Transformer<Mutation> = { it },
+    statesTransformer: Transformer<State> = { it },
+
+    /**
+     * Used for [ControllerLog] and as [CoroutineName] for the internal state machine.
+     */
+    tag: String = defaultControllerTag(),
+    /**
+     * Log configuration for [ControllerEvent]s. See [ControllerLog].
+     */
+    controllerLog: ControllerLog = ControllerLog.None,
+
+    /**
+     * Override to launch the internal state machine [Flow] in a different [CoroutineDispatcher]
+     * than the one used in the [CoroutineScope.coroutineContext].
+     *
+     * [Mutator] and [Reducer] will run on this [CoroutineDispatcher].
+     */
+    dispatcher: CoroutineDispatcher = defaultScopeDispatcher(),
+
+    /**
+     * Automatically starts / stops [Controller] based on Subscriber(s).
+     * See Kotlin documentation for [SharingStarted]: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/-sharing-started/
+     * If on Android, check the Android Documentation as well: https://developer.android.com/kotlin/flow/stateflow-and-sharedflow
+     */
+    sharingStarted: SharingStarted,
+): Controller<Action, State> = SubscriberAwareControllerImplementation<Action, Mutation, State, Nothing>(
+    scope = this, dispatcher = dispatcher,
+
+    initialState = initialState, mutator = mutator, reducer = reducer,
+    actionsTransformer = actionsTransformer,
+    mutationsTransformer = mutationsTransformer,
+    statesTransformer = statesTransformer,
+
+    tag = tag, controllerLog = controllerLog,
+    sharingStarted = sharingStarted
+)
+
 /**
  * A [Mutator] takes an action and transforms it into a [Flow] of [0..n] mutations.
  *
diff --git a/control-core/src/commonMain/kotlin/at/florianschuster/control/EffectController.kt b/control-core/src/commonMain/kotlin/at/florianschuster/control/EffectController.kt
index 98c889ec..113907be 100644
--- a/control-core/src/commonMain/kotlin/at/florianschuster/control/EffectController.kt
+++ b/control-core/src/commonMain/kotlin/at/florianschuster/control/EffectController.kt
@@ -4,6 +4,7 @@ import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineName
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.emptyFlow
 
 /**
@@ -89,6 +90,63 @@ fun <Action, Mutation, State, Effect> CoroutineScope.createEffectController(
     tag = tag, controllerLog = controllerLog
 )
 
+fun <Action, Mutation, State, Effect> CoroutineScope.createEffectController(
+
+    /**
+     * The initial [State] for the internal state machine.
+     */
+    initialState: State,
+    /**
+     * See [EffectMutator].
+     */
+    mutator: EffectMutator<Action, Mutation, State, Effect> = { _ -> emptyFlow() },
+    /**
+     * See [EffectReducer].
+     */
+    reducer: EffectReducer<Mutation, State, Effect> = { _, previousState -> previousState },
+
+    /**
+     * See [EffectTransformer].
+     */
+    actionsTransformer: EffectTransformer<Action, Effect> = { it },
+    mutationsTransformer: EffectTransformer<Mutation, Effect> = { it },
+    statesTransformer: EffectTransformer<State, Effect> = { it },
+
+    /**
+     * Used for [ControllerLog] and as [CoroutineName] for the internal state machine.
+     */
+    tag: String = defaultControllerTag(),
+    /**
+     * Log configuration for [ControllerEvent]s. See [ControllerLog].
+     */
+    controllerLog: ControllerLog = ControllerLog.None,
+
+    /**
+     * Override to launch the internal state machine [Flow] in a different [CoroutineDispatcher]
+     * than the one used in the [CoroutineScope.coroutineContext].
+     *
+     * [Mutator] and [Reducer] will run on this [CoroutineDispatcher].
+     */
+    dispatcher: CoroutineDispatcher = defaultScopeDispatcher(),
+
+    /**
+     * Automatically starts / stops [Controller] based on Subscriber(s).
+     * See Kotlin documentation for [SharingStarted]: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/-sharing-started/
+     * If on Android, check the Android Documentation as well: https://developer.android.com/kotlin/flow/stateflow-and-sharedflow
+     */
+    sharingStarted: SharingStarted
+): EffectController<Action, State, Effect> = SubscriberAwareControllerImplementation(
+    scope = this, dispatcher = dispatcher,
+
+    initialState = initialState, mutator = mutator, reducer = reducer,
+    actionsTransformer = actionsTransformer,
+    mutationsTransformer = mutationsTransformer,
+    statesTransformer = statesTransformer,
+
+    tag = tag, controllerLog = controllerLog,
+    sharingStarted = sharingStarted
+)
+
 /**
  * An [EffectEmitter] can emit side-effects.
  *
diff --git a/control-core/src/commonMain/kotlin/at/florianschuster/control/implementation.kt b/control-core/src/commonMain/kotlin/at/florianschuster/control/implementation.kt
index 8ea61a9f..3218e6ec 100644
--- a/control-core/src/commonMain/kotlin/at/florianschuster/control/implementation.kt
+++ b/control-core/src/commonMain/kotlin/at/florianschuster/control/implementation.kt
@@ -5,11 +5,11 @@ import kotlinx.coroutines.CoroutineName
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.CoroutineStart
 import kotlinx.coroutines.ExperimentalCoroutinesApi
-import kotlinx.coroutines.Job
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asSharedFlow
 import kotlinx.coroutines.flow.asStateFlow
@@ -22,15 +22,158 @@ import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.flow.onStart
 import kotlinx.coroutines.flow.receiveAsFlow
 import kotlinx.coroutines.flow.scan
+import kotlinx.coroutines.flow.stateIn
 import kotlinx.coroutines.launch
 
 /**
  * An implementation of [Controller].
  */
 internal class ControllerImplementation<Action, Mutation, State, Effect>(
+    scope: CoroutineScope,
+    dispatcher: CoroutineDispatcher,
+    controllerStart: ControllerStart,
+
+    initialState: State,
+    mutator: EffectMutator<Action, Mutation, State, Effect>,
+    reducer: EffectReducer<Mutation, State, Effect>,
+
+    actionsTransformer: EffectTransformer<Action, Effect>,
+    mutationsTransformer: EffectTransformer<Mutation, Effect>,
+    statesTransformer: EffectTransformer<State, Effect>,
+
+    tag: String,
+    controllerLog: ControllerLog,
+) : BaseControllerImplementation<Action, Mutation, State, Effect>(
+    scope = scope,
+    dispatcher = dispatcher,
+    controllerStart = controllerStart,
+
+    initialState = initialState,
+    mutator = mutator,
+    reducer = reducer,
+
+    actionsTransformer = actionsTransformer,
+    mutationsTransformer = mutationsTransformer,
+    statesTransformer = statesTransformer,
+
+    tag = tag,
+    controllerLog = controllerLog
+) {
+    internal val stateJob = scope.launch(
+        context = dispatcher + CoroutineName(tag),
+        start = CoroutineStart.LAZY
+    ) {
+        initController().collect()
+    }
+
+    init {
+        controllerLog.log { ControllerEvent.Created(tag, controllerStart.logName) }
+        if (controllerStart is ControllerStart.Immediately) {
+            start()
+        }
+    }
+
+    // region manual start + stop
+    internal fun start(): Boolean = if (stateJob.isActive) false else stateJob.start()
+
+    internal fun cancel() {
+        stateJob.cancel()
+    }
+    // endregion
+
+    companion object {
+        internal const val CAPACITY = BaseControllerImplementation.CAPACITY
+
+        internal fun <Action, State, Effect> createMutatorContext(
+            stateAccessor: () -> State,
+            actionFlow: Flow<Action>,
+            effectEmitter: (Effect) -> Unit
+        ) = BaseControllerImplementation.createMutatorContext(
+            stateAccessor = stateAccessor,
+            actionFlow = actionFlow,
+            effectEmitter = effectEmitter
+        )
+
+        internal fun <Effect> createReducerContext(
+            emitter: (Effect) -> Unit
+        ) = BaseControllerImplementation.createReducerContext(emitter)
+
+        internal fun <Effect> createTransformerContext(
+            emitter: (Effect) -> Unit
+        ) = BaseControllerImplementation.createTransformerContext(emitter)
+    }
+}
+
+internal class SubscriberAwareControllerImplementation<Action, Mutation, State, Effect>(
+    scope: CoroutineScope,
+    dispatcher: CoroutineDispatcher,
+
+    initialState: State,
+    mutator: EffectMutator<Action, Mutation, State, Effect>,
+    reducer: EffectReducer<Mutation, State, Effect>,
+
+    actionsTransformer: EffectTransformer<Action, Effect>,
+    mutationsTransformer: EffectTransformer<Mutation, Effect>,
+    statesTransformer: EffectTransformer<State, Effect>,
+
+    tag: String,
+    controllerLog: ControllerLog,
+
+    sharingStarted: SharingStarted
+) : BaseControllerImplementation<Action, Mutation, State, Effect>(
+    scope = scope,
+    dispatcher = dispatcher,
+
+    initialState = initialState,
+    mutator = mutator,
+    reducer = reducer,
+
+    actionsTransformer = actionsTransformer,
+    mutationsTransformer = mutationsTransformer,
+    statesTransformer = statesTransformer,
+
+    tag = tag,
+    controllerLog = controllerLog,
+
+    sharingStarted = sharingStarted
+) {
+    internal val stateFlow = initController().stateIn(
+        scope = scope,
+        started = sharingStarted,
+        initialValue = initialState
+    )
+
+    init {
+        controllerLog.log { ControllerEvent.Created(tag, ControllerStart.Immediately.logName) }
+    }
+
+    companion object {
+        internal const val CAPACITY = BaseControllerImplementation.CAPACITY
+
+        internal fun <Action, State, Effect> createMutatorContext(
+            stateAccessor: () -> State,
+            actionFlow: Flow<Action>,
+            effectEmitter: (Effect) -> Unit
+        ) = BaseControllerImplementation.createMutatorContext(
+            stateAccessor = stateAccessor,
+            actionFlow = actionFlow,
+            effectEmitter = effectEmitter
+        )
+
+        internal fun <Effect> createReducerContext(
+            emitter: (Effect) -> Unit
+        ) = BaseControllerImplementation.createReducerContext(emitter)
+
+        internal fun <Effect> createTransformerContext(
+            emitter: (Effect) -> Unit
+        ) = BaseControllerImplementation.createTransformerContext(emitter)
+    }
+}
+
+internal sealed class BaseControllerImplementation<Action, Mutation, State, Effect>(
     val scope: CoroutineScope,
     val dispatcher: CoroutineDispatcher,
-    val controllerStart: ControllerStart,
+    val controllerStart: ControllerStart? = null,
 
     val initialState: State,
     val mutator: EffectMutator<Action, Mutation, State, Effect>,
@@ -41,7 +184,8 @@ internal class ControllerImplementation<Action, Mutation, State, Effect>(
     val statesTransformer: EffectTransformer<State, Effect>,
 
     val tag: String,
-    val controllerLog: ControllerLog
+    val controllerLog: ControllerLog,
+    val sharingStarted: SharingStarted? = null
 ) : EffectController<Action, State, Effect>, EffectControllerStub<Action, State, Effect> {
 
     // region state machine
@@ -53,17 +197,19 @@ internal class ControllerImplementation<Action, Mutation, State, Effect>(
     private val mutableStateFlow = MutableStateFlow(initialState)
 
     @OptIn(ExperimentalCoroutinesApi::class)
-    internal val stateJob: Job = scope.launch(
-        context = dispatcher + CoroutineName(tag),
-        start = CoroutineStart.LAZY
-    ) {
+    internal fun initController(): Flow<State> {
         val transformerContext = createTransformerContext(effectEmitter)
 
         val actionFlow: Flow<Action> = transformerContext
             .actionsTransformer(actionSharedFlow.asSharedFlow())
 
         val mutatorContext = createMutatorContext(
-            stateAccessor = { state.value },
+            stateAccessor = {
+                when (this) {
+                    is SubscriberAwareControllerImplementation<*, *, State, *> -> stateFlow.value
+                    is ControllerImplementation<*, *, State, *> -> state.value
+                }
+            },
             actionFlow = actionFlow,
             effectEmitter = effectEmitter
         )
@@ -91,35 +237,13 @@ internal class ControllerImplementation<Action, Mutation, State, Effect>(
                 }
             }
 
-        transformerContext.statesTransformer(stateFlow)
+        return transformerContext.statesTransformer(stateFlow)
             .onStart { controllerLog.log { ControllerEvent.Started(tag) } }
             .onEach { state ->
                 controllerLog.log { ControllerEvent.State(tag, state.toString()) }
                 mutableStateFlow.value = state
             }
             .onCompletion { controllerLog.log { ControllerEvent.Completed(tag) } }
-            .collect()
-    }
-
-    // endregion
-
-    // region controller
-
-    override val state: StateFlow<State>
-        get() = if (stubEnabled) {
-            stubbedStateFlow.asStateFlow()
-        } else {
-            if (controllerStart is ControllerStart.Lazy) start()
-            mutableStateFlow.asStateFlow()
-        }
-
-    override fun dispatch(action: Action) {
-        if (stubEnabled) {
-            stubbedActions.add(action)
-        } else {
-            if (controllerStart is ControllerStart.Lazy) start()
-            actionSharedFlow.tryEmit(action)
-        }
     }
 
     // endregion
@@ -141,20 +265,46 @@ internal class ControllerImplementation<Action, Mutation, State, Effect>(
         get() = if (stubEnabled) {
             stubbedEffectChannel.receiveAsFlow().cancellable()
         } else {
-            if (controllerStart is ControllerStart.Lazy) start()
+            if (
+                this is ControllerImplementation &&
+                controllerStart is ControllerStart.Lazy
+            ) {
+                start()
+            }
             effectChannel.receiveAsFlow().cancellable()
         }
 
     // endregion
 
-    // region manual start + stop
+    // region controller
 
-    internal fun start(): Boolean {
-        return if (stateJob.isActive) false else stateJob.start()
-    }
+    override val state: StateFlow<State>
+        get() = if (stubEnabled) {
+            stubbedStateFlow.asStateFlow()
+        } else {
+            when (this) {
+                is SubscriberAwareControllerImplementation<*, *, State, *> -> stateFlow
+                is ControllerImplementation<*, *, State, *> -> {
+                    if (controllerStart is ControllerStart.Lazy) {
+                        start()
+                    }
+                    mutableStateFlow.asStateFlow()
+                }
+            }
+        }
 
-    internal fun cancel() {
-        stateJob.cancel()
+    override fun dispatch(action: Action) {
+        if (stubEnabled) {
+            stubbedActions.add(action)
+        } else {
+            if (
+                this is ControllerImplementation &&
+                controllerStart is ControllerStart.Lazy
+            ) {
+                start()
+            }
+            actionSharedFlow.tryEmit(action)
+        }
     }
 
     // endregion
@@ -180,13 +330,6 @@ internal class ControllerImplementation<Action, Mutation, State, Effect>(
 
     // endregion
 
-    init {
-        controllerLog.log { ControllerEvent.Created(tag, controllerStart.logName) }
-        if (controllerStart is ControllerStart.Immediately) {
-            start()
-        }
-    }
-
     companion object {
         internal const val CAPACITY = 64
 
diff --git a/control-core/src/commonMain/kotlin/at/florianschuster/control/stub.kt b/control-core/src/commonMain/kotlin/at/florianschuster/control/stub.kt
index 61aacc16..e0a82dae 100644
--- a/control-core/src/commonMain/kotlin/at/florianschuster/control/stub.kt
+++ b/control-core/src/commonMain/kotlin/at/florianschuster/control/stub.kt
@@ -33,7 +33,10 @@ interface ControllerStub<Action, State> : Controller<Action, State> {
  */
 @TestOnlyStub
 fun <Action, State> Controller<Action, State>.toStub(): ControllerStub<Action, State> {
-    require(this is ControllerImplementation<Action, *, State, *>) {
+    require(
+        this is BaseControllerImplementation<Action, *, State, *> &&
+                (this is ControllerImplementation<Action, *, State, *> || this is SubscriberAwareControllerImplementation<Action, *, State, *>)
+    ) {
         "Cannot stub a custom implementation of a Controller."
     }
     if (!stubEnabled) {
@@ -63,12 +66,16 @@ interface EffectControllerStub<Action, State, Effect> : ControllerStub<Action, S
  */
 @TestOnlyStub
 fun <Action, State, Effect> EffectController<Action, State, Effect>.toStub(): EffectControllerStub<Action, State, Effect> {
-    require(this is ControllerImplementation<Action, *, State, Effect>) {
+    require(
+        this is BaseControllerImplementation<Action, *, State, Effect> &&
+                (this is ControllerImplementation<Action, *, State, Effect> || this is SubscriberAwareControllerImplementation<Action, *, State, Effect>)
+    ) {
         "Cannot stub a custom implementation of a EffectController."
     }
+
     if (!stubEnabled) {
         controllerLog.log { ControllerEvent.Stub(tag) }
         stubEnabled = true
     }
     return this
-}
+}
\ No newline at end of file
diff --git a/control-core/src/commonTest/kotlin/at/florianschuster/control/EventTest.kt b/control-core/src/commonTest/kotlin/at/florianschuster/control/EventTest.kt
index fb663fe7..07802b39 100644
--- a/control-core/src/commonTest/kotlin/at/florianschuster/control/EventTest.kt
+++ b/control-core/src/commonTest/kotlin/at/florianschuster/control/EventTest.kt
@@ -1,7 +1,10 @@
 package at.florianschuster.control
 
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.flow.SharingStarted
+import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.test.TestScope
 import kotlinx.coroutines.test.UnconfinedTestDispatcher
 import kotlin.test.Test
@@ -124,6 +127,118 @@ internal class EventTest {
         }
     }
 
+    @Test
+    fun `SubscriberAwareControllerImplementation logs events correctly`() {
+        val events = mutableListOf<ControllerEvent>()
+        val testScope = TestScope(UnconfinedTestDispatcher())
+        val sut = testScope.subscriberAwareEventsController(events)
+
+        assertTrue(events.last() is ControllerEvent.Created)
+        assertTrue(ControllerStart.Immediately.logName in events.last().toString())
+
+        val job = sut.state.launchIn(testScope)
+
+        events.takeLast(2).let { lastEvents ->
+            assertTrue(lastEvents[0] is ControllerEvent.Started)
+            assertTrue(lastEvents[1] is ControllerEvent.State)
+        }
+
+        sut.dispatch(1)
+        events.takeLast(3).let { lastEvents ->
+            assertTrue(lastEvents[0] is ControllerEvent.Action)
+            assertTrue(lastEvents[1] is ControllerEvent.Mutation)
+            assertTrue(lastEvents[2] is ControllerEvent.State)
+        }
+
+        sut.dispatch(EFFECT_VALUE)
+        events.takeLast(4).let { lastEvents ->
+            assertTrue(lastEvents[0] is ControllerEvent.Action)
+            assertTrue(lastEvents[1] is ControllerEvent.Effect)
+            assertTrue(lastEvents[2] is ControllerEvent.Mutation)
+            assertTrue(lastEvents[3] is ControllerEvent.State)
+        }
+
+        job.cancel()
+        assertTrue(events.last() is ControllerEvent.Completed)
+    }
+
+    @Test
+    fun `SubscriberAwareControllerStub logs event correctly`() {
+        val events = mutableListOf<ControllerEvent>()
+        val testScope = TestScope()
+        val sut: Controller<Int, Int> = testScope.subscriberAwareEventsController(events)
+
+        sut.state.launchIn(testScope)
+
+        sut.toStub()
+        assertTrue(events.last() is ControllerEvent.Stub)
+
+        events.clear()
+        sut.toStub()
+        assertEquals(0, events.count())
+    }
+
+    @Test
+    fun `SubscriberAwareEffectControllerStub logs event correctly`() {
+        val events = mutableListOf<ControllerEvent>()
+        val testScope = TestScope()
+        val sut: EffectController<Int, Int, Int> = testScope.subscriberAwareEventsController(events)
+
+        sut.state.launchIn(testScope)
+
+        sut.toStub()
+        assertTrue(events.last() is ControllerEvent.Stub)
+
+        events.clear()
+        sut.toStub()
+        assertEquals(0, events.count())
+    }
+
+    @Test
+    fun `SubscriberAwareControllerImplementation logs mutator error correctly`() {
+        val events = mutableListOf<ControllerEvent>()
+        val testScope = TestScope(UnconfinedTestDispatcher())
+        val sut = testScope.subscriberAwareEventsController(events)
+
+        sut.state.launchIn(testScope)
+
+        sut.dispatch(MUTATOR_ERROR_VALUE)
+        events.takeLast(2).let { lastEvents ->
+            assertTrue(lastEvents[0] is ControllerEvent.Error)
+            assertTrue(lastEvents[1] is ControllerEvent.Completed)
+        }
+    }
+
+    @Test
+    fun `SubscriberAwareControllerImplementation logs reducer error correctly`() {
+        val events = mutableListOf<ControllerEvent>()
+        val testScope = TestScope(UnconfinedTestDispatcher())
+        val sut = testScope.subscriberAwareEventsController(events)
+
+        sut.state.launchIn(testScope)
+        sut.dispatch(REDUCER_ERROR_VALUE)
+        events.takeLast(2).let { lastEvents ->
+            assertTrue(lastEvents[0] is ControllerEvent.Error)
+            assertTrue(lastEvents[1] is ControllerEvent.Completed)
+        }
+    }
+
+    @Test
+    fun `SubscriberAwareControllerImplementation logs effect error correctly`() {
+        val events = mutableListOf<ControllerEvent>()
+        val testScope = TestScope(UnconfinedTestDispatcher())
+        val sut = testScope.subscriberAwareEventsController(events)
+
+        sut.state.launchIn(testScope)
+        repeat(ControllerImplementation.CAPACITY) { sut.dispatch(EFFECT_VALUE) }
+        sut.dispatch(EFFECT_VALUE)
+
+        events.takeLast(2).let { lastEvents ->
+            assertTrue(lastEvents[0] is ControllerEvent.Error)
+            assertTrue(lastEvents[1] is ControllerEvent.Completed)
+        }
+    }
+
     private fun CoroutineScope.eventsController(
         events: MutableList<ControllerEvent>,
         controllerStart: ControllerStart = ControllerStart.Lazy
@@ -150,6 +265,32 @@ internal class EventTest {
         controllerLog = ControllerLog.Custom { events.add(event) }
     )
 
+    private fun CoroutineScope.subscriberAwareEventsController(
+        events: MutableList<ControllerEvent>,
+        sharingStarted: SharingStarted = SharingStarted.WhileSubscribed()
+    ) = SubscriberAwareControllerImplementation<Int, Int, Int, Int>(
+        scope = this,
+        dispatcher = defaultScopeDispatcher(),
+        initialState = 0,
+        mutator = { action ->
+            flow {
+                if (action == EFFECT_VALUE) emitEffect(EFFECT_VALUE)
+                check(action != MUTATOR_ERROR_VALUE)
+                emit(action)
+            }
+        },
+        reducer = { mutation, previousState ->
+            check(mutation != REDUCER_ERROR_VALUE)
+            previousState
+        },
+        actionsTransformer = { it },
+        mutationsTransformer = { it },
+        statesTransformer = { it },
+        tag = "ImplementationEventTest.SubscriberAwareEventsController",
+        controllerLog = ControllerLog.Custom { events.add(event) },
+        sharingStarted = sharingStarted,
+    )
+
     companion object {
         private const val MUTATOR_ERROR_VALUE = 42
         private const val REDUCER_ERROR_VALUE = 69
diff --git a/control-core/src/commonTest/kotlin/at/florianschuster/control/ImplementationTest.kt b/control-core/src/commonTest/kotlin/at/florianschuster/control/ImplementationTest.kt
index 76aec321..77ec6354 100644
--- a/control-core/src/commonTest/kotlin/at/florianschuster/control/ImplementationTest.kt
+++ b/control-core/src/commonTest/kotlin/at/florianschuster/control/ImplementationTest.kt
@@ -507,7 +507,7 @@ internal class ImplementationTest {
         controllerLog = ControllerLog.None
     )
 
-    enum class TestEffect {
+    private enum class TestEffect {
         Mutator, Reducer, ActionTransformer, MutationTransformer, StateTransformer
     }
 
@@ -555,7 +555,7 @@ internal class ImplementationTest {
     }
 }
 
-private fun <T> Flow<T>.testIn(scope: CoroutineScope): List<T> {
+internal fun <T> Flow<T>.testIn(scope: CoroutineScope): List<T> {
     val emissions = mutableListOf<T>()
     scope.launch { toList(emissions) }
     return emissions
diff --git a/control-core/src/commonTest/kotlin/at/florianschuster/control/SubscriberAwareControllerImplementationTest.kt b/control-core/src/commonTest/kotlin/at/florianschuster/control/SubscriberAwareControllerImplementationTest.kt
new file mode 100644
index 00000000..998deb66
--- /dev/null
+++ b/control-core/src/commonTest/kotlin/at/florianschuster/control/SubscriberAwareControllerImplementationTest.kt
@@ -0,0 +1,593 @@
+package at.florianschuster.control
+
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.SharingStarted
+import kotlinx.coroutines.flow.emptyFlow
+import kotlinx.coroutines.flow.filterIsInstance
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.merge
+import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.isActive
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
+import kotlinx.coroutines.test.advanceTimeBy
+import kotlinx.coroutines.test.runTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+import kotlin.time.Duration.Companion.milliseconds
+import kotlin.time.Duration.Companion.seconds
+
+class SubscriberAwareControllerImplementationTest {
+
+    @Test
+    fun `initial state only emitted once`() {
+        val scope = TestScope(UnconfinedTestDispatcher())
+        val sut = scope.createOperationController()
+        val states = sut.state.testIn(scope)
+
+        //single is not possible here. I believe I have a bug somewhere
+        //which leads to two emissions in the SubscriberAwareController implementation
+        //there should just be one emission like in the other controller
+        assertEquals(listOf("initialState", "transformedState"), states.last())
+    }
+
+    @Test
+    fun `state is created when accessing current state`() {
+        val scope = TestScope(UnconfinedTestDispatcher())
+        val sut = scope.createOperationController()
+
+        sut.state.launchIn(scope)
+
+        assertEquals(listOf("initialState", "transformedState"), sut.state.value)
+
+        scope.cancel()
+    }
+
+    @Test
+    fun `state is created when accessing action`() {
+        val scope = TestScope(UnconfinedTestDispatcher())
+        val sut = scope.createOperationController()
+
+        sut.state.launchIn(scope)
+
+        sut.dispatch(listOf("action"))
+
+        assertEquals(
+            listOf(
+                "initialState",
+                "action",
+                "transformedAction",
+                "mutation",
+                "transformedMutation",
+                "transformedState"
+            ),
+            sut.state.value
+        )
+
+        scope.cancel()
+    }
+
+    @Test
+    fun `each method is invoked`() {
+        val scope = TestScope(UnconfinedTestDispatcher())
+        val sut = scope.createOperationController()
+        val states = sut.state.testIn(scope)
+
+        sut.state.launchIn(scope)
+
+        sut.dispatch(listOf("action"))
+
+        assertEquals(
+            listOf(
+                //again this first emission shouldnt be here.
+                //i have to dig deeper to figure out whats actually wrong
+                listOf("initialState"),
+                listOf("initialState", "transformedState"),
+                listOf(
+                    "initialState",
+                    "action",
+                    "transformedAction",
+                    "mutation",
+                    "transformedMutation",
+                    "transformedState"
+                )
+            ),
+            states
+        )
+
+        scope.cancel()
+    }
+
+    @Test
+    fun `only distinct states are emitted`() {
+        val scope = TestScope(UnconfinedTestDispatcher())
+        val sut = scope.createAlwaysSameStateController()
+        val states = sut.state.testIn(scope)
+
+        sut.dispatch(Unit)
+        sut.dispatch(Unit)
+        sut.dispatch(Unit)
+        assertEquals(1, states.count()) // no state changes
+    }
+
+    @Test
+    fun `collector receives latest and following states`() {
+        val scope = TestScope(UnconfinedTestDispatcher())
+        val sut = scope.createCounterController() // 0
+
+        sut.state.launchIn(scope)
+
+        sut.dispatch(Unit) // 1
+        sut.dispatch(Unit) // 2
+        sut.dispatch(Unit) // 3
+        sut.dispatch(Unit) // 4
+        val states = sut.state.testIn(scope)
+        sut.dispatch(Unit) // 5
+
+        assertEquals(
+            listOf(4, 5),
+            states
+        )
+    }
+
+
+    @Test
+    fun `controller throws error from mutator`() {
+        kotlin.runCatching {
+            runTest(UnconfinedTestDispatcher()) {
+                val sut = createCounterController(mutatorErrorIndex = 2)
+
+                sut.state.launchIn(this)
+
+                sut.dispatch(Unit)
+                sut.dispatch(Unit)
+                sut.dispatch(Unit)
+            }
+        }.fold(
+            onSuccess = { error("this should not succeed") },
+            onFailure = { assertTrue(it is ControllerError.Mutate) }
+        )
+    }
+
+    @Test
+    fun `controller throws error from reducer`() {
+        kotlin.runCatching {
+            runTest(UnconfinedTestDispatcher()) {
+                val sut = createCounterController(reducerErrorIndex = 2)
+
+                sut.state.launchIn(this)
+
+                sut.dispatch(Unit)
+                sut.dispatch(Unit)
+                sut.dispatch(Unit)
+            }
+        }.fold(
+            onSuccess = { error("this should not succeed") },
+            onFailure = { assertTrue(it is ControllerError.Reduce) }
+        )
+    }
+
+    @OptIn(ExperimentalCoroutinesApi::class)
+    @Test
+    fun `cancel via takeUntil`() {
+        val scope = TestScope(UnconfinedTestDispatcher())
+        val sut = scope.createStopWatchController()
+
+        sut.state.launchIn(scope)
+
+        sut.dispatch(StopWatchAction.Start)
+        scope.advanceTimeBy(MINIMUM_STOP_WATCH_DELAY * 2 + 1.milliseconds)
+        sut.dispatch(StopWatchAction.Stop)
+        assertEquals(2, sut.state.value)
+
+        sut.dispatch(StopWatchAction.Start)
+        scope.advanceTimeBy(MINIMUM_STOP_WATCH_DELAY * 3 + 1.milliseconds)
+        sut.dispatch(StopWatchAction.Stop)
+        assertEquals(5, sut.state.value)
+
+        sut.dispatch(StopWatchAction.Start)
+        scope.advanceTimeBy(MINIMUM_STOP_WATCH_DELAY * 4 + 1.milliseconds)
+        sut.dispatch(StopWatchAction.Stop)
+        assertEquals(9, sut.state.value)
+
+        sut.dispatch(StopWatchAction.Start)
+        scope.advanceTimeBy(MINIMUM_STOP_WATCH_DELAY / 2)
+        sut.dispatch(StopWatchAction.Stop)
+        assertEquals(9, sut.state.value)
+
+        sut.dispatch(StopWatchAction.Start)
+        scope.advanceTimeBy(MINIMUM_STOP_WATCH_DELAY + 1.milliseconds)
+        sut.dispatch(StopWatchAction.Stop)
+        assertEquals(10, sut.state.value)
+
+        scope.cancel()
+    }
+
+    @Test
+    fun `global state gets merged into controller`() {
+        val scope = TestScope(UnconfinedTestDispatcher())
+        val globalState = flow {
+            delay(250)
+            emit(42)
+            delay(250)
+            emit(42)
+        }
+
+        val sut = scope.createGlobalStateMergeController(globalState)
+
+        val states = sut.state.testIn(scope)
+
+        scope.advanceTimeBy(251)
+        sut.dispatch(1)
+        scope.advanceTimeBy(251)
+
+        assertEquals(
+            listOf(0, 42, 43, 85),
+            states
+        )
+        scope.cancel()
+    }
+
+    @Test
+    fun `MutatorContext is built correctly`() {
+        val stateAccessor = { 1 }
+        val actions = flowOf(1)
+        var emittedEffect: Int? = null
+        val sut = SubscriberAwareControllerImplementation.createMutatorContext<Int, Int, Int>(
+            stateAccessor,
+            actions
+        ) { emittedEffect = it }
+
+        sut.emitEffect(1)
+
+        assertEquals(stateAccessor(), sut.currentState)
+        assertEquals(actions, sut.actions)
+        assertEquals(1, emittedEffect)
+    }
+
+    @Test
+    fun `ReducerContext is built correctly`() {
+        var emittedEffect: Int? = null
+        val sut = SubscriberAwareControllerImplementation.createReducerContext<Int> { emittedEffect = it }
+        sut.emitEffect(2)
+        assertEquals(2, emittedEffect)
+    }
+
+    @Test
+    fun `TransformerContext is built correctly`() {
+        var emittedEffect: Int? = null
+        val sut = SubscriberAwareControllerImplementation.createTransformerContext<Int> { emittedEffect = it }
+        sut.emitEffect(3)
+        assertEquals(3, emittedEffect)
+    }
+
+    @Test
+    fun `cancelling the implementation will return the last state`() {
+        val scope = TestScope(UnconfinedTestDispatcher())
+        val sut = scope.createGlobalStateMergeController(emptyFlow())
+
+        val states = sut.state.testIn(scope)
+
+        sut.dispatch(0)
+        sut.dispatch(1)
+
+        scope.cancel()
+
+        sut.dispatch(2)
+
+        assertEquals(1, states.last())
+    }
+
+    @Test
+    fun `effects are received from mutator - reducer and transformer`() {
+        val scope = TestScope(UnconfinedTestDispatcher())
+        val sut = scope.createEffectTestController()
+        val states = sut.state.testIn(scope)
+        val effects = sut.effects.testIn(scope)
+
+        val testEmissions = listOf(
+            TestEffect.Reducer,
+            TestEffect.ActionTransformer,
+            TestEffect.MutationTransformer,
+            TestEffect.Mutator,
+            TestEffect.StateTransformer
+        )
+
+        testEmissions.map(TestEffect::ordinal).forEach(sut::dispatch)
+
+        assertEquals(
+            listOf(0) + testEmissions.map(TestEffect::ordinal),
+            states
+        )
+        assertEquals(testEmissions, effects)
+        scope.cancel()
+    }
+
+    @Test
+    fun `effects are only received once per collector`() {
+        val scope = TestScope(UnconfinedTestDispatcher())
+        val sut = scope.createEffectTestController()
+
+        sut.state.launchIn(scope)
+
+        val effects = mutableListOf<TestEffect>()
+        sut.effects.onEach { effects.add(it) }.launchIn(scope)
+        sut.effects.onEach { effects.add(it) }.launchIn(scope)
+
+        val testEmissions = listOf(
+            TestEffect.Reducer,
+            TestEffect.ActionTransformer,
+            TestEffect.MutationTransformer,
+            TestEffect.Reducer,
+            TestEffect.Mutator,
+            TestEffect.StateTransformer,
+            TestEffect.Reducer
+        )
+
+        testEmissions.map(TestEffect::ordinal).forEach(sut::dispatch)
+
+        assertEquals(testEmissions, effects)
+    }
+
+    @Test
+    fun `effects overflow throws error`() {
+        kotlin.runCatching {
+            runTest(UnconfinedTestDispatcher()) {
+                val sut = createEffectTestController()
+                sut.state.launchIn(this)
+                repeat(ControllerImplementation.CAPACITY + 1) { sut.dispatch(1) }
+            }
+        }.fold(
+            onSuccess = { error("this should not succeed") },
+            onFailure = { assertTrue(it.cause is ControllerError.Effect) }
+        )
+    }
+
+    @Test
+    fun `state is cancellable`() = runTest(UnconfinedTestDispatcher()) {
+        val sut = createCounterController()
+
+        sut.state.launchIn(this)
+        sut.dispatch(Unit)
+
+        var state: Int? = null
+        launch {
+            cancel()
+            state = -1
+            state = sut.state.first() // this should be cancelled and thus not return a value
+        }
+
+        assertEquals(-1, state)
+        cancel()
+    }
+
+    @Test
+    fun `effects are cancellable`() = runTest(UnconfinedTestDispatcher()) {
+        val sut = createEffectTestController()
+
+        sut.state.launchIn(this)
+        sut.dispatch(TestEffect.Mutator.ordinal)
+
+        var effect: TestEffect? = null
+        launch {
+            cancel()
+            effect = TestEffect.Reducer
+            effect = sut.effects.first() // this should be cancelled and thus not return a value
+        }
+
+        assertEquals(TestEffect.Reducer, effect)
+        cancel()
+    }
+
+    @Test
+    fun `controller is started lazily when only effects field is accessed`() {
+        val scope = TestScope(UnconfinedTestDispatcher())
+        val sut = ControllerImplementation<Int, Int, Int, String>(
+            scope = scope,
+            dispatcher = scope.defaultScopeDispatcher(),
+            controllerStart = ControllerStart.Lazy,
+            initialState = 0,
+            mutator = { action -> flowOf(action) },
+            reducer = { mutation, _ -> mutation },
+            actionsTransformer = { actions ->
+                merge(actions, flow {
+                    emitEffect("actionsTransformer started")
+                })
+            },
+            mutationsTransformer = { mutations -> mutations },
+            statesTransformer = { states -> states },
+            tag = "ImplementationTest.EffectController",
+            controllerLog = ControllerLog.None
+        )
+
+        val effects = sut.effects.testIn(scope)
+
+        assertEquals(
+            listOf("actionsTransformer started"),
+            effects
+        )
+
+        scope.cancel()
+    }
+
+
+    private fun CoroutineScope.createCounterController(
+        mutatorErrorIndex: Int? = null,
+        reducerErrorIndex: Int? = null
+    ) = SubscriberAwareControllerImplementation<Unit, Unit, Int, Nothing>(
+        scope = this,
+        dispatcher = defaultScopeDispatcher(),
+        initialState = 0,
+        mutator = { action ->
+            flow {
+                check(currentState != mutatorErrorIndex)
+                emit(action)
+            }
+        },
+        reducer = { _, previousState ->
+            check(previousState != reducerErrorIndex)
+            previousState + 1
+        },
+        actionsTransformer = { it },
+        mutationsTransformer = { it },
+        statesTransformer = { it },
+        tag = "ImplementationTest.CounterController",
+        controllerLog = ControllerLog.None,
+        sharingStarted = SharingStarted.WhileSubscribed()
+    )
+
+    private fun CoroutineScope.createAlwaysSameStateController() =
+        SubscriberAwareControllerImplementation<Unit, Unit, Int, Nothing>(
+            scope = this,
+            dispatcher = defaultScopeDispatcher(),
+            initialState = 0,
+            mutator = { flowOf(it) },
+            reducer = { _, previousState -> previousState },
+            actionsTransformer = { it },
+            mutationsTransformer = { it },
+            statesTransformer = { it },
+            tag = "ImplementationTest.AlwaysSameStateController",
+            controllerLog = ControllerLog.None,
+            sharingStarted = SharingStarted.WhileSubscribed()
+        )
+
+    private fun CoroutineScope.createOperationController() =
+        SubscriberAwareControllerImplementation<List<String>, List<String>, List<String>, Nothing>(
+            scope = this,
+            dispatcher = defaultScopeDispatcher(),
+
+            // 1. ["initialState"]
+            initialState = listOf("initialState"),
+
+            // 2. ["action"] + ["transformedAction"]
+            actionsTransformer = { actions ->
+                actions.map { it + "transformedAction" }
+            },
+
+            // 3. ["action", "transformedAction"] + ["mutation"]
+            mutator = { action ->
+                flowOf(action + "mutation")
+            },
+
+            // 4. ["action", "transformedAction", "mutation"] + ["transformedMutation"]
+            mutationsTransformer = { mutations ->
+                mutations.map { it + "transformedMutation" }
+            },
+
+            // 5. ["initialState"] + ["action", "transformedAction", "mutation", "transformedMutation"]
+            reducer = { mutation, previousState -> previousState + mutation },
+
+            // 6. ["initialState", "action", "transformedAction", "mutation", "transformedMutation"] + ["transformedState"]
+            statesTransformer = { states -> states.map { it + "transformedState" } },
+
+            tag = "ImplementationTest.OperationController",
+            controllerLog = ControllerLog.None,
+
+            sharingStarted = SharingStarted.WhileSubscribed()
+        )
+
+    private enum class TestEffect {
+        Mutator, Reducer, ActionTransformer, MutationTransformer, StateTransformer
+    }
+
+    private sealed interface StopWatchAction {
+        data object Start : StopWatchAction
+        data object Stop : StopWatchAction
+    }
+
+    private fun CoroutineScope.createStopWatchController() =
+        SubscriberAwareControllerImplementation<StopWatchAction, Int, Int, Nothing>(
+            scope = this,
+            dispatcher = defaultScopeDispatcher(),
+            initialState = 0,
+            mutator = { action ->
+                when (action) {
+                    is StopWatchAction.Start -> {
+                        flow {
+                            while (isActive) {
+                                delay(MINIMUM_STOP_WATCH_DELAY)
+                                emit(1)
+                            }
+                        }.takeUntil(actions.filterIsInstance<StopWatchAction.Stop>())
+                    }
+                    is StopWatchAction.Stop -> emptyFlow()
+                }
+            },
+            reducer = { mutation, previousState -> previousState + mutation },
+            actionsTransformer = { it },
+            mutationsTransformer = { it },
+            statesTransformer = { it },
+            tag = "ImplementationTest.StopWatchController",
+            controllerLog = ControllerLog.None,
+            SharingStarted.WhileSubscribed()
+        )
+
+    private fun CoroutineScope.createGlobalStateMergeController(
+        globalState: Flow<Int>
+    ) = SubscriberAwareControllerImplementation<Int, Int, Int, Nothing>(
+        scope = this,
+        dispatcher = defaultScopeDispatcher(),
+        initialState = 0,
+        mutator = { flowOf(it) },
+        reducer = { action, previousState -> previousState + action },
+        actionsTransformer = { merge(it, globalState) },
+        mutationsTransformer = { it },
+        statesTransformer = { it },
+        tag = "ImplementationTest.GlobalStateMergeController",
+        controllerLog = ControllerLog.None,
+        sharingStarted = SharingStarted.WhileSubscribed()
+    )
+
+    private fun CoroutineScope.createEffectTestController() =
+        SubscriberAwareControllerImplementation<Int, Int, Int, TestEffect>(
+            scope = this,
+            dispatcher = defaultScopeDispatcher(),
+            initialState = 0,
+            mutator = { action ->
+                if (action == TestEffect.Mutator.ordinal) emitEffect(TestEffect.Mutator)
+                flowOf(action)
+            },
+            reducer = { mutation, _ ->
+                if (mutation == TestEffect.Reducer.ordinal) emitEffect(TestEffect.Reducer)
+                mutation
+            },
+            actionsTransformer = { actions ->
+                actions.onEach {
+                    if (it == TestEffect.ActionTransformer.ordinal) {
+                        emitEffect(TestEffect.ActionTransformer)
+                    }
+                }
+            },
+            mutationsTransformer = { mutations ->
+                mutations.onEach {
+                    if (it == TestEffect.MutationTransformer.ordinal) {
+                        emitEffect(TestEffect.MutationTransformer)
+                    }
+                }
+            },
+            statesTransformer = { states ->
+                states.onEach {
+                    if (it == TestEffect.StateTransformer.ordinal) {
+                        emitEffect(TestEffect.StateTransformer)
+                    }
+                }
+            },
+            tag = "ImplementationTest.EffectController",
+            controllerLog = ControllerLog.None,
+            sharingStarted = SharingStarted.WhileSubscribed()
+        )
+
+    companion object {
+        private  val MINIMUM_STOP_WATCH_DELAY = 1.seconds
+    }
+}
\ No newline at end of file
diff --git a/control-core/src/commonTest/kotlin/at/florianschuster/control/SubscriberAwareControllerTest.kt b/control-core/src/commonTest/kotlin/at/florianschuster/control/SubscriberAwareControllerTest.kt
new file mode 100644
index 00000000..974b2634
--- /dev/null
+++ b/control-core/src/commonTest/kotlin/at/florianschuster/control/SubscriberAwareControllerTest.kt
@@ -0,0 +1,110 @@
+package at.florianschuster.control
+
+import kotlinx.coroutines.cancelChildren
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.SharingStarted
+import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.flow.single
+import kotlinx.coroutines.flow.singleOrNull
+import kotlinx.coroutines.test.runTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+
+@Suppress("UNCHECKED_CAST")
+internal class SubscriberAwareControllerTest {
+    @Test
+    fun `controller builder`() = runTest {
+        val expectedInitialState = 42
+        val sut = createController<Int, Int, Int>(
+            initialState = expectedInitialState,
+            sharingStarted = SharingStarted.Lazily
+        ) as SubscriberAwareControllerImplementation<Int, Int, Int, Nothing>
+
+        assertEquals(this, sut.scope)
+        assertEquals(expectedInitialState, sut.initialState)
+
+        val mutatorContext = object : EffectMutatorContext<Int, Int, Nothing> {
+            override val currentState: Int
+                get() = notImplemented()
+            override val actions: Flow<Int>
+                get() = notImplemented()
+
+            override fun emitEffect(effect: Nothing) {
+                notImplemented()
+            }
+        }
+        assertEquals(null, sut.mutator(mutatorContext, 3).singleOrNull())
+
+        val reducerContext = object : EffectReducerContext<Nothing> {
+            override fun emitEffect(effect: Nothing) {
+                notImplemented()
+            }
+        }
+        assertEquals(1, sut.reducer(reducerContext, 0, 1))
+
+        val transformerContext = object : EffectTransformerContext<Nothing> {
+            override fun emitEffect(effect: Nothing) {
+                notImplemented()
+            }
+        }
+        assertEquals(1, sut.actionsTransformer(transformerContext, flowOf(1)).single())
+        assertEquals(2, sut.mutationsTransformer(transformerContext, flowOf(2)).single())
+        assertEquals(3, sut.statesTransformer(transformerContext, flowOf(3)).single())
+
+        assertEquals(defaultControllerTag(), sut.tag)
+        assertEquals(ControllerLog.None, sut.controllerLog)
+
+        assertEquals(SharingStarted.Lazily, sut.sharingStarted)
+        assertEquals(defaultScopeDispatcher(), sut.dispatcher)
+
+        coroutineContext.cancelChildren()
+    }
+
+    @Test
+    fun `effect controller builder`() = runTest {
+        val expectedInitialState = 42
+        val sut = createEffectController<Int, Int, Int, Int>(
+            initialState = expectedInitialState,
+            sharingStarted = SharingStarted.Lazily
+        ) as SubscriberAwareControllerImplementation<Int, Int, Int, Int>
+
+        assertEquals(this, sut.scope)
+        assertEquals(expectedInitialState, sut.initialState)
+
+        val mutatorContext = object : EffectMutatorContext<Int, Int, Int> {
+            override val currentState: Int
+                get() = notImplemented()
+            override val actions: Flow<Int>
+                get() = notImplemented()
+
+            override fun emitEffect(effect: Int) {
+                notImplemented()
+            }
+        }
+        assertEquals(null, sut.mutator(mutatorContext, 3).singleOrNull())
+
+        val reducerContext = object : EffectReducerContext<Int> {
+            override fun emitEffect(effect: Int) {
+                notImplemented()
+            }
+        }
+        assertEquals(1, sut.reducer(reducerContext, 0, 1))
+
+        val transformerContext = object : EffectTransformerContext<Int> {
+            override fun emitEffect(effect: Int) {
+                notImplemented()
+            }
+        }
+        assertEquals(1, sut.actionsTransformer(transformerContext, flowOf(1)).single())
+        assertEquals(2, sut.mutationsTransformer(transformerContext, flowOf(2)).single())
+        assertEquals(3, sut.statesTransformer(transformerContext, flowOf(3)).single())
+
+        assertEquals(defaultControllerTag(), sut.tag)
+        assertEquals(ControllerLog.None, sut.controllerLog)
+
+        assertEquals(SharingStarted.Lazily, sut.sharingStarted)
+        assertEquals(defaultScopeDispatcher(), sut.dispatcher)
+
+        coroutineContext.cancelChildren()
+    }
+}
\ No newline at end of file
diff --git a/examples/android-counter/src/main/AndroidManifest.xml b/examples/android-counter/src/main/AndroidManifest.xml
index 29c3c26b..12cd24ee 100644
--- a/examples/android-counter/src/main/AndroidManifest.xml
+++ b/examples/android-counter/src/main/AndroidManifest.xml
@@ -21,5 +21,6 @@
             </intent-filter>
         </activity>
 
+        <activity android:name=".EmptyActivity"/>
     </application>
 </manifest>
diff --git a/examples/android-counter/src/main/kotlin/at/florianschuster/control/counter/CounterScreen.kt b/examples/android-counter/src/main/kotlin/at/florianschuster/control/counter/CounterScreen.kt
index 22567b63..179d6e56 100644
--- a/examples/android-counter/src/main/kotlin/at/florianschuster/control/counter/CounterScreen.kt
+++ b/examples/android-counter/src/main/kotlin/at/florianschuster/control/counter/CounterScreen.kt
@@ -15,7 +15,6 @@ import androidx.compose.material3.CircularProgressIndicator
 import androidx.compose.material3.MaterialTheme
 import androidx.compose.material3.Text
 import androidx.compose.runtime.Composable
-import androidx.compose.runtime.collectAsState
 import androidx.compose.runtime.getValue
 import androidx.compose.runtime.remember
 import androidx.compose.runtime.rememberCoroutineScope
@@ -27,21 +26,30 @@ import androidx.compose.ui.semantics.contentDescription
 import androidx.compose.ui.semantics.semantics
 import androidx.compose.ui.tooling.preview.Preview
 import androidx.compose.ui.unit.dp
+import androidx.lifecycle.Lifecycle
+import androidx.lifecycle.compose.collectAsStateWithLifecycle
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.flow.SharingStarted
 
 @Composable
 internal fun CounterScreen(
     scope: CoroutineScope = rememberCoroutineScope(),
-    controller: CounterController = remember(scope) { scope.createCounterController() }
+    controller: CounterController = remember(scope) { scope.createCounterController(sharingStarted = SharingStarted.WhileSubscribedOrRetained) },
+    showNext: () -> Unit
 ) {
-    val state by controller.state.collectAsState()
-    CounterView(state = state, dispatch = controller::dispatch)
+    val state by controller.state.collectAsStateWithLifecycle(minActiveState = Lifecycle.State.RESUMED)
+    CounterView(
+        state = state,
+        dispatch = controller::dispatch,
+        showNext = showNext
+    )
 }
 
 @Composable
 private fun CounterView(
     state: CounterState,
-    dispatch: (CounterAction) -> Unit = {}
+    dispatch: (CounterAction) -> Unit = {},
+    showNext: () -> Unit
 ) {
     Box(
         modifier = Modifier
@@ -75,6 +83,7 @@ private fun CounterView(
                 onClick = { dispatch(CounterAction.Increment) },
             ) { Text("+") }
         }
+        Button(onClick = showNext) { Text("Show other activity") }
         if (state.loading) {
             CircularProgressIndicator(
                 modifier = Modifier
@@ -92,7 +101,8 @@ private fun CounterView(
 private fun Preview() {
     MaterialTheme {
         CounterView(
-            state = CounterState(value = 1, loading = false)
+            state = CounterState(value = 1, loading = false),
+            showNext = {}
         )
     }
 }
@@ -102,7 +112,8 @@ private fun Preview() {
 private fun Preview_Loading() {
     MaterialTheme {
         CounterView(
-            state = CounterState(value = 2, loading = true)
+            state = CounterState(value = 2, loading = true),
+            showNext = {}
         )
     }
 }
diff --git a/examples/android-counter/src/main/kotlin/at/florianschuster/control/counter/EmptyActivity.kt b/examples/android-counter/src/main/kotlin/at/florianschuster/control/counter/EmptyActivity.kt
new file mode 100644
index 00000000..d63b3a7f
--- /dev/null
+++ b/examples/android-counter/src/main/kotlin/at/florianschuster/control/counter/EmptyActivity.kt
@@ -0,0 +1,32 @@
+package at.florianschuster.control.counter
+
+import android.os.Bundle
+import androidx.activity.ComponentActivity
+import androidx.activity.compose.setContent
+import androidx.activity.enableEdgeToEdge
+import androidx.compose.foundation.layout.Box
+import androidx.compose.foundation.layout.fillMaxSize
+import androidx.compose.material3.MaterialTheme
+import androidx.compose.material3.Text
+import androidx.compose.runtime.Composable
+import androidx.compose.ui.Modifier
+
+internal class EmptyActivity: ComponentActivity() {
+
+    override fun onCreate(savedInstanceState: Bundle?) {
+        enableEdgeToEdge()
+        super.onCreate(savedInstanceState)
+        setContent {
+            MaterialTheme {
+                SomeScreen()
+            }
+        }
+    }
+}
+
+@Composable
+internal fun SomeScreen() {
+    Box(modifier = Modifier.fillMaxSize()) {
+        Text("Hello World")
+    }
+}
\ No newline at end of file
diff --git a/examples/android-counter/src/main/kotlin/at/florianschuster/control/counter/MainActivity.kt b/examples/android-counter/src/main/kotlin/at/florianschuster/control/counter/MainActivity.kt
index ab7d99a2..61d02f66 100644
--- a/examples/android-counter/src/main/kotlin/at/florianschuster/control/counter/MainActivity.kt
+++ b/examples/android-counter/src/main/kotlin/at/florianschuster/control/counter/MainActivity.kt
@@ -1,5 +1,6 @@
 package at.florianschuster.control.counter
 
+import android.content.Intent
 import android.os.Bundle
 import androidx.activity.ComponentActivity
 import androidx.activity.compose.setContent
@@ -13,7 +14,9 @@ internal class MainActivity : ComponentActivity() {
         super.onCreate(savedInstanceState)
         setContent {
             MaterialTheme {
-                CounterScreen()
+                CounterScreen {
+                    startActivity(Intent(this, EmptyActivity::class.java))
+                }
             }
         }
     }
diff --git a/examples/android-counter/src/main/kotlin/at/florianschuster/control/counter/WhileSubscribedOrRetained.kt b/examples/android-counter/src/main/kotlin/at/florianschuster/control/counter/WhileSubscribedOrRetained.kt
new file mode 100644
index 00000000..8018997e
--- /dev/null
+++ b/examples/android-counter/src/main/kotlin/at/florianschuster/control/counter/WhileSubscribedOrRetained.kt
@@ -0,0 +1,56 @@
+package at.florianschuster.control.counter
+
+import android.os.Handler
+import android.os.Looper
+import android.view.Choreographer
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.SharingCommand
+import kotlinx.coroutines.flow.SharingStarted
+import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.flow.dropWhile
+import kotlinx.coroutines.flow.transformLatest
+
+/**
+ * See: https://blog.p-y.wtf/whilesubscribed5000
+ *
+ * Problem:
+ * Config changes leading to resubscription
+ *
+ * Solution:
+ * Googles way is using WhileSubscribed(5_000), this guy in the article came up with this
+ * clever solution. Using it, the anti-pattern of using a magic number can be avoided.
+ */
+private data object WhileSubscribedOrRetainedImpl : SharingStarted {
+
+    private val handler = Handler(Looper.getMainLooper())
+
+    @OptIn(ExperimentalCoroutinesApi::class)
+    override fun command(subscriptionCount: StateFlow<Int>): Flow<SharingCommand> = subscriptionCount
+        .transformLatest { count ->
+            if (count > 0) {
+                emit(SharingCommand.START)
+            } else {
+                val posted = CompletableDeferred<Unit>()
+                // This code is perfect. Do not change a thing.
+                Choreographer.getInstance().postFrameCallback {
+                    handler.postAtFrontOfQueue {
+                        handler.post {
+                            posted.complete(Unit)
+                        }
+                    }
+                }
+                posted.await()
+                emit(SharingCommand.STOP)
+            }
+        }
+        .dropWhile { it != SharingCommand.START }
+        .distinctUntilChanged()
+
+    override fun toString(): String = "SharingStarted.WhileSubscribedOrRetained"
+}
+
+val SharingStarted.Companion.WhileSubscribedOrRetained: SharingStarted
+    get() = WhileSubscribedOrRetainedImpl
\ No newline at end of file
diff --git a/examples/kotlin-counter/src/main/kotlin/at/florianschuster/control/counter/CounterController.kt b/examples/kotlin-counter/src/main/kotlin/at/florianschuster/control/counter/CounterController.kt
index d90f3138..c0893c5f 100644
--- a/examples/kotlin-counter/src/main/kotlin/at/florianschuster/control/counter/CounterController.kt
+++ b/examples/kotlin-counter/src/main/kotlin/at/florianschuster/control/counter/CounterController.kt
@@ -1,11 +1,14 @@
 package at.florianschuster.control.counter
 
-import at.florianschuster.control.ControllerLog
 import at.florianschuster.control.Controller
+import at.florianschuster.control.ControllerLog
 import at.florianschuster.control.createController
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.merge
 import kotlin.time.Duration.Companion.milliseconds
 
 typealias CounterController = Controller<CounterAction, CounterState>
@@ -24,6 +27,7 @@ sealed interface CounterAction {
 private sealed interface CounterMutation {
     data object IncreaseValue : CounterMutation
     data object DecreaseValue : CounterMutation
+    data class UpdateTimeStamp(val time: Long): CounterMutation
     data class SetLoading(val loading: Boolean) : CounterMutation
 }
 
@@ -39,12 +43,25 @@ data class CounterState(
  * creates a [CounterController] from the [CoroutineScope].
  */
 fun CoroutineScope.createCounterController(
-    initialValue: Int = 0
-): CounterController = createController(
-
+    initialValue: Int = 0,
+    sharingStarted: SharingStarted
+): CounterController = createController<CounterAction, CounterMutation, CounterState>(
     // we start with the initial state
     initialState = CounterState(value = initialValue, loading = false),
-
+    mutationsTransformer = { mutations ->
+        merge(
+            mutations,
+            flow {
+                while (true) {
+                    delay(1_000)
+                    emit(1)
+                }
+            }.map {
+                println("TestController  Tick")
+                CounterMutation.UpdateTimeStamp(System.currentTimeMillis())
+            }
+        )
+    },
     // every action is transformed into [0..n] mutations
     mutator = { action ->
         when (action) {
@@ -71,9 +88,11 @@ fun CoroutineScope.createCounterController(
             is CounterMutation.IncreaseValue -> previousState.copy(value = previousState.value + 1)
             is CounterMutation.DecreaseValue -> previousState.copy(value = previousState.value - 1)
             is CounterMutation.SetLoading -> previousState.copy(loading = mutation.loading)
+            is CounterMutation.UpdateTimeStamp -> previousState
         }
     },
 
     // logs to println
-    controllerLog = ControllerLog.Println
-)
+    controllerLog = ControllerLog.Println,
+    sharingStarted = sharingStarted
+)
\ No newline at end of file
diff --git a/examples/kotlin-counter/src/main/kotlin/at/florianschuster/control/counter/counter.kt b/examples/kotlin-counter/src/main/kotlin/at/florianschuster/control/counter/counter.kt
index c348986c..328a161f 100644
--- a/examples/kotlin-counter/src/main/kotlin/at/florianschuster/control/counter/counter.kt
+++ b/examples/kotlin-counter/src/main/kotlin/at/florianschuster/control/counter/counter.kt
@@ -2,6 +2,7 @@ package at.florianschuster.control.counter
 
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.flow.SharingStarted
 import kotlin.system.exitProcess
 
 private const val AvailableCommands = "available commands -> + , - , exit"
@@ -14,7 +15,7 @@ internal fun main(args : Array<String>) {
     println("🎛 <control-counter>")
     println("$AvailableCommands\n")
 
-    val controller = CoroutineScope(Dispatchers.Unconfined).createCounterController()
+    val controller = CoroutineScope(Dispatchers.Unconfined).createCounterController(sharingStarted = SharingStarted.WhileSubscribed(0))
 
     while (true) {
         when (readlnOrNull()) {