Не може да се отпише от Rx

Rx използва интерфейса IDisposable, за да ви позволи да се отпишете от наблюдаван абонамент, който все още не е приключил естествено. Ако вашето наблюдаемо изпрати OnCompleted или OnError известие, тогава абонаментът автоматично се отменя за вас.

Едно ясно предимство на този подход е, че можете да създадете един CompositeDisposable, за да обедините всичките си абонаменти, за да активирате едно отписване. Това е много по-добре от смесица от отделяния, необходими за премахване на манипулатори на събития.

Във вашия код вие не прекратявате абонамента, така че за всяко кликване върху button_go вие създавате нов абонамент.

Има четири решения, които можете да използвате, всяко от които е различно.

(1)

Не забравяйте, че ключът към необходимостта да се обадите на .Dispose() е, ако вашият наблюдаем абонамент не е приключил естествено и искате той да приключи. Така че можете просто да добавите .Take(1) към вашата заявка, за да я накарате да завърши естествено, след като бъде произведена една стойност.

private void button_go_Click(object sender, RoutedEventArgs e)
{
    var uiContext = SynchronizationContext.Current;
    var results = Observable.FromEventPattern<TestResultHandler, TestResultArgs>(
        handler => (s, a) => handler(s, a),
        handler => this.myTest.Results += handler,
        handler => this.myTest.Results -= handler)
        .Take(1)
        .ObserveOn(uiContext)
        .Subscribe(x => this.richTextBox_testResults.Document.Blocks.Add(new Paragraph(new Run(x.EventArgs.Results))));

    // start running the test
    this.runningTest = new Task(() => { this.myTest.Run(); });
    this.runningTest.Start();

    // block on purpose, wait for the task to finish
    this.runningTest.Wait();
}

След това абонаментът се изхвърля автоматично вместо вас.

(2)

Можете да използвате SerialDisposable за управление на всеки абонамент. Документацията на MSDN го описва като:

Представлява материал за еднократна употреба, чийто основен артикул за еднократна употреба може да бъде заменен с друг артикул за еднократна употреба, което води до изхвърляне на предишния базов артикул за еднократна употреба.

Вашият код тогава ще изглежда така:

private SerialDisposable _results = new SerialDisposable();

private void button_go_Click(object sender, RoutedEventArgs e)
{
    var uiContext = SynchronizationContext.Current;
    _results.Disposable = Observable.FromEventPattern<TestResultHandler, TestResultArgs>(
        handler => (s, a) => handler(s, a),
        handler => this.myTest.Results += handler,
        handler => this.myTest.Results -= handler)
        .ObserveOn(uiContext)
        .Subscribe(x => this.richTextBox_testResults.Document.Blocks.Add(new Paragraph(new Run(x.EventArgs.Results))));

    // start running the test
    this.runningTest = new Task(() => { this.myTest.Run(); });
    this.runningTest.Start();

    // block on purpose, wait for the task to finish
    this.runningTest.Wait();
}

(3)

Винаги можете да се уверите, че създавате абонамента само веднъж.

private IDisposable _results = null;

private void button_go_Click(object sender, RoutedEventArgs e)
{
    if (_results == null)
    {
        var uiContext = SynchronizationContext.Current;
        _results = Observable.FromEventPattern<TestResultHandler, TestResultArgs>(
            handler => (s, a) => handler(s, a),
            handler => this.myTest.Results += handler,
            handler => this.myTest.Results -= handler)
            .ObserveOn(uiContext)
            .Subscribe(x => this.richTextBox_testResults.Document.Blocks.Add(new Paragraph(new Run(x.EventArgs.Results))));
    }

    // start running the test
    this.runningTest = new Task(() => { this.myTest.Run(); });
    this.runningTest.Start();

    // block on purpose, wait for the task to finish
    this.runningTest.Wait();
}

(4)

Крайният подход е, когато цялата операция е обвита в наблюдаем и се подбужда за всеки нов абонамент. Това е правилният начин за работа с Rx. Наблюдаемите трябва да поддържат собствено състояние, така че абонаментите да могат да бъдат 100% независими един от друг.

Сега вашият код използва this.myTest & this.runningTest, така че ясно има състояние. Трябва да опитате да ги премахнете. Но без да го направите, вашият код ще изглежда така:

private void button_go_Click(object sender, RoutedEventArgs e)
{
    var uiContext = SynchronizationContext.Current;
    Observable
        .Create<System.Reactive.EventPattern<TestResultArgs>>(o =>
        {
            var subscription =
                Observable
                    .FromEventPattern<TestResultHandler, TestResultArgs>(
                        h => this.myTest.Results += h,
                        h => this.myTest.Results -= h)
                    .Take(1)
                    .Subscribe(o);
            this.runningTest = new Task(() => { this.myTest.Run(); });
            this.runningTest.Start();
            return subscription;
        })
        .ObserveOn(uiContext)
        .Subscribe(x => this.richTextBox_testResults.Document.Blocks.Add(new Paragraph(new Run(x.EventArgs.Results))));

    // block on purpose, wait for the task to finish
    this.runningTest.Wait();
}

В идеалния случай трябва да включите създаването и унищожаването на myTest в наблюдаемото.

Така че бих бил склонен да направя нещо подобно:

private void button_go_Click(object sender, RoutedEventArgs e)
{
    var uiContext = SynchronizationContext.Current;
    Observable
        .Create<System.Reactive.EventPattern<TestResultArgs>>(o =>
        {
            var myTest = new MyTest();
            var subscription =
                Observable
                    .FromEventPattern<TestResultHandler, TestResultArgs>(
                        h => myTest.Results += h,
                        h => myTest.Results -= h)
                    .Take(1)
                    .Subscribe(o);
            myTest.Run();
            return subscription;
        })
        .ObserveOn(uiContext)
        .Subscribe(x => this.richTextBox_testResults.Document.Blocks.Add(new Paragraph(new Run(x.EventArgs.Results))));
}

Това последното наистина е отговорът на въпроса ви: „Или има ли начин да наблюдавам задачата си само докато съществува?“