Я должен обеспечить завершение Observable. Это наблюдение за потоковыми данными, изменение которых не входит в мои обязанности. Потоковые данные могут вообще не тикать. Я столкнулся с использованием предметного члена, который я могу использовать для ручного завершения наблюдаемого.
private readonly Subject<bool> _mySubject = new Subject<bool>();
// relevant property that should be completed
public IObservable<myModel> StreamObservable { get; private set; }
IObservable<myModel> someDataObservable = ... ; // may not tick in scenario
// this doesn't work
StreamObservable = someDataObservable
.TakeUntil(_mySubject)
.Finally(() => logger.LogInformation("completed!"));
_mySubject.OnNext(true);
// I tried Amb.. still not going into Finally(), also when someDataObservable doesn't tick
StreamObservable = Observable.Amb(_mySubject.SelectMany(value => Observable.Empty<myModel>()), someDataObservable)
.Finally(() => logger.LogInformation("completed!"));
_mySubject.OnNext(true);
Как я могу завершить, предпочтительно используя Subject.OnNext() ?
Триггер темы - это просто идея. Я предпочитаю оставаться в мире наблюдения и преобразовывать последовательность в Observable.Empty(), например. когда вызывается такой обратный вызов:
private Task<bool> FinishStreamCallback()
{
_mySubject.OnNext(true); // my first approach
return Task.FromResult(true);
}
.. нацеливание с использованием OnCompleted()-подписок на соответствующие общедоступные наблюдаемые объекты в другом классе.