处理 System.Reactive 订阅中的异常

Question

我想使用System.Reactive显示以下所有订阅的异常。

下面的一小段代码，你可以看到我尝试了什么，但它仍然没有显示消息。

client.Streams.AggregateTradesStream
.Subscribe(response =>
{
throw new Exception("Asd");
Guard.Against.Null(response, nameof(response));
Guard.Against.Null(response.Data, nameof(response.Data),
"Something went wrong and the aggregated trade object is null");
var trade = response.Data;
Log.Information($"Aggregated trade [{trade.Symbol}] [{trade.Side}] " +
$"Price: {trade.Price} Size: {trade.Quantity}");
}, ex => Console.WriteLine("Exception: {0} {1}", ex.Message, DateTime.Now))
.DisposeWith(disposable);

我如何处理/显示所有.Subscribe的异常，因为我所做的不起作用？也许，扩展方法可能会有所帮助。

完整代码

public class Program
{
private static readonly ManualResetEvent ExitEvent = new(false);
private static async Task Main()
{
Log.Logger = new LoggerConfiguration()
.MinimumLevel.Verbose()
.Enrich.FromLogContext()
.WriteTo.Console(LogEventLevel.Debug, theme: SystemConsoleTheme.Colored)
.WriteTo.File(Path.Combine("logs", "verbose.log"), rollingInterval: RollingInterval.Day)
.CreateLogger();
var disposable = new CompositeDisposable();
var uri = new Uri("wss://stream.binance.com:9443");
using var communicator = new BinanceWebSocketCommunicator(uri);
communicator.Name = "Binance-Spot";
communicator.ReconnectTimeout = TimeSpan.FromMinutes(10);
communicator.ReconnectionHappened
.Subscribe(info => Log.Information($"Reconnection happened, type: {info.Type}"))
.DisposeWith(disposable);
communicator.DisconnectionHappened
.Subscribe(info => Log.Information($"Disconnection happened, type: {info.Type}"))
.DisposeWith(disposable);
using var client = new BinanceWebSocketClient(communicator);
client.Streams.PongStream
.Subscribe(x => Log.Information($"Pong received ({x.Message})"))
.DisposeWith(disposable);
client.Streams.AggregateTradesStream
.Subscribe(response =>
{
throw new Exception("Asd");
Guard.Against.Null(response, nameof(response));
Guard.Against.Null(response.Data, nameof(response.Data),
"Something went wrong and the aggregated trade object is null");
var trade = response.Data;
Log.Information($"Aggregated trade [{trade.Symbol}] [{trade.Side}] " +
$"Price: {trade.Price} Size: {trade.Quantity}");
}, ex => Console.WriteLine("Exception: {0} {1}", ex.Message, DateTime.Now))
.DisposeWith(disposable);
client.Streams.KlineStream
.Subscribe(response =>
{
Guard.Against.Null(response, nameof(response));
Guard.Against.Null(response.Data, nameof(response.Data),
"Something went wrong and the kline object is null");
Guard.Against.Null(response.Data.Data, nameof(response.Data.Data),
"Something went wrong and the kline data object is null");
var kline = response.Data;
var klineData = response.Data.Data;
Log.Information($"Kline [{kline.Symbol}] " +
$"Kline start time: {klineData.StartTime} " +
$"Kline close time: {klineData.CloseTime} " +
$"Interval: {klineData.Interval} " +
$"First trade ID: {klineData.FirstTradeId} " +
$"Last trade ID: {klineData.LastTradeId} " +
$"Open price: {klineData.OpenPrice} " +
$"Close price: {klineData.ClosePrice} " +
$"High price: {klineData.HighPrice} " +
$"Low price: {klineData.LowPrice} " +
$"Base asset volume: {klineData.BaseAssetVolume} " +
$"Number of trades: {klineData.NumberTrades} " +
$"Is this kline closed?: {klineData.IsClosed} " +
$"Quote asset volume: {klineData.QuoteAssetVolume} " +
$"Taker buy base: {klineData.TakerBuyBaseAssetVolume} " +
$"Taker buy quote: {klineData.TakerBuyQuoteAssetVolume} " +
$"Ignore: {klineData.Ignore} ");
})
.DisposeWith(disposable);
client.AddSubscription(
new AggregateTradeSubscription("bnbusdt"),
new KlineSubscription("btcusdt", "1h"));
await communicator.Start().ConfigureAwait(false);
ExitEvent.WaitOne();
disposable.Dispose();
Log.CloseAndFlush();
}
}

Answer 1

我意识到，为了在.Subscribe中显示异常，我必须将其封装在.Select/.SelectMany或try catch块中。

client.Streams.AggregateTradesStream
.Subscribe(response =>
{
try
{
Guard.Against.Null(response, nameof(response));
Guard.Against.Null(response.Data, nameof(response.Data));
var trade = response.Data;
Log.Information($"Aggregated trade [{trade.Symbol}] [{trade.Side}] " +
$"Price: {trade.Price} Size: {trade.Quantity}");
}
catch (Exception ex)
{
logger.LogError(ex, "Exception while receiving message");
}
})
.DisposeWith(disposable);