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
144
145
146
147
148
149
150
151
|
// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information.
using System;
using System.Reactive.Disposables;
#if !NO_WEAKTABLE
using System.Runtime.CompilerServices;
#endif
namespace System.Reactive.Concurrency
{
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)
{
if (!_handler(exception))
throw;
return Disposable.Empty;
}
};
}
#if !NO_WEAKTABLE
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);
}
#else
protected override SchedulerWrapper Clone(IScheduler scheduler)
{
return new CatchScheduler<TException>(scheduler, _handler);
}
#endif
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;
}
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(state, (state_, cancel) =>
{
try
{
action(state_, cancel);
}
catch (TException exception)
{
if (!_handler(exception))
throw;
}
});
}
}
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;
}
}
}
}
|
