Jak hermetyzacji tworzenie długich reaktywnych łańcuchów obserwabli

Question

Obecnie mam następujący fragment kodu Rx/ReactiveUI:

 this.WhenAnyValue(x => x.Listras) 
      .Where(item => item != null) 
      .Throttle(TimeSpan.FromMilliseconds(millis)) 
      .ObserveOn(TaskPoolScheduler.Default) 
      .Select(im => GetArray.FromChannels(im, 0, 1)) 
      .ObserveOn(RxApp.MainThreadScheduler) 
      .ToProperty(this, x => x.Grayscale, out _grayscale); 

     this.WhenAnyValue(x => x.Grayscale) 
      .Where(item => item != null) 
      .Throttle(TimeSpan.FromMilliseconds(millis)) 
      .ObserveOn(TaskPoolScheduler.Default) 
      .Select(ar => Gaussian.GaussianConvolution(ar, 1.5)) 
      .ObserveOn(RxApp.MainThreadScheduler) 
      .ToProperty(this, x => x.BlurMenor, out _blurMenor); 

     this.WhenAnyValue(x => x.BlurMenor) 
      .Where(item => item != null) 
      .Throttle(TimeSpan.FromMilliseconds(millis)) 
      .ObserveOn(TaskPoolScheduler.Default) 
      .Select(ar => { ConversorImagem.Converter(ar, out BitmapSource im); return im; }) 
      .ObserveOn(RxApp.MainThreadScheduler) 
      .ToProperty(this, x => x.ImagemBlurMenor, out _imagemBlurMenor); 

     this.WhenAnyValue(x => x.BlurMenor) 
      .Where(item => item != null) 
      .Throttle(TimeSpan.FromMilliseconds(millis)) 
      .ObserveOn(TaskPoolScheduler.Default) 
      .Select(ar => Gaussian.VerticalGaussianConvolution(ar, 5)) 
      .ObserveOn(RxApp.MainThreadScheduler) 
      .ToProperty(this, x => x.BlurMaior, out _blurMaior); 

     this.WhenAnyValue(x => x.BlurMaior) 
      .Where(item => item != null) 
      .Throttle(TimeSpan.FromMilliseconds(millis)) 
      .ObserveOn(TaskPoolScheduler.Default) 
      .Select(ar => { ConversorImagem.Converter(ar, out BitmapSource im); return im; }) 
      .ObserveOn(RxApp.MainThreadScheduler) 
      .ToProperty(this, x => x.ImagemBlurMaior, out _imagemBlurMaior); 

     this.WhenAnyValue(x => x.BlurMenor, x => x.BlurMaior) 
      .Where(tuple => tuple.Item1 != null && tuple.Item2 != null) 
      .Throttle(TimeSpan.FromMilliseconds(millis)) 
      .ObserveOn(TaskPoolScheduler.Default) 
      .Select(tuple => ArrayOperations.Diferença(tuple.Item1, tuple.Item2)) 
      .ObserveOn(RxApp.MainThreadScheduler) 
      .ToProperty(this, x => x.Diferença, out _diferença); 

     this.WhenAnyValue(x => x.Diferença) 
      .Where(item => item != null) 
      .Throttle(TimeSpan.FromMilliseconds(millis)) 
      .ObserveOn(TaskPoolScheduler.Default) 
      .Select(ar => { ConversorImagem.Converter(ar, out BitmapSource im); return im; }) 
      .ObserveOn(RxApp.MainThreadScheduler) 
      .ToProperty(this, x => x.ImagemDiferença, out _imagemDiferença); 

Jak widać, to rażąco narusza zasadę DRY, ale nie wiem jak Mogę sparametryzować przekazywanie właściwości i delegatów.

Jaki jest zwykle sposób automatyzacji tworzenia tych łańcuchów metod w Rx/ReactiveUI?

Answer 1

Piękno Rx polega na tym, że można tworzyć własne operatory na podstawie innych operatorów. Dzieje się tak ze względu na funkcjonalną stronę Rx.

Można utworzyć nowy podmiot, który hermetyzuje cały wspólny zachowanie i przyjmuje małych różnic jako parametry:

// Put this class somewhere useful. Either beside your VM inside the same namespace 
// Or in a seperate file for all your custom operators 
public static class ObservableMixins 
{ 
    public static IObservable<TOut> ThrottledSelect<TIn, TOut>(this IObservable<TIn> source, int milliseconds, Func<TIn, TOut> selector) => 
     source 
      .Where(item => item != null) 
      .Throttle(TimeSpan.FromMilliseconds(milliseconds)) 
      .ObserveOn(TaskPoolScheduler.Default) 
      .Select(selector) 
      .ObserveOn(RxApp.MainThreadScheduler) 
} 

Zastosowanie to tak:

this.WhenAnyValue(x => x.Listras) 
    .ThrottledSelect(millis, im => GetArray.FromChannels(im, 0, 1)) 
    .ToProperty(this, x => x.Grayscale, out _grayscale); 

this.WhenAnyValue(x => x.Grayscale) 
    .ThrottledSelect(millis, ar => Gaussian.GaussianConvolution(ar, 1.5)) 
    .ToProperty(this, x => x.BlurMenor, out _blurMenor); 

this.WhenAnyValue(x => x.BlurMenor) 
    .ThrottledSelect(millis, ar => { ConversorImagem.Converter(ar, out BitmapSource im); return im; }) 
    .ToProperty(this, x => x.ImagemBlurMenor, out _imagemBlurMenor); 

this.WhenAnyValue(x => x.BlurMenor) 
    .ThrottledSelect(millis, ar => Gaussian.VerticalGaussianConvolution(ar, 5)) 
    .ToProperty(this, x => x.BlurMaior, out _blurMaior); 

this.WhenAnyValue(x => x.BlurMaior) 
    .ThrottledSelect(millis, ar => { ConversorImagem.Converter(ar, out BitmapSource im); return im; }) 
    .ToProperty(this, x => x.ImagemBlurMaior, out _imagemBlurMaior); 

this.WhenAnyValue(x => x.BlurMenor, x => x.BlurMaior) 
    // Notice how I'm returning a null if either item is null 
    // It will be filtered in the operator 
    .Select(tuple => tuple.Item1 != null || tuple.Item2 != null ? null : tuple) 
    .ThrottledSelect(millis, tuple => ArrayOperations.Diferença(tuple.Item1, tuple.Item2)) 
    .ToProperty(this, x => x.Diferença, out _diferença); 

this.WhenAnyValue(x => x.Diferença) 
    .ThrottledSelect(millis, ar => { ConversorImagem.Converter(ar, out BitmapSource im); return im; }) 
    .ToProperty(this, x => x.ImagemDiferença, out _imagemDiferença); 

Jeśli czujesz się nieco mniej przygód, można oczywiście użyć go jako zwykłej metody, która ma obserwowalne:

public IObservable<T> ThrottledSelect<TIn, TOut>(IObservable<TIn> source, int milliseconds, Func<TIn, TOut> selector) => 
    source 
     .Where(item => item != null) 
     .Throttle(TimeSpan.FromMilliseconds(milliseconds)) 
     .ObserveOn(TaskPoolScheduler.Default) 
     .Select(selector) 
     .ObserveOn(RxApp.MainThreadScheduler) 

I użyj tego tak:

ThrottledSelect(this.WhenAnyValue(x => x.Diferença), millis, ar => { ConversorImagem.Converter(ar, out BitmapSource im); return im; }) 
    .ToProperty(this, x => x.ImagemDiferença, out _imagemDiferença);