/
CatchScheduler.cs
143 lines (122 loc) · 5 KB
/
CatchScheduler.cs
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the Apache 2.0 License.
// See the LICENSE file in the project root for more information.
using System.Reactive.Disposables;
using System.Runtime.CompilerServices;
namespace System.Reactive.Concurrency
{
internal sealed class CatchScheduler<TException> : SchedulerWrapper
where TException : Exception
{
private readonly Func<TException, bool> _handler;
public CatchScheduler(IScheduler scheduler, Func<TException, bool> handler)
: base(scheduler)
{
_handler = handler;
}
protected override Func<IScheduler, TState, IDisposable> Wrap<TState>(Func<IScheduler, TState, IDisposable> action)
{
return (self, state) =>
{
try
{
return action(GetRecursiveWrapper(self), state);
}
catch (TException exception) when (_handler(exception))
{
return Disposable.Empty;
}
};
}
public CatchScheduler(IScheduler scheduler, Func<TException, bool> handler, ConditionalWeakTable<IScheduler, IScheduler> cache)
: base(scheduler, cache)
{
_handler = handler;
}
protected override SchedulerWrapper Clone(IScheduler scheduler, ConditionalWeakTable<IScheduler, IScheduler> cache)
{
return new CatchScheduler<TException>(scheduler, _handler, cache);
}
protected override bool TryGetService(IServiceProvider provider, Type serviceType, out object service)
{
service = provider.GetService(serviceType);
if (service != null)
{
if (serviceType == typeof(ISchedulerLongRunning))
{
service = new CatchSchedulerLongRunning((ISchedulerLongRunning)service, _handler);
}
else if (serviceType == typeof(ISchedulerPeriodic))
{
service = new CatchSchedulerPeriodic((ISchedulerPeriodic)service, _handler);
}
}
return true;
}
private class CatchSchedulerLongRunning : ISchedulerLongRunning
{
private readonly ISchedulerLongRunning _scheduler;
private readonly Func<TException, bool> _handler;
public CatchSchedulerLongRunning(ISchedulerLongRunning scheduler, Func<TException, bool> handler)
{
_scheduler = scheduler;
_handler = handler;
}
public IDisposable ScheduleLongRunning<TState>(TState state, Action<TState, ICancelable> action)
{
return _scheduler.ScheduleLongRunning(
(scheduler: this, action, state),
(tuple, cancel) =>
{
try
{
tuple.action(tuple.state, cancel);
}
catch (TException exception) when (tuple.scheduler._handler(exception))
{
}
});
}
}
private sealed class CatchSchedulerPeriodic : ISchedulerPeriodic
{
private readonly ISchedulerPeriodic _scheduler;
private readonly Func<TException, bool> _handler;
public CatchSchedulerPeriodic(ISchedulerPeriodic scheduler, Func<TException, bool> handler)
{
_scheduler = scheduler;
_handler = handler;
}
public IDisposable SchedulePeriodic<TState>(TState state, TimeSpan period, Func<TState, TState> action)
{
var failed = false;
var d = new SingleAssignmentDisposable();
d.Disposable = _scheduler.SchedulePeriodic(state, period, state_ =>
{
//
// Cancellation may not be granted immediately; prevent from running user
// code in that case. Periodic schedulers are assumed to introduce some
// degree of concurrency, so we should return from the SchedulePeriodic
// call eventually, allowing the d.Dispose() call in the catch block to
// take effect.
//
if (failed)
return default(TState);
try
{
return action(state_);
}
catch (TException exception)
{
failed = true;
if (!_handler(exception))
throw;
d.Dispose();
return default(TState);
}
});
return d;
}
}
}
}