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
|
// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information.
#if HAS_WINRT
using System.Reactive.Disposables;
using System.Reactive.Subjects;
using Windows.Foundation;
namespace System.Reactive.Windows.Foundation
{
class AsyncInfoToObservableBridge<TResult, TProgress> : ObservableBase<TResult>
{
private readonly Action<IAsyncInfo, Action<IAsyncInfo, AsyncStatus>> _onCompleted;
private readonly Func<IAsyncInfo, TResult> _getResult;
private readonly AsyncSubject<TResult> _subject;
public AsyncInfoToObservableBridge(IAsyncInfo info, Action<IAsyncInfo, Action<IAsyncInfo, AsyncStatus>> onCompleted, Func<IAsyncInfo, TResult> getResult, Action<IAsyncInfo, Action<IAsyncInfo, TProgress>> onProgress, IProgress<TProgress> progress, bool multiValue)
{
_onCompleted = onCompleted;
_getResult = getResult;
_subject = new AsyncSubject<TResult>();
if (onProgress != null)
{
onProgress(info, (iai, p) =>
{
if (multiValue && getResult != null)
_subject.OnNext(getResult(iai));
if (progress != null)
progress.Report(p);
});
}
Done(info, info.Status, true);
}
private void Done(IAsyncInfo info, AsyncStatus status, bool initial)
{
var error = default(Exception);
var result = default(TResult);
//
// Initial interactions with the IAsyncInfo object. Those could fail, which indicates
// a rogue implementation. Failure is just propagated out.
//
switch (status)
{
case AsyncStatus.Error:
error = info.ErrorCode;
if (error == null)
throw new InvalidOperationException("The asynchronous operation failed with a null error code.");
break;
case AsyncStatus.Canceled:
error = new OperationCanceledException();
break;
case AsyncStatus.Completed:
if (_getResult != null)
result = _getResult(info);
break;
default:
if (!initial)
throw new InvalidOperationException("The asynchronous operation completed unexpectedly.");
_onCompleted(info, (iai, s) => Done(iai, s, false));
return;
}
//
// Close as early as possible, before running continuations which could fail. In case of
// failure above, we don't close out the object in order to allow for debugging of the
// rogue implementation without losing state prematurely. Notice _getResults is merely
// an indirect call to the appropriate GetResults method, which is not supposed to throw.
// Instead, an Error status should be returned.
//
info.Close();
//
// Now we run the continuations, which could take a long time. Failure here is catastrophic
// and under control of the upstream subscriber.
//
if (error != null)
{
_subject.OnError(error);
}
else
{
if (_getResult != null)
_subject.OnNext(result);
_subject.OnCompleted();
}
}
protected override IDisposable SubscribeCore(IObserver<TResult> observer)
{
return _subject.Subscribe(observer);
}
}
}
#endif
|
