446 lines
17 KiB
C#
446 lines
17 KiB
C#
// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information.
|
|
|
|
using System.Collections.Generic;
|
|
using System.Linq;
|
|
using System.Reactive.Concurrency;
|
|
using System.Reactive.Disposables;
|
|
using System.Reactive.Subjects;
|
|
|
|
namespace System.Reactive.Linq
|
|
{
|
|
internal class QueryLanguageEx : IQueryLanguageEx
|
|
{
|
|
#region Create
|
|
|
|
public virtual IObservable<TResult> Create<TResult>(Func<IObserver<TResult>, IEnumerable<IObservable<object>>> iteratorMethod)
|
|
{
|
|
return new AnonymousObservable<TResult>(observer =>
|
|
iteratorMethod(observer).Concat().Subscribe(_ => { }, observer.OnError, observer.OnCompleted));
|
|
}
|
|
|
|
public virtual IObservable<Unit> Create(Func<IEnumerable<IObservable<object>>> iteratorMethod)
|
|
{
|
|
return new AnonymousObservable<Unit>(observer =>
|
|
iteratorMethod().Concat().Subscribe(_ => { }, observer.OnError, observer.OnCompleted));
|
|
}
|
|
|
|
#endregion
|
|
|
|
#region Expand
|
|
|
|
public virtual IObservable<TSource> Expand<TSource>(IObservable<TSource> source, Func<TSource, IObservable<TSource>> selector, IScheduler scheduler)
|
|
{
|
|
return new AnonymousObservable<TSource>(observer =>
|
|
{
|
|
var outGate = new object();
|
|
var q = new Queue<IObservable<TSource>>();
|
|
var m = new SerialDisposable();
|
|
var d = new CompositeDisposable { m };
|
|
var activeCount = 0;
|
|
var isAcquired = false;
|
|
|
|
var ensureActive = default(Action);
|
|
|
|
ensureActive = () =>
|
|
{
|
|
var isOwner = false;
|
|
|
|
lock (q)
|
|
{
|
|
if (q.Count > 0)
|
|
{
|
|
isOwner = !isAcquired;
|
|
isAcquired = true;
|
|
}
|
|
}
|
|
|
|
if (isOwner)
|
|
{
|
|
m.Disposable = scheduler.Schedule(self =>
|
|
{
|
|
var work = default(IObservable<TSource>);
|
|
|
|
lock (q)
|
|
{
|
|
if (q.Count > 0)
|
|
work = q.Dequeue();
|
|
else
|
|
{
|
|
isAcquired = false;
|
|
return;
|
|
}
|
|
}
|
|
|
|
var m1 = new SingleAssignmentDisposable();
|
|
d.Add(m1);
|
|
m1.Disposable = work.Subscribe(
|
|
x =>
|
|
{
|
|
lock (outGate)
|
|
observer.OnNext(x);
|
|
|
|
var result = default(IObservable<TSource>);
|
|
try
|
|
{
|
|
result = selector(x);
|
|
}
|
|
catch (Exception exception)
|
|
{
|
|
lock (outGate)
|
|
observer.OnError(exception);
|
|
}
|
|
|
|
lock (q)
|
|
{
|
|
q.Enqueue(result);
|
|
activeCount++;
|
|
}
|
|
|
|
ensureActive();
|
|
},
|
|
exception =>
|
|
{
|
|
lock (outGate)
|
|
observer.OnError(exception);
|
|
},
|
|
() =>
|
|
{
|
|
d.Remove(m1);
|
|
|
|
var done = false;
|
|
lock (q)
|
|
{
|
|
activeCount--;
|
|
if (activeCount == 0)
|
|
done = true;
|
|
}
|
|
if (done)
|
|
lock (outGate)
|
|
observer.OnCompleted();
|
|
});
|
|
self();
|
|
});
|
|
}
|
|
};
|
|
|
|
lock (q)
|
|
{
|
|
q.Enqueue(source);
|
|
activeCount++;
|
|
}
|
|
ensureActive();
|
|
|
|
return d;
|
|
});
|
|
}
|
|
|
|
public virtual IObservable<TSource> Expand<TSource>(IObservable<TSource> source, Func<TSource, IObservable<TSource>> selector)
|
|
{
|
|
return source.Expand(selector, SchedulerDefaults.Iteration);
|
|
}
|
|
|
|
#endregion
|
|
|
|
#region ForkJoin
|
|
|
|
public virtual IObservable<TResult> ForkJoin<TFirst, TSecond, TResult>(IObservable<TFirst> first, IObservable<TSecond> second, Func<TFirst, TSecond, TResult> resultSelector)
|
|
{
|
|
return Combine<TFirst, TSecond, TResult>(first, second, (observer, leftSubscription, rightSubscription) =>
|
|
{
|
|
var leftStopped = false;
|
|
var rightStopped = false;
|
|
var hasLeft = false;
|
|
var hasRight = false;
|
|
var lastLeft = default(TFirst);
|
|
var lastRight = default(TSecond);
|
|
|
|
return new BinaryObserver<TFirst, TSecond>(
|
|
left =>
|
|
{
|
|
switch (left.Kind)
|
|
{
|
|
case NotificationKind.OnNext:
|
|
hasLeft = true;
|
|
lastLeft = left.Value;
|
|
break;
|
|
case NotificationKind.OnError:
|
|
rightSubscription.Dispose();
|
|
observer.OnError(left.Exception);
|
|
break;
|
|
case NotificationKind.OnCompleted:
|
|
leftStopped = true;
|
|
if (rightStopped)
|
|
{
|
|
if (!hasLeft)
|
|
observer.OnCompleted();
|
|
else if (!hasRight)
|
|
observer.OnCompleted();
|
|
else
|
|
{
|
|
TResult result;
|
|
try
|
|
{
|
|
result = resultSelector(lastLeft, lastRight);
|
|
}
|
|
catch (Exception exception)
|
|
{
|
|
observer.OnError(exception);
|
|
return;
|
|
}
|
|
observer.OnNext(result);
|
|
observer.OnCompleted();
|
|
}
|
|
}
|
|
break;
|
|
}
|
|
},
|
|
right =>
|
|
{
|
|
switch (right.Kind)
|
|
{
|
|
case NotificationKind.OnNext:
|
|
hasRight = true;
|
|
lastRight = right.Value;
|
|
break;
|
|
case NotificationKind.OnError:
|
|
leftSubscription.Dispose();
|
|
observer.OnError(right.Exception);
|
|
break;
|
|
case NotificationKind.OnCompleted:
|
|
rightStopped = true;
|
|
if (leftStopped)
|
|
{
|
|
if (!hasLeft)
|
|
observer.OnCompleted();
|
|
else if (!hasRight)
|
|
observer.OnCompleted();
|
|
else
|
|
{
|
|
TResult result;
|
|
try
|
|
{
|
|
result = resultSelector(lastLeft, lastRight);
|
|
}
|
|
catch (Exception exception)
|
|
{
|
|
observer.OnError(exception);
|
|
return;
|
|
}
|
|
observer.OnNext(result);
|
|
observer.OnCompleted();
|
|
}
|
|
}
|
|
break;
|
|
}
|
|
});
|
|
});
|
|
}
|
|
|
|
public virtual IObservable<TSource[]> ForkJoin<TSource>(params IObservable<TSource>[] sources)
|
|
{
|
|
return sources.ForkJoin();
|
|
}
|
|
|
|
public virtual IObservable<TSource[]> ForkJoin<TSource>(IEnumerable<IObservable<TSource>> sources)
|
|
{
|
|
return new AnonymousObservable<TSource[]>(subscriber =>
|
|
{
|
|
var allSources = sources.ToArray();
|
|
var count = allSources.Length;
|
|
|
|
if (count == 0)
|
|
{
|
|
subscriber.OnCompleted();
|
|
return Disposable.Empty;
|
|
}
|
|
|
|
var group = new CompositeDisposable(allSources.Length);
|
|
var gate = new object();
|
|
|
|
var finished = false;
|
|
var hasResults = new bool[count];
|
|
var hasCompleted = new bool[count];
|
|
var results = new List<TSource>(count);
|
|
|
|
lock (gate)
|
|
{
|
|
for (var index = 0; index < count; index++)
|
|
{
|
|
var currentIndex = index;
|
|
var source = allSources[index];
|
|
results.Add(default(TSource));
|
|
group.Add(source.Subscribe(
|
|
value =>
|
|
{
|
|
lock (gate)
|
|
{
|
|
if (!finished)
|
|
{
|
|
hasResults[currentIndex] = true;
|
|
results[currentIndex] = value;
|
|
}
|
|
}
|
|
},
|
|
error =>
|
|
{
|
|
lock (gate)
|
|
{
|
|
finished = true;
|
|
subscriber.OnError(error);
|
|
group.Dispose();
|
|
}
|
|
},
|
|
() =>
|
|
{
|
|
lock (gate)
|
|
{
|
|
if (!finished)
|
|
{
|
|
if (!hasResults[currentIndex])
|
|
{
|
|
subscriber.OnCompleted();
|
|
return;
|
|
}
|
|
hasCompleted[currentIndex] = true;
|
|
foreach (var completed in hasCompleted)
|
|
{
|
|
if (!completed)
|
|
return;
|
|
}
|
|
finished = true;
|
|
subscriber.OnNext(results.ToArray());
|
|
subscriber.OnCompleted();
|
|
}
|
|
}
|
|
}));
|
|
}
|
|
}
|
|
return group;
|
|
});
|
|
}
|
|
|
|
#endregion
|
|
|
|
#region Let
|
|
|
|
public virtual IObservable<TResult> Let<TSource, TResult>(IObservable<TSource> source, Func<IObservable<TSource>, IObservable<TResult>> function)
|
|
{
|
|
return function(source);
|
|
}
|
|
|
|
#endregion
|
|
|
|
#region ManySelect
|
|
|
|
public virtual IObservable<TResult> ManySelect<TSource, TResult>(IObservable<TSource> source, Func<IObservable<TSource>, TResult> selector)
|
|
{
|
|
return ManySelect(source, selector, DefaultScheduler.Instance);
|
|
}
|
|
|
|
public virtual IObservable<TResult> ManySelect<TSource, TResult>(IObservable<TSource> source, Func<IObservable<TSource>, TResult> selector, IScheduler scheduler)
|
|
{
|
|
return Observable.Defer<TResult>(() =>
|
|
{
|
|
var chain = default(ChainObservable<TSource>);
|
|
|
|
return source
|
|
.Select(
|
|
x =>
|
|
{
|
|
var curr = new ChainObservable<TSource>(x);
|
|
|
|
if (chain != null)
|
|
chain.OnNext(curr);
|
|
chain = curr;
|
|
|
|
return (IObservable<TSource>)curr;
|
|
})
|
|
.Do(
|
|
_ => { },
|
|
exception =>
|
|
{
|
|
if (chain != null)
|
|
chain.OnError(exception);
|
|
},
|
|
() =>
|
|
{
|
|
if (chain != null)
|
|
chain.OnCompleted();
|
|
})
|
|
.ObserveOn(scheduler)
|
|
.Select(selector);
|
|
});
|
|
}
|
|
|
|
class ChainObservable<T> : ISubject<IObservable<T>, T>
|
|
{
|
|
T head;
|
|
AsyncSubject<IObservable<T>> tail = new AsyncSubject<IObservable<T>>();
|
|
|
|
public ChainObservable(T head)
|
|
{
|
|
this.head = head;
|
|
}
|
|
|
|
public IDisposable Subscribe(IObserver<T> observer)
|
|
{
|
|
var g = new CompositeDisposable();
|
|
g.Add(CurrentThreadScheduler.Instance.Schedule(() =>
|
|
{
|
|
observer.OnNext(head);
|
|
g.Add(tail.Merge().Subscribe(observer));
|
|
}));
|
|
return g;
|
|
}
|
|
|
|
public void OnCompleted()
|
|
{
|
|
OnNext(Observable.Empty<T>());
|
|
}
|
|
|
|
public void OnError(Exception error)
|
|
{
|
|
OnNext(Observable.Throw<T>(error));
|
|
}
|
|
|
|
public void OnNext(IObservable<T> value)
|
|
{
|
|
tail.OnNext(value);
|
|
tail.OnCompleted();
|
|
}
|
|
}
|
|
|
|
#endregion
|
|
|
|
#region ToListObservable
|
|
|
|
public virtual ListObservable<TSource> ToListObservable<TSource>(IObservable<TSource> source)
|
|
{
|
|
return new ListObservable<TSource>(source);
|
|
}
|
|
|
|
#endregion
|
|
|
|
#region |> Helpers <|
|
|
|
|
private static IObservable<TResult> Combine<TLeft, TRight, TResult>(IObservable<TLeft> leftSource, IObservable<TRight> rightSource, Func<IObserver<TResult>, IDisposable, IDisposable, IObserver<Either<Notification<TLeft>, Notification<TRight>>>> combinerSelector)
|
|
{
|
|
return new AnonymousObservable<TResult>(observer =>
|
|
{
|
|
var leftSubscription = new SingleAssignmentDisposable();
|
|
var rightSubscription = new SingleAssignmentDisposable();
|
|
|
|
var combiner = combinerSelector(observer, leftSubscription, rightSubscription);
|
|
var gate = new object();
|
|
|
|
leftSubscription.Disposable = leftSource.Materialize().Select(x => Either<Notification<TLeft>, Notification<TRight>>.CreateLeft(x)).Synchronize(gate).Subscribe(combiner);
|
|
rightSubscription.Disposable = rightSource.Materialize().Select(x => Either<Notification<TLeft>, Notification<TRight>>.CreateRight(x)).Synchronize(gate).Subscribe(combiner);
|
|
|
|
return new CompositeDisposable(leftSubscription, rightSubscription);
|
|
});
|
|
}
|
|
|
|
#endregion
|
|
}
|
|
}
|