可观察的事件答案

【问题标题】：Observable events可观察的事件
【发布时间】：2015-10-18 03:41:55
【问题描述】：

我有这段代码，原代码here，我需要修改以在字符串列表中插入输出字符串。

我该怎么做？

    private static List<string> list_of_proc = new List<string>();

    static void Run<T>(Func<IObservable<T>> f, Action a)
    {
        Action<string> messageTarget = (x) => list_of_proc.Add(x);
        using (f().Subscribe(t => Console.WriteLine(t), p => messageTarget(p)))
        {
            Console.ReadLine();
            Console.WriteLine();
        }
    }

我的操作 messageTarget 没有编译。

输出错误：

1>Program.cs(20,34,20,59)：错误 CS1660：无法将 lambda 表达式转换为类型“System.IObserver”，因为它不是委托类型 1>Program.cs(20,61,20,82): 错误 CS1593: Delegate 'System.Action' 不接受 1 个参数 1>Program.cs(20,61,20,82): error CS1660: Cannot convert lambda expression to type 'System.Threading.CancellationToken' 因为它不是委托类型 1>Program.cs(20,66,20,82): 错误 CS1594: Delegate 'System.Action' 有一些无效参数 1>Program.cs(20,80,20,81): error CS1503: Argument 1: cannot convert from 'System.Exception' to 'string'

这段代码编译正常，没有错误：

        static void Run<T>(Func<IObservable<T>> f, Action a)
    {
        Action<string> messageTarget = (x) => list_of_proc.Add(x);
        using (f().Subscribe(t => Console.WriteLine(t)/*, p => messageTarget(p)*/))
        {
            Console.ReadLine();
            Console.WriteLine();
        }
    }

订阅有这种形式（System.Reactive.Core.dll，v2.2.5.0）：

公共静态 IDisposable 订阅（此 IObservable 源，Action onNext）；

这个正确的代码：

    static void Run<T>(Func<IObservable<T>> f, Action a)
    {
        Action<string> messageTarget = (x) => list_of_proc.Add(x);
        using (f().Subscribe( t => messageTarget(t.ToString())))
        {
            Console.ReadLine();
            Console.WriteLine();
        }
    }

【问题讨论】：

Subscribe的签名是什么？

标签： c# events system.reactive

【解决方案1】：

IObservable<T> 的.Subscribe(...) 扩展方法有很多。编译器将尝试找到最合适的重载。在您的情况下，它按以下顺序尝试这些重载：

IDisposable Subscribe<T>(this IObservable<T> source)
IDisposable Subscribe<T>(this IObservable<T> source, Action<T> onNext, Action onCompleted)
void Subscribe<T>(this IObservable<T> source, Action<T> onNext, CancellationToken token)
IDisposable Subscribe<T>(this IObservable<T> source, Action<T> onNext, Action<Exception> onError)

然而，你的电话是这样的：

.Subscribe(this IObservable<T> source, Action<T> onNext, Action<string> ???)

不匹配，但正在尝试为您提供有用的提示：

Cannot convert lambda expression to type 'System.IObserver<T>' because it is not a delegate type
Delegate 'System.Action' does not take 1 arguments
Cannot convert lambda expression to type 'System.Threading.CancellationToken' because it is not a delegate type
Delegate 'System.Action<string>' has some invalid arguments

每个都有效地尝试匹配上述重载之一。

您的代码不清楚的是，当您运行代码时，T 的实际类型是什么。为 t 类型的 T 调用 Console.WriteLine(t) 意味着您可能会强制 t.ToString() 输出结果 - 这并不总是会产生有意义的东西。

然后用Action<string> 跟进Action<T> 也没有意义，因为string 不是Exception，并且.Subscribe(...) 没有过载，这将需要一个明确的Action<string>。

我认为你需要的是这样的：

static IObservable<IList<string>> Run<T>(Func<IObservable<T>> f)
{
    return Observable
        .Defer(f)
        .Select(t => t.ToString())
        .ToArray()
        .Select(ss => ss.ToList());
}

或者，也许，这个：

static IObservable<IList<string>> Run<T>(Func<IObservable<T>> f, Func<T, string> toString)
{
    return Observable
        .Defer(f)
        .Select(t => toString(t))
        .ToArray()
        .Select(ss => ss.ToList());
}

然后你可以这样称呼：

Run<int>(() => Observable.Range(0, 3), n => n.ToString())
    .Subscribe(results => /* handle list of results */);

【讨论】：

【解决方案2】：

您使用 IObserver 作为函数的第一个参数，但实际上给了它一个委托，编译器告诉您它不是委托。就这么简单。

【讨论】：

【解决方案3】：

您的Subscribe 方法是一个扩展方法。 this 参数隐含地是您调用该方法的对象。这意味着您应该只将 Action 委托作为唯一的参数传递。

f().Subscribe(() => Console.WriteLine("Hello"))

【讨论】：