   }
 
   public func createStateStream() -> Observable<State> {
-    let state = self.transform(action: self.actionSubject)
+    let action = self.actionSubject.asObservable()
+    let transformedAction = self.transform(action: action)
+    let mutation = transformedAction
       .flatMap { [weak self] action -> Observable<Mutation> in
         guard let `self` = self else { return .empty() }
         return self.mutate(action: action)
       }
+    let transformedMutation = self.transform(mutation: mutation)
+    let state = transformedMutation
       .scan(self.initialState) { [weak self] state, mutation -> State in
         guard let `self` = self else { return state }
         return self.reduce(state: state, mutation: mutation)

    return .empty()
  }

  public func reduce(state: State, mutation: Mutation) -> State {
    return state
  }

  /// Commits mutation from the action. This is the best place to perform side-effects such as
  /// async tasks.
  func mutate(action: Action) -> Observable<Mutation>

  /// Transforms the mutation stream. Implement this method to transform or combine with other
  /// observables. This method is called once before the state stream is created.
  func transform(mutation: Observable<Mutation>) -> Observable<Mutation>

  /// Generates a new state with the previous state and the action. It should be purely functional
  /// so it should not perform any side-effects here. This method is called every time when the
