Реализация метода хвоста с помощью DynamicData от Roland Pheasant

Question

Я пытаюсь понять, как использовать библиотеку DynamicData .Мне нужно сделать метод, который будет обрабатывать изменения источника и передавать только последние n изменений.Может быть названо Tail () .В пакете есть метод с именем Top () , он использует IVirtualRequest.Я написал пример, основанный на этом функционале:

public static class Extensions
    {
        public static IObservable<IChangeSet<T>> Tail<T>([NotNull] this IObservable<IChangeSet<T>> source,
                                                         int numberOfItems)
        {    
            var request = new TailRequest<T>(source, numberOfItems);
            return source.Virtualise(Observable.Return(request));
        }
    }

    public class TailRequest<T> : IVirtualRequest, IDisposable
    {
        private readonly IDisposable subscription;
        private int _count;

        public int Size { get; }

        public int StartIndex => _count > Size ? _count - Size : 0;

        public TailRequest(IObservable<IChangeSet<T>> source, int numberOfItems)
        {
            //how to dispose this from outside???
            subscription = source.Subscribe(RefreshStartIndex);
            Size = numberOfItems;
        }

        private void RefreshStartIndex(IChangeSet<T> changeSet)
        {
            _count += changeSet.Adds;
            _count -= changeSet.Removes;
        }

        public void Dispose()
        {
            subscription.Dispose();
        }
    }

Но я не могу понять, как мне распоряжаться этим запросом извне, если я использую этот метод, например:

SourceList<Message> sourceList = new SourceList<Message>();
var subscription = _sourceList.Connect()
            .Tail(15)
            .ObserveOn(RxApp.MainThreadScheduler)
            .Bind(Messages)
            .Subscribe();

Iдумаю, что подписка. Dispose () не будет распоряжаться моим TailRequest.Или есть лучшее решение?

Answer 1

Это сделает свое дело

public static IObservable<IChangeSet<T>> Tail<T>([NotNull] this IObservable<IChangeSet<T>> source, int numberOfItems)
{
    return Observable.Create<IChangeSet<T>>(observer =>
    {
        var request = new TailRequest<T>(source, numberOfItems);

        return new CompositeDisposable
        (
         request,
         source.Virtualise(Observable.Return(request)).SubscribeSafe(observer)
        );
});

Функция внутри Observable.Create возвращает одноразовое использование, которое используется для очистки любых ресурсов, созданных наблюдаемой. Когда абонент-потребитель утилизируется, внутренний расходник, созданный в функции, также утилизируется.