c# - 如何在不使用内部主题<T,U>的情况下链接变革性 ISubject<T,U>

Question

我们有一系列正在使用响应式处理的项目。处理的一部分是一个可配置的管道，用于转换项目(T -> U，例如在一个简单的情况下 int -> char)。例如，我们的简单实现看起来像

// simple base class, implements IObserver and IObservable which is
// equivalent to ISubject<T,U>
public abstract class ObservableTask<T, U> : IObserver<T>, IObservable<U>
{
    // NOTE: stateful, blech
    private readonly Subject<U> observable = new Subject<U>();

    public void OnCompleted() { }
    public void OnError(Exception error) { }
    public void OnNext(T value) { observable.OnNext(Process(value)); }
    public IDisposable Subscribe(IObserver<U> observer) 
    {
        return observable.Subscribe(observer);
    }
    public abstract U Process(T value);
}

// trivial implementation of a transform task, transforms an
// input of type int into an output of type char
public class TransformTask : ObservableTask<int, char>
{
    public override char Process(int value)
    {
        Console.WriteLine("Transform '{0}'", value);
        return (char)(value + 64);
    }
}

// trivial report, does not transform but performs IO-bound
// task and passes value to any other subsequent subscribers
public class ReportTask : ObservableTask<char, char>
{
    public override char Process(char value)
    {
        Console.WriteLine("Report '{0}'", value);
        return value;
    }
}

// simple harness that produces desired output/behaviour
public static class ObservableTasks
{
    public static void ChainThings()
    {
        Console.WriteLine("begin observable tasks");

        // NOTE: would use config/reflection to assemble pipe;
        // here we use concrete instances for demonstration only
        TransformTask a = new TransformTask();
        ReportTask b = new ReportTask();

        int[] numbers = new int[] { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, };
        var s = numbers.ToObservable().Publish();
        a.Subscribe(b);
        s.Subscribe(a);
        s.Connect();
        Console.WriteLine("begin observable tasks");
    }
}

上述模型有很多好处；也就是说，我们可以直观地开发工作单元并创建一个简单的框架来将任何类型的管道组装在一起。

但是，如上所述，我们的内部Subject<T,U>是 faux-pas 。我在 Observable.* 上使用工厂方法来模拟前面的行为(即创建一个单独的 Observable 供订阅者在元素到达时订阅和调用)的运气不太好。

唯一的其他引用transforming sequences稍后，指的是 Linq 用法。理论上我们可以适应这样的事情

public class TransformTask
{
    public char Select(int value)
    {
        Console.WriteLine("Transform '{0}'", value);
        return (char)(value + 64);
    }
}

public class ReportTask
{
    public char Select(char value)
    {
        Console.WriteLine("Report '{0}'", value);
        return value;
    }
}

public static class SelectTasks
{
    public static void ChainThings()
    {
        Console.WriteLine("begin select tasks");
        TransformTask a = new TransformTask();
        ReportTask b = new ReportTask();
        int[] numbers = new int[] { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, };

        // in theory we could build this up dynamically 
        // with expression trees
        var s = numbers.
            ToObservable().
            Select(a.Select).
            Select(b.Select).
            Publish();

        // empty subscription?
        s.Subscribe(value => { });
        s.Connect();
        Console.WriteLine("end select tasks");
    }

}

再说一遍，一些好处和成本。相比之下，实现起来更简单，但基础设施会更困难(可观察的动态表达式构造)。

首先，有人解决过类似的问题并能够分享一些见解吗？

第二，(我对响应式(Reactive)和函数式编程很陌生)转换和“行为”(即绑定(bind) TransformTask 和 Observer ReportTask)绝对不同，一个应该使用 Select，另一个应该使用 Subscribe(observer)？

Answer 1

我不确定您是否需要主题...

您可能希望转换整个可观察对象，而不是将自己限制在映射上，这样您就可以使用这样的接口(interface)...

public interface IStep<T, TResult>
{
    public IObservable<TResult> Transform(IObservable<T> source);
}

一旦存在，您可以定义几个扩展方法(只是为了方便)来帮助使用该步骤，就像这样......

public static class ObservableExtensions
{
    public static IObservable<TResult> Let(this IObservable<T> source, Func<IObservable<T>, IObservable<TResult>> let)
    {
        return let(source);
    }

    public static IObservable<TResult> Let(this IObservable<T> source, IStep<T, TResult> step)
    {
      return source.Let(step.Transform);
    }
}

然后你可以像这样定义你的步骤......

public class TransformStep : IStep<int, char>
{
    public IObservable<char> Transform(IObservable<int> source)
    {
        return source.Map(IntToChar); 
    }

    public char IntToChar(int value)
    {
        return (char)(value + 64);
    }
}

public class ReportStep : IStep<char, char>
{
    private readonly Logger logger;

    public ReporterStep(Logger logger)
    {
      this.logger = logger;
    }

    public IObservable<char> Transform(IObservable<char> source)
    {
        return source.Do(Report);
    }

    public void Report(char value)
    {
        logger.Log("Report '{0}'", value);
    }
}

并以或多或少统一的方式使用它们......

Observable.Return<int>(10)
  .Let(new TransformStep())
  .Let(new ReportStep(logger))
  .Subscribe();

这样，与每个步骤相关的所有逻辑都可以在该步骤的内部，您只需进行序列化/反序列化，然后将它们链接在一起即可。