RxMiddleware for Cachopo (A ReSwift wrapper with RxSwift)

RxMiddleware.swift

import Foundation
import RxSwift
import ReSwift

class RxMiddleware<State:StateType, Environment:EnvironmentType>
{
    let context:ContextType<State, Environment>

    var disposeBag = DisposeBag()
    
    init(context:ContextType<State, Environment>) {
        self.context = context
    }
    
    func middleware() -> Middleware
    {
        return { dispatch, getState in
            
            return { next in
                
                return { action in
                    
                    guard let rxaction = action as? RxStore<State, Environment>.RxAction else {
                        next(action)
                        return Observable<Action>.empty()
                    }
                    
                    
                    let subject = PublishSubject<Action>()
                    
                    rxaction.run(self.context)
                        .subscribe(
                            onNext: { action in
                                next(action)
                                subject.onNext(action)
                            },
                            onError: { error in
                                next(ErrorAction(error: error))
                                subject.onError(error)
                            },
                            onCompleted: {
                                subject.onCompleted()
                        })
                        .addDisposableTo(self.disposeBag)
                    
                    return subject
                        .observeOn(MainScheduler.instance)
                        .asObservable()
                }
            }
        }
    }
}