c# - 如何在不使用内部 Subject 的情况下链接转换 ISubject

Question

我们有一系列正在使用 Reactive 处理的项目。处理的一部分是一个可配置的管道，它转换项目(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");
    }

}

同样，一些 yield 和成本。相比之下更容易实现，但基础设施会更困难(可观察的动态表达式构造)。

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

其次，(我对响应式(Reactive)和函数式编程还很陌生)转换和“行动”(即绑定(bind)转换任务和观察者报告任务)是绝对不同的，选择应该用于一个而订阅(观察者)另一个？

Answer 1

我不确定你是否需要 Subject 的...

而不是将自己局限于映射，您可能想要转换整个可观察对象，因此您可以使用这样的接口(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();

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