如何封装可观察量长反应链的创建

Question

目前我有以下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);

如您所见，它公然违反了 DRY 原则，但我不知道如何参数化属性和委托的传递。

在 Rx/ReactiveUI 中自动创建这些方法链的常用方法是什么？

Answer 1

Rx 的美妙之处在于您可以基于其他运算符创建自己的运算符。这是因为 Rx 的功能方面。

您可以创建一个新的运算符，该运算符封装所有常见行为并将微小的差异作为参数：

// 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)
}

像这样使用它：

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);

如果你觉得不那么冒险，你当然可以把它作为一种常规方法，它需要一个可观察的：

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)

并像这样使用它：

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);