在 RX 和信号器中返回错误

Question

我有一个信号器(.Net Core版本(中心，它使用IObservable返回数据流。这几乎是直接从此流式处理示例中提升的：https://github.com/radu-matei/signalr-samples/tree/master/streaming/web

public IObservable<MyStreamItem> StreamData(SomeRequestData request)
{}

我想验证传入的请求是否正常。只是标准的验证内容。有效的电子邮件、日期等

所以我正在尝试使用 Observable.Throw 将此错误传达回客户端

public IObservable<MyStreamItem> StreamData(SomeRequestData request)
{
    if (!IsValid(request))
    {
        return Observable.Throw<MyStreamItem>(new Exception("Invalid Request"));
    }   
}

这是这样做的方法吗？错误消息在浏览器的通信中以某种方式丢失，因此流被终止，但没有错误消息发送到客户端

我最初尝试在某些中间件中执行此操作，但似乎只有 websocket 的初始创建被中间件捕获，而不是后续调用

知道我做错了什么吗？

Answer 1

这是一个已知问题，现在有一个拉取请求，用于将异常从 Stream hub 方法传播到现在打开的客户端。它将在即将发布的 SignalR Core 预览版 1 版本中提供。链接到拉取请求：https://github.com/aspnet/SignalR/pull/1331

Answer 2

为了帮助您了解 Rx，我建议您重写 StreamData 。当调用StreamData和订阅生成的流之间有一些时间时，当前实现会产生问题。调用 StreamData 时请求可能无效，但不久之后，在订阅之前，请求无效。

以下是您的编写方法：

public IObservable<MyStreamItem> StreamData(SomeRequestData request)
{
    return Observable.Defer(() =>
    {
        if (!IsValid(request))
        {
            return Observable.Throw<MyStreamItem>(new Exception("Invalid Request"));
        }
        else
            //return observable when request is valid
    });
}

Answer 3

如果您以与链接示例中相同的方式执行此操作，则应执行此操作：

observer.OnError(new Exception("Invalid Request"));

在示例中，他的误差函数是错误的(名为"err"(。它应该是：

error: function(error) {
    console.log(error);
}

为了适应传递异常，我会更改上述内容并发送另一个OnNext，但传递一个这样的对象：

StreamingHub

public class Message
{
    public string Text { get; set; }
    public Exception Error { get; set; }
}
public IObservable<Message> StartStreaming()
{
    return Observable.Create(
        async (IObserver<Message> observer) =>
        {
            for (int i = 0; i < 2; i++)
            {
                observer.OnNext(new Message { Text = $"sending...{i}" });
                await Task.Delay(1000);
            }
            observer.OnNext(new Message { Error = new Exception("Invalid Request") });
        });

索引.html

function onStreamReceived(data){
    if (data.text) {
        console.log("received: " + data.text);
        var liElement = document.createElement('li');
        liElement.innerHTML = '<strong>' + "received" + '</strong>:&nbsp;&nbsp;' + data.text;
        document.getElementById('discussion').appendChild(liElement);
    }
    if (data.error) {
        console.log("error: " + data.error.Message);
    }
}