Hi. I have problem with error handling within R. I'm using SourceCache in .net. And I'm using.Subscribe(()=> throw new Exception, e=> Log.Error(e))
I would suspect lambda passed to onError will trigger, its not triggering.
At this moment I'm getting Unhandled exception and application crash.
I need a way to not crash the system when exception in subscribe happening.
Does anyone have any clue why it's behaving like this or how can I catch this exception?
.Select(()=> throw new Exception())
.Subscribe(() => {}, e=> Log.Error(e));
public void StartGetEventsTimer()
{
Observable
.Timer(TimeSpan.FromSeconds(0), TimeSpan.FromSeconds(5))
.ObserveOn(MainFrmSyncContext)
.Subscribe(async x =>
{
if (!GetDataIsrunning) await GetData();
});
}
public bool GetDataIsrunning;
private async Task GetData()
{
if (GetDataIsrunning) return;
Debug.WriteLine($"Started @ {DateTime.Now}");
try
{
GetDataIsrunning = true;
await UpdateGridsAsync();
}
finally
{
GetDataIsrunning = false;
}
}Subscribe does not support async calls. This will run under fire-and-forget mechanism.https://stackoverflow.com/a/35190003/402027
Repeat() after that.
public class EventServer
{
public static IObservable<List<Event>> EventsList()
{
var waitTime = TimeSpan.FromSeconds(5);
var exceptionwaitTime = TimeSpan.FromSeconds(10);
return Observable.FromAsync(EventsTableQuery)
.Delay(waitTime)
.RetryWhen(onException => onException.Delay(exceptionwaitTime))
.Repeat();
}
private static async Task<List<Event>> EventsTableQuery()
{
var result =
await Task.Run(() =>
{
var returnList = new List<Event>();
for (int i = 1; i < 5; i++)
{
returnList.Add(new Event()
{ID = i, Date = DateTime.Now, Message = $"Test{i}"});
}
return returnList;
});
return result;
}
}
public class Event
{
public int ID { get; set; }
public DateTime Date { get; set; }
public string Message { get; set; }
}@davi1td yes that works, but this delays the data after it has been generated. You could reverse that with:
Observable.Timer(waitTime)
.SelectMany(_ => Observable.FromAsync(EventsTableQuery))
.RetryWhen(onException => onException.Delay(exceptionwaitTime))
.Repeat();Or if you want to stay closer to what you had:
Observable.FromAsync(EventsTableQuery)
.RetryWhen(onException => onException.Delay(exceptionwaitTime))
.Concat(Observable.Empty().Delay(waitTime))
.Repeat();(I inverted the retryWhen and the Delay in both cases, so you might need to extend exceptionwaitTime with waitTime to get your behavior)
@davi1td What i meant is that in your code, the delay for exceptions is waitTime + exceptionwaitTime, so 15 seconds. In mine it's only exceptionwaitTime, so 10 seconds :)
As for the many options: Yes Rx surely is incredible flexible, there are just so many ways to set things up and methods generalize so well many can give you the same properties if you combine them correctly. Some years ago (whoa, last edit 2017), i tried to recreate a bunch of RxJs method using a small subset of them, and i got quite a few :-D
@Dorus Love that Gist :)
So this is what I have now, both run exactly the same, 10 sec on retry
return Observable.Timer(TimeSpan.FromSeconds(0), waitTime)
.SelectMany(_ => Observable.FromAsync(x => BoschEventsTableQueryAsync()))
.RetryWhen(onException => onException.Delay(exceptionwaitTime))
.Repeat();
return Observable.FromAsync(_ => BoschEventsTableQueryAsync())
.RetryWhen(onException => onException.Delay(exceptionwaitTime))
.Concat(Observable.Empty<List<BoschEvent>>().Delay(waitTime))
.Repeat();Is one preferred over the other? Something I don't see?
Thanks again ;)
So in the below code, while swallowing all errors is the desired flow, I'm now trying to bubble up separately for other observers to see via the Subject:EventErrors and exposing with MyErrorListener observable , is this OK or is there a better way to do this?
private static readonly Subject<Exception> EventErrors = new();
public static IObservable<Exception> MyErrorListener = EventErrors.AsObservable();
Observable.FromAsync(EventsTableQuery)
.RetryWhen(onException => onException.Delay(exceptionwaitTime).Do(ex =>EventErrors.OnNext(new IOException())))
.Concat(Observable.Empty().Delay(waitTime))
.Repeat();Thanks -TD
OnNext() before the observable is allowed to start blocking for the next item...
- think in advance how 'bounded' your queue should be - if data comes in consistently faster than you consume, you'll eventually run out of memory. you need smth like a 'drop' strategy...
- Is consuming the I/O as quickly as you can absolutely necessary? I've learned that instead of 'reading as fast as I can, worry about consuming later' is not always better than 'consuming as fast as I can' => the latter can be achieved by eg. letting the consumer enumerating the I/O (sync or async) instead of observing I/O. "let the consumer pull" instead of "push to the consumer".
but only if that scheduler is applied after the producer
Not sure what you mean by that, in fact
What I meant here was that even when I specifically ask the ThreadPoolScheduler to schedule events, OnNext() continues to block. Eg this blocks -
return Observable.Create<int>(o =>
{
return ThreadPoolScheduler.Instance.Schedule(1, (ii, recurse) =>
{
Thread.Sleep(TimeSpan.FromSeconds(1));
o.OnNext(ii);
recurse(ii + 1);
});
});
OnNext() blocking causing the loss of clock ticks, which I'd like to avoid. Uncommenting the ObserveOn() call is the only way I've found to solve it, but I'd love to know if there is a more nuanced approach I am missing:using System;
using System.Reactive.Concurrency;
using System.Reactive.Linq;
using System.Threading;
namespace rx
{
internal static class Program
{
private static void Main()
{
using var subscription = TickProducer()
//.ObserveOn(ThreadPoolScheduler.Instance)
.Select(Transform)
.Subscribe(Consume);
Thread.Sleep(10000);
}
private static IObservable<int> TickProducer()
{
return Observable.Create<int>(o =>
{
var stopping = false;
var thread = new Thread(() =>
{
var started = DateTime.UtcNow;
var lastTick = 0;
while (!stopping)
{
var seconds = (int) DateTime.UtcNow.Subtract(started).TotalSeconds;
if (seconds != lastTick)
{
o.OnNext(lastTick = seconds);
}
}
});
thread.Start();
return () =>
{
stopping = true;
thread.Join();
};
});
}
private static int Transform(int counter)
{
Thread.Sleep(TimeSpan.FromSeconds(0.5));
return -counter;
}
private static void Consume(int counter)
{
Console.WriteLine(counter);
if (new Random().Next(3) == 0)
Thread.Sleep(TimeSpan.FromSeconds(2)); // Occasional delay
}
}
}