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
|
// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information.
#if !NO_EXPRESSIONS
#pragma warning disable 1591
using System.Collections.Generic;
using System.Linq;
using System.Linq.Expressions;
using System.Reflection;
using System.Reactive.Concurrency;
namespace System.Reactive.Linq
{
/// <summary>
/// Provides a set of static methods for writing queries over observable sequences, allowing translation to a target query language.
/// </summary>
public static partial class Qbservable
{
/// <summary>
/// Returns the input typed as an IObservable<TSource>.
/// This operator is used to separate the part of the query that's captured as an expression tree from the part that's executed locally.
/// </summary>
/// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
/// <param name="source">An IQbservable<TSource> sequence to convert to an IObservable<TSource> sequence.</param>
/// <returns>The original source object, but typed as an IObservable<TSource>.</returns>
/// <exception cref="ArgumentNullException"><paramref name="source"/> is null.</exception>
public static IObservable<TSource> AsObservable<TSource>(this IQbservable<TSource> source)
{
if (source == null)
throw new ArgumentNullException("source");
return source;
}
/// <summary>
/// Converts an enumerable sequence to an observable sequence.
/// </summary>
/// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
/// <param name="source">Enumerable sequence to convert to an observable sequence.</param>
/// <returns>The observable sequence whose elements are pulled from the given enumerable sequence.</returns>
/// <exception cref="ArgumentNullException"><paramref name="source"/> is null.</exception>
/// <remarks>This operator requires the source's <see cref="IQueryProvider"/> object (see <see cref="IQueryable.Provider"/>) to implement <see cref="IQbservableProvider"/>.</remarks>
public static IQbservable<TSource> ToQbservable<TSource>(this IQueryable<TSource> source)
{
if (source == null)
throw new ArgumentNullException("source");
return ((IQbservableProvider)source.Provider).CreateQuery<TSource>(
Expression.Call(
null,
#if CRIPPLED_REFLECTION
InfoOf(() => Qbservable.ToQbservable<TSource>(default(IQueryable<TSource>))),
#else
((MethodInfo)MethodInfo.GetCurrentMethod()).MakeGenericMethod(typeof(TSource)),
#endif
source.Expression
)
);
}
/// <summary>
/// Converts an enumerable sequence to an observable sequence, using the specified scheduler to run the enumeration loop.
/// </summary>
/// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
/// <param name="source">Enumerable sequence to convert to an observable sequence.</param>
/// <param name="scheduler">Scheduler to run the enumeration of the input sequence on.</param>
/// <returns>The observable sequence whose elements are pulled from the given enumerable sequence.</returns>
/// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="scheduler"/> is null.</exception>
/// <remarks>This operator requires the source's <see cref="IQueryProvider"/> object (see <see cref="IQueryable.Provider"/>) to implement <see cref="IQbservableProvider"/>.</remarks>
public static IQbservable<TSource> ToQbservable<TSource>(this IQueryable<TSource> source, IScheduler scheduler)
{
if (source == null)
throw new ArgumentNullException("source");
if (scheduler == null)
throw new ArgumentNullException("scheduler");
return ((IQbservableProvider)source.Provider).CreateQuery<TSource>(
Expression.Call(
null,
#if CRIPPLED_REFLECTION
InfoOf(() => Qbservable.ToQbservable<TSource>(default(IQueryable<TSource>))),
#else
((MethodInfo)MethodInfo.GetCurrentMethod()).MakeGenericMethod(typeof(TSource)),
#endif
source.Expression,
Expression.Constant(scheduler)
)
);
}
internal static Expression GetSourceExpression<TSource>(IObservable<TSource> source)
{
var q = source as IQbservable<TSource>;
if (q != null)
return q.Expression;
return Expression.Constant(source, typeof(IObservable<TSource>));
}
internal static Expression GetSourceExpression<TSource>(IEnumerable<TSource> source)
{
var q = source as IQueryable<TSource>;
if (q != null)
return q.Expression;
return Expression.Constant(source, typeof(IEnumerable<TSource>));
}
internal static Expression GetSourceExpression<TSource>(IObservable<TSource>[] sources)
{
return Expression.NewArrayInit(
typeof(IObservable<TSource>),
sources.Select(source => GetSourceExpression(source))
);
}
internal static Expression GetSourceExpression<TSource>(IEnumerable<TSource>[] sources)
{
return Expression.NewArrayInit(
typeof(IEnumerable<TSource>),
sources.Select(source => GetSourceExpression(source))
);
}
internal static MethodInfo InfoOf<R>(Expression<Func<R>> f)
{
return ((MethodCallExpression)f.Body).Method;
}
}
}
#pragma warning restore 1591
#endif
|
