RxSwift 清理序列

Question

我对 Rx 的性质有点问题。正如我正确理解的那样，被推送到序列的每个值都被持有而不是被释放。 我想问一下是否有办法将这些值从序列中推开或释放。

编辑：

让 progressSessionValue = PublishSubject() 每秒调用 10 次，这会产生大量内存消耗。这种情况下如何释放内存？

selectedValue = Driver
        .merge(
            // you either use slider
            sliderChangedValue
                .withLatestFrom(progressTypePressed){ ($0, $1) }
                .map{ Double($0).stringUnitValue(from: $1) }
                .asDriver(),
            // or take value from the picker
            mainPickerValue
                .withLatestFrom(progressTypePressed){ ($0, $1) }
                .map{ Double($0.value).stringUnitValue(from: $1) }
                .asDriver(),
            // or use stored/calculated values
             progressSessionValue.asDriver()
                .withLatestFrom(mainPressCombined)
                .map{ type, gen, segment in audio.retrieveValue(gen, type: type, segment: segment) }
        )
        .distinctUntilChanged()

编辑2：

class BMMain_VM {
/// in
let didAppear = PublishSubject<Void>()
let didLoad = PublishSubject<Void>()
// pick up all generators
let genButtonPressed = PublishSubject<AudioGenerator.ItemEnum>()
let progressionType = PublishSubject<BM.ProgressionType>()
let selectedGeneratorType = PublishSubject<BM.GeneratorType>()

let manageSoundAction = PublishSubject<BMMain.SoundAction>()
//
let sessionType = PublishSubject<AudioGenerator.SessionType>()
//
let switchControlPanel = PublishSubject<Void>()
// slider changes
let sliderChangedValue = PublishSubject<Float>()
// picker changes
let pickerSelectedValues = PublishSubject<[Int]>()
// navigation
let infoButtonPressed = PublishSubject<Void>()
let toneButtonPressed = PublishSubject<Void>()
let menuButtonPressed = PublishSubject<Void>()
let navigateRoute = PublishSubject<BM.ProgressionType>()
//
let storeIsochronicType = PublishSubject<BM.IsoGeneratorType>()
/////////////////////////////////////////////////////////////////////
/// out
// selected value to be presented
let selectedValue: Driver<String>
// values to be displayed in the main picker
let pickerValues: Driver<PickerViewAdapter.Element>
// select main picker
let selectPickerValue: Driver<[(row:Int, component:Int)]>
// select .static or .dynamic
let selectSessionType: Driver<Int>
// slider value
let selectSliderValue: Driver<Float>
// set slider dimensions
let sliderValues: Driver<ClosedRange<Float>>
//
let depthButtonHidden: Driver<Bool>
//
let manageButtonState: Driver<(AudioGenerator.ItemEnum, Bool)>
let manageTimerMaxLabel: Driver<String>
let manageTimerMinLabel: Driver<String>
let manageTimerProgress: Driver<Float>
//
let changeLayout: Driver<AudioGenerator.ItemEnum>
// let initAnimations: Driver<(BM.ProgressionType, Float)>
let updateButtonsState: Driver<Void>
//
let mainPressCombined: Driver<(BM.ProgressionType, AudioGenerator.ItemEnum, AudioGenerator.SessionType)>
//
private let disposeBag = DisposeBag()
/////////////////////////////////////////////////////////////////////
//
init(context: BMMain.Context, audio: AudioGenerator, dataService: BMDBService) {
    
    let isRunning = audio.rx.isSessionPlaying.distinctUntilChanged()
    
    let sessionTimerValue = BehaviorRelay(value: dataService.getTimer())
    
    let progressTypePressed = progressionType.startWith(.amplitude).asDriver()
    let generatorPressed = genButtonPressed.startWith(.A).asDriver()
    let sessionTypeChange = sessionType.startWith(.static).asDriver()
    //
    mainPressCombined = Driver
        .combineLatest(progressTypePressed, generatorPressed, sessionTypeChange)
    
    // manage notifications
    NotificationCenter.default.rx
        .notification(NSNotification.Name(rawValue: "changeGenerator"))
        .filter{ $0.object is AudioGenerator.ItemEnum }
        .map{ $0.object as! AudioGenerator.ItemEnum }
        .bind(to: genButtonPressed)
        .disposed(by: disposeBag)
    // set note value
    let updateProgressValueAction = NotificationCenter.default.rx
        .notification(NSNotification.Name(rawValue: "updateMain"))
    //
    changeLayout = generatorPressed.asDriver()
    //
    toneButtonPressed
        .withLatestFrom(generatorPressed)
        .bind(to: context.navigateNoteView)
        .disposed(by: disposeBag)
    // auto switch
    toneButtonPressed
        .map{ _ in .frequency }
        .bind(to: progressionType)
        .disposed(by: disposeBag)
    //
    infoButtonPressed
        .bind(to: context.navigateInfoView)
        .disposed(by: disposeBag)
    
    menuButtonPressed
        .withLatestFrom(generatorPressed)
        .bind(to: context.navigateMenuView)
        .disposed(by: disposeBag)
    
    navigateRoute
        .withLatestFrom(generatorPressed){ ($0, $1) }
        .bind { (arg) in let (route, gen) = arg
            switch route {
            case .binaural:
                context.navigateBinauralView.onNext(gen)
            case .frequency:
                context.navigateFrequencyView.onNext(gen)
            case .amplitude:
                context.navigateVolumeView.onNext(gen)
            case .depth:
                context.navigateDepthView.onNext(gen)
            }
        }.disposed(by: disposeBag)
    //
    pickerValues = progressTypePressed.map{ $0.pickerTitles }
    sliderValues = progressTypePressed.map{ $0.interval }
    //
    sessionTypeChange
        .drive(onNext: { type in
            audio.reset(type)
            audio.manageGeneral(type) })
        .disposed(by: disposeBag)
    //
    let mainPickerValue = pickerSelectedValues
        .withLatestFrom(progressTypePressed){ ($0, $1) }
        .map{ selection, type in type.consolidate(value: selection) }
    //
    let progressSessionValue = PublishSubject<Void>()
    //
    let changeValueAction = Driver
        .merge(updateProgressValueAction.map { _ in }.asDriver(),
               didLoad.map{ _ in }.asDriver(),
               progressSessionValue.map{ _ in }.asDriver(),
               generatorPressed.map{ _ in },
               progressTypePressed.map{ _ in },
               sessionTypeChange.map{ _ in }.asDriver(),
               switchControlPanel.map{ _ in }.asDriver()
        )
    //
    selectedValue = Driver
        .merge(
            // you either use slider
            sliderChangedValue
                .withLatestFrom(progressTypePressed){ ($0, $1) }
                .map{ Double($0).stringUnitValue(from: $1) }
                .asDriver(),
            // or take value from the picker
            mainPickerValue
                .withLatestFrom(progressTypePressed){ ($0, $1) }
                .map{ Double($0.value).stringUnitValue(from: $1) }
                .asDriver(),
            // or use stored/calculated values
            changeValueAction
                .withLatestFrom(mainPressCombined)
                .map{ type, gen, segment in audio.retrieveValue(gen, type: type, segment: segment) }
        )
        .distinctUntilChanged()
        .debug()
    // two ways how to manage values
    Observable.merge(sliderChangedValue,
                     mainPickerValue.map{ $0.value })
        .withLatestFrom(mainPressCombined){ ($0, $1) }
        .bind{ (arg) in let (value, type) = arg
            dataService.store(value, type: type.0, generator: type.1)
            audio.manage(value, gen: type.1, type: type.0) }
        .disposed(by: disposeBag)
    // assign picker
    selectPickerValue = Driver
        .merge(
            changeValueAction
                .withLatestFrom(mainPressCombined)
                .map{ (type, gen, session) in type
                    .select(value: audio.retrieve(gen, type: type, segment: session)) },
            mainPickerValue
                .map{ selection, value in selection }
                .unwrap()
                .asDriver()
        )
    // assign slider - do not know, why it needs to be twice when type pressed
    selectSliderValue = Driver
        .merge(
            changeValueAction.map{ _ in return () },
            progressTypePressed.map{ _ in return () }
        )
        .withLatestFrom(mainPressCombined)
        .map{ (type, gen, session) in audio.retrieveValue(gen, type: type, segment: session) }
         //.debug("Slider value: ")
    // assign audio type
    selectSessionType = Driver
        .merge(updateProgressValueAction.map{ _ in return () }.asDriver(),
               generatorPressed.map{ _ in return () },
               sessionTypeChange.map{ _ in return () },
               context.showPurchaseAlert.map{ _ in return () }.as????())
        .withLatestFrom(mainPressCombined)
        .map{ _, gen, session in gen.getStoredType(session).rawValue }
        // .distinctUntilChanged()
        .debug("generator type: ")
    // update audio type
    selectedGeneratorType
        .withLatestFrom(sessionTypeChange){ ($0, $1) }
        //.filter{ isPurchased || $0.1 == .dynamic }
        .withLatestFrom(generatorPressed){ ($0, $1) }
        .bind{ type, gen in
            dataService.change(type.0, gen: gen)
            audio.manageAudioType(type.0, gen: gen) }
        .disposed(by: disposeBag)
    //
    depthButtonHidden = Driver.merge(updateProgressValueAction.map{ _ in }.asDriver(),
                                     generatorPressed.map{ _ in },
                                     sessionTypeChange.map{ _ in },
                                     selectedGeneratorType.map{ _ in }.asDriver())
        .withLatestFrom(mainPressCombined)
        .map{ $1.getStoredType($2) }.debug()
        .map{ type in type != .isochronic }
        .distinctUntilChanged()
        .asDriver()
    // autoswitch to amplitude when depth is not visible
    depthButtonHidden
        .withLatestFrom(mainPressCombined){ ($0, $1) }
        .filter{ $0 && $1.0 == .depth }
        .map{ _ in  BM.ProgressionType.amplitude }
        .drive(progressionType)
        .disposed(by: disposeBag)
    // audio
    storeIsochronicType
        .withLatestFrom(generatorPressed){ ($0, $1) }
        .map(audio.changeIsochronic)
        .subscribe()
        .disposed(by: disposeBag)
    // db
    storeIsochronicType
        .withLatestFrom(mainPressCombined){ ($0, $1) }
        .map{ ($0, $1.1, $1.2) }
        .map(dataService.updateIsochronic)
        .subscribe()
        .disposed(by: disposeBag)
    ////////////////////////////////////////////////////////////////////////////////////////////////////////
    // Timers
    let sessionTimerStopAction = PublishSubject<BMMain.SoundAction>()
    // start session
    let progressTimerValue = isRunning
        .debug("isRunning")
        .flatMapLatest { isRunning  in
            isRunning
                ? Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance)
                : .empty() }
        .enumerated()
        .flatMap { (arg) -> Observable<Int> in let ( _, int ) = arg
            return Observable.just(int + 1) }
        .debug("timer")
        .startWith(0)
        .share()
    // stop session
    progressTimerValue
        .withLatestFrom(sessionTimerValue){ ($0, $1) }
        .filter{ $0 == $1 }
        .map{ _ in .stop }
        .bind(to: sessionTimerStopAction)
        .disposed(by: disposeBag)
    
