Удаление реактивной наблюдаемой подписки

Отказ от ответственности: я также все еще изучаю Rx. Так что я на самом деле не эксперт, но я считаю, что одноразовый товар, возвращенный Subscribe, только отменит подписку. Также, если источник завершается, как в вашем случае, отписка выполняется автоматически. Поэтому я думаю, что Dispose здесь избыточен и может быть безопасно удален.

См. ответ на этот вопрос для получения дополнительной информации.

Одноразовый объект, возвращаемый методами расширения Subscribe, возвращается исключительно для того, чтобы позволить вам вручную отказаться от подписки на наблюдаемый объект до естественного завершения наблюдаемого.

Если наблюдаемое завершается — либо с OnCompleted, либо с OnError — тогда подписка уже готова для вас.

Попробуйте этот код:

var xs = Observable.Create<int>(o =>
{
    var d = Observable.Return(1).Subscribe(o);
    return Disposable.Create(() =>
    {
        Console.WriteLine("Disposed!");
        d.Dispose();
    });
});

var subscription = xs.Subscribe(x => Console.WriteLine(x));

Если вы запустите приведенное выше, вы увидите, что «Disposed!» записывается в консоль, когда наблюдаемое завершается без необходимости вызова .Dispose() для подписки.

Следует отметить одну важную вещь: сборщик мусора никогда не вызывает .Dispose() для наблюдаемых подписок, поэтому вы должны удалять свои подписки, если они не завершились (или могли не закончиться) естественным образом до того, как ваша подписка выйдет из области действия.

Возьмите это, например:

var wc = new WebClient();

var ds = Observable
    .FromEventPattern<
        DownloadStringCompletedEventHandler,
        DownloadStringCompletedEventArgs>(
            h => wc.DownloadStringCompleted += h,
            h => wc.DownloadStringCompleted -= h);

var subscription =
    ds.Subscribe(d =>
        Console.WriteLine(d.EventArgs.Result));

Наблюдаемый объект ds будет присоединяться к обработчику событий только тогда, когда у него есть подписка, и будет отсоединяться только после завершения наблюдаемого объекта или удаления подписки. Поскольку это обработчик событий, наблюдаемый никогда не завершится, потому что он ожидает большего количества событий, и, следовательно, удаление — единственный способ отсоединиться от события (для приведенного выше примера).

Когда у вас есть наблюдаемая FromEventPattern, которая, как вы знаете, когда-либо вернет только одно значение, разумно добавить метод расширения .Take(1) перед подпиской, чтобы позволить обработчику событий автоматически отсоединяться, и тогда вам не нужно вручную удалять подписку.

Вот так:

var ds = Observable
    .FromEventPattern<
        DownloadStringCompletedEventHandler,
        DownloadStringCompletedEventArgs>(
            h => wc.DownloadStringCompleted += h,
            h => wc.DownloadStringCompleted -= h)
    .Take(1);

Надеюсь, это поможет.

Функция Take сделает именно то, что вы ищете. В данном случае Take(1).

В отличие от некоторых комментариев, совсем не редкость избавиться от подписки изнутри OnNext.

Хотя верно, что удаление в OnCompleted и OnError выполняется за вас с помощью обернутой подписки, которую создает метод расширения Subscribe, вы может захотеть отказаться от подписки на основе значения, которое вы наблюдаете (например, в вашем случае: 1-е). У вас не всегда есть наблюдаемая, которая, как известно, производит только одно значение.

Проблема в том, что вы получаете IDisposable только после подписки. Наблюдаемый объект может перезвонить вам по OnNext еще до того, как он вернет вам IDisposable для отказа от подписки (в зависимости от таких вещей, как IScheduler, которые он использует).

System.Reactive.Disposables.SingleAssignmentDisposable пригодится в этом случае. Он упаковывает IDisposable, который вы можете назначить поздно, и немедленно утилизирует его при назначении, если SingleAssignmentDisposable к тому времени уже был удален. Также он содержит свойство IsDisposed, которое изначально равно false и устанавливается в true при вызове Dispose().

So:

IObservable<string> source = ...;

var subscription = new SingleAssignmentDisposable();
subscription.Disposable = source.Subscribe(x =>
{
    if (subscription.IsDisposed) // getting notified though I've told it to stop
        return;
    DoThingsWithItem(x);
    if (x == "the last item I'm interested in")
        subscription.Dispose();
});