forked from ThatRendle/Simple.Data
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathSimpleObservable.cs
More file actions
112 lines (95 loc) · 3.22 KB
/
SimpleObservable.cs
File metadata and controls
112 lines (95 loc) · 3.22 KB
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
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
namespace Simple.Data
{
public static class ColdObservable
{
public static IObservable<T> Create<T>(Func<IObserver<T>, IDisposable> func)
{
return new ColdObservable<T>(func);
}
public static IObservable<T> ToObservable<T>(this IEnumerable<T> source)
{
return Create<T>(o =>
{
try
{
foreach (var item in source)
{
o.OnNext(item);
}
o.OnCompleted();
}
catch (Exception ex)
{
o.OnError(ex);
}
return EmptyDisposable;
});
}
public static readonly IDisposable EmptyDisposable = new _EmptyDisposable();
private class _EmptyDisposable : IDisposable
{
public void Dispose()
{
}
}
}
internal class ColdObservable<T> : IObservable<T>
{
private readonly Func<IObserver<T>, IDisposable> _func;
public ColdObservable(Func<IObserver<T>, IDisposable> func)
{
_func = func;
}
public IDisposable Subscribe(IObserver<T> observer)
{
return _func(observer);
}
}
static class MapObservable
{
public static IObservable<TOut> Map<TIn,TOut>(this IObservable<TIn> source, Func<TIn,TOut> mapFunc)
{
return new MapObservable<TIn, TOut>(source, mapFunc);
}
}
class MapObservable<TIn, TOut> : IObservable<TOut>
{
private readonly IObservable<TIn> _source;
private readonly Func<TIn, TOut> _map;
public MapObservable(IObservable<TIn> source, Func<TIn,TOut> map)
{
_source = source;
_map = map;
}
public IDisposable Subscribe(IObserver<TOut> observer)
{
return _source.Subscribe(new MapObserver(observer, _map));
}
private class MapObserver : IObserver<TIn>
{
private readonly IObserver<TOut> _observer;
private readonly Func<TIn, TOut> _map;
public MapObserver(IObserver<TOut> observer, Func<TIn, TOut> map)
{
_observer = observer;
_map = map;
}
public void OnNext(TIn value)
{
_observer.OnNext(_map(value));
}
public void OnError(Exception error)
{
_observer.OnError(error);
}
public void OnCompleted()
{
_observer.OnCompleted();
}
}
}
}