    updateButtonsState = Observable.merge(sessionTimerStopAction.map{_ in },
                                          sessionType.map{_ in }).as????()
    //
    let switchSegment = PublishSubject<Bool>()
    switchSegment.onNext(true)
    let timerAction = Observable.combineLatest(switchSegment, sessionType)
    //
    isRunning
        .withLatestFrom(timerAction){ ($0, $1) }
        .filter{ state, type in type.1 == .dynamic || type.0 }
        .flatMapLatest{ data -> Observable<Int> in
            if !data.0 && data.1.1 == .dynamic {
                audio.reset(.dynamic)
            }
            return data.0
                ? Observable<Int>.interval(.milliseconds(100), scheduler: MainScheduler.instance)
                : .empty() }
        //.debug("Progressive timer: ")
        .map{ _ in audio.calculateFrequency() }
        .bind(to: progressSessionValue)
        .disposed(by: disposeBag)
    
    // Progressive session recalculation timer
    isRunning
        .debug("Start recalculate")
        .withLatestFrom(timerAction){ ($0, $1) }
        .filter{ state, type in (type.1 == .dynamic || type.0) && state }
        .bind{ _ in audio.recalculateStep() }
        .disposed(by: disposeBag)
    //
    isRunning
        .withLatestFrom(timerAction){ ($0, $1) }
        .filter{ state, type in type.1 == .dynamic || type.0 }
        .flatMapLatest { object in
            object.0
                ? Observable<Int>.interval(.seconds(Timer.stageTime), scheduler: MainScheduler.instance)
                : .empty() }
        .map{ _ in audio.recalculateStep() }.debug("Recalculate")
        .subscribe()
        .disposed(by: disposeBag)
    //
    sessionType
        .bind {
            sessionTimerStopAction.onNext(.stop)
            switchSegment.onNext($0 == .dynamic)
            dataService.updateType($0) }
        .disposed(by: disposeBag)
    
    dataService.timerChanged()
        .bind(to: sessionTimerValue)
        .disposed(by: disposeBag)
    //
    let setTimer = Observable.merge(sessionTimerValue.asObservable(), progressTimerValue)
    //
    manageTimerMinLabel = setTimer
        .map{ $0.timeString }
        .asDriver()
    
    manageTimerMaxLabel = setTimer
        .withLatestFrom(sessionTimerValue){ ($0, $1) }
        .map{ ($1 - Int($0)).timeString }
        .asDriver()
    //
    manageTimerProgress = setTimer
        .withLatestFrom(sessionTimerValue){ ($0, $1) }
        .map{ Float($0)/Float($1) }
        .asDriver()
    //
    manageButtonState = Driver
        .merge(Driver
            .combineLatest(manageSoundAction.asDriver(), generatorPressed)
            .map{ _ in .manage },
               sessionTimerStopAction.asDriver())
        .withLatestFrom(generatorPressed){ ($0, $1) }
        .map{ ($1, audio.buttonState($1)) }
    //
    Observable.merge(sessionTimerStopAction, manageSoundAction)
        .withLatestFrom(generatorPressed){ ($0, $1) }
        .bind{ reason, gen in
            switch reason {
            case .manage: audio.manageGenerators()
            case .switched: audio.switchGenerator(gen)
            case .stop: audio.stopAllGenerators()
            }
    }.disposed(by: disposeBag)
}

}

Answer 1

一个 Observable，默认情况下不保存值。它拥有一个价值生产者函数，每次订阅它时都会调用该函数。 Observable 本身永远不会改变状态，它只是在其生产者创建值时发出值。

因此，发布没有任何价值，您的问题也没有太大意义。也许如果您解释了您遇到的实际问题，我们也许可以帮助您。