dotnet / reactive Goto Github PK
View Code? Open in Web Editor NEWThe Reactive Extensions for .NET
Home Page: http://reactivex.io
License: MIT License
The Reactive Extensions for .NET
Home Page: http://reactivex.io
License: MIT License
Copied from https://rx.codeplex.com/workitem/10
Related discussion:
http://social.msdn.microsoft.com/Forums/en-US/rx/thread/1be3a07f-68c2-42a0-b5f5-ec3446f9f514
Add new properties to the Scheduler class (examples provided below) to simplify the choice for users, to hide internal optimization details and to make the semantics of choosing a scheduler reflect the actual underlying concerns of operators.
Furthermore, consider making existing platform-specific schedulers' internal (at least their parameterless constructors) and removing the "Instance" and "Default" factories for them; Alternatively, consider changing the behavior of operators so that when they receive a concrete scheduler no optimizations are applied and the specified scheduler is used directly, and only apply optimizations when a façade scheduler is passed in, avoiding the need for the DisableOptimizations operator.
PROPOSAL FOR COMPLETE LIST OF FACADES AND SCHEDULER CLASSES
NOTE: The schedulers suggested below were given precedence based on implementations for .NET 4.5 only; however, the suggested abstractions would allow Rx to internally choose the best scheduler for the job by considering the current platform.
• Scheduler.Synchronous = ImmediateScheduler
• Scheduler.Asynchronous
With precedence from left to right:
Standard = DispatcherScheduler / ControlScheduler -> SynchronizationContextScheduler -> CurrentThreadScheduler
ISchedulerLongRunning = NewThreadScheduler
ISchedulerPeriodic = DispatcherScheduler / ControlScheduler -> ThreadPoolScheduler
• Scheduler.Concurrent
With precedence from left to right:
Standard = ThreadPoolScheduler
ISchedulerLongRunning = NewThreadScheduler
ISchedulerPeriodic = ThreadPoolScheduler
• ControlScheduler: Useful for when thread-affinity is required; e.g., Scheduler.Asynchronous may choose NewThreadScheduler, so users must be able to call .ObserveOn(ControlScheduler) explicitly at the end of a query if it must notify observers on the UI thread.
• CurrentThreadScheduler: Useful for simple applications and experimentation.
• DispatcherScheduler: Useful for when thread-affinity is required; e.g., Scheduler.Asynchronous may choose NewThreadScheduler, so users must be able to call .ObserveOnDispatcher() explicitly at the end of a query if it must notify observers on the UI thread.
• EventLoopScheduler: Useful for dedicated work queues. Its use implies that a new reference must be passed to more than one operator, thus making it difficult or impossible to hide behind scheduler façades.
• HistoricalScheduler: Useful for instrumentation and diagnostics.
• VirtualTimeScheduler: Useful for testing and experimentation.
• TestScheduler: Useful for testing and experimentation.
Copied from https://rx.codeplex.com/workitem/79
Using version 2.2.4.0 with .net 4.5.1.
TestScheduler
throws a NullReferenceException when using it with multi-threaded access, however this is not in the TestScheduler
itself, but the VirtualTimeScheduler
.
Scenario: One thread is responsible for advancing the scheduler, the other threads are responsible for yielding and sleeping on that scheduler. A reproducible test environment, either in a unit test or a console app, shows that it happens irregularly and that the scheduler is not disposed when this happens.
It seems likely that this is related to #72, but as it occurs when using TestScheduler
and not VirtualTimeScheduler
I have logged it separately.
Stack trace below.
System.AggregateException: One or more errors occurred. ---> System.NullReferenceException: Object reference not set to an instance of an object.
Result StackTrace:
at System.Reactive.Concurrency.VirtualTimeScheduler`2.GetNext()
at System.Reactive.Concurrency.VirtualTimeSchedulerBase`2.AdvanceTo(TAbsolute time)
at System.Reactive.Concurrency.VirtualTimeSchedulerBase`2.AdvanceBy(TRelative time)
Copied from https://rx.codeplex.com/workitem/16
Consider creating factory methods that return IConnectableObservable<T>
similar to the Create
factory methods that return IObservable<T>
.
Related discussion and example code:
http://social.msdn.microsoft.com/Forums/en-US/rx/thread/4b89144c-045f-4497-919c-24fb6e1d1215/
Copied from https://rx.codeplex.com/workitem/13
Even though the documentation says period == 0 is supported:
/// <param name="period">Period to produce subsequent values. If this value is equal to TimeSpan.Zero, the timer will recur as fast as possible.</param>
Here is an example exception stack trace:
at System.Reactive.Concurrency.ConcurrencyAbstractionLayerImpl.StartPeriodicTimer(Action action, TimeSpan period)
at System.Reactive.Concurrency.DefaultScheduler.SchedulePeriodic[TState](TState state, TimeSpan period, Func`2 action)
at System.Reactive.Concurrency.Scheduler.SchedulePeriodic_[TState](IScheduler scheduler, TState state, TimeSpan period, Func`2 action)
at System.Reactive.Concurrency.Scheduler.SchedulePeriodic[TState](IScheduler scheduler, TState state, TimeSpan period, Func`2 action)
at System.Reactive.Linq.Observαble.Timer.π.Run()
at System.Reactive.Linq.Observαble.Timer.Run(IObserver`1 observer, IDisposable cancel, Action`1 setSink)
(remainder elided)
The issue seems to be this code in
System.Reactive.Concurrency.ConcurrencyAbstractionLayerImpl.StartPeriodicTimer(Action action, TimeSpan period):
//
// MSDN documentation states the following:
//
// "If period is zero (0) or negative one (-1) milliseconds and dueTime is positive, callback is invoked once;
// the periodic behavior of the timer is disabled, but can be re-enabled using the Change method."
//
if (period <= TimeSpan.Zero)
throw new ArgumentOutOfRangeException("period");
Even the comment seems to indicate Period==0 is allowed.
Copied from https://rx.codeplex.com/workitem/12
One of the overloads on IObservable.Subscribe is
public static IDisposable Subscribe(this IObservable source, Action onNext)
Internally, the unspecified onError parameter is set to Stubs.Throw which is a simple lambda that rethrows the exception passed in.
The code in Observer.OnError
public void OnError(Exception error)
{
foreach (var observer in _observers.Data)
observer.OnError(error);
}
Since Stubs.Throw
throws the exception, the exception passed up through the foreach loop and other observers in _observers.Data
never have their OnError
called. The exception itself is swallowed somewhere inside Rx.
It seems to me that either Observer<T>.OnError
should wrap observer.OnErro
r in a try-catch, or Stubs.Throw
should swallow the exception instead of throwing it. By not passing an onError parameter, the user of IObservable<T>.Subscribe
wishes the error to be ignored only for that Subscribe
. The other subscribers registered with their own onError callbacks should be unaffected.
I've worked around this for now by making a new extension method on IObservable<T>
which passes ex => { }
for the onError parameter to IObservable<T>.Subscribe
Copied from https://rx.codeplex.com/workitem/73
I hope nuget package support taget platform wpa81.
Copied from https://rx.codeplex.com/workitem/7
We use Windows 7 with 64 bit in a production computer. in RX1.1.11111.1 all was fine. when we migrated to RX 2.0.20823.2 we have encountered a random exception from the EventLoopScheduler
:
an Unhandled Exception occured in non UI thread.
System.NullReferenceException: Object reference not set to an instance of an object.
at System.Reactive.Concurrency.EventLoopScheduler.Run()
at System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean ignoreSyncCtx)
at System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback callback, Object state)
at System.Threading.ThreadHelper.ThreadStart()
Here's the gist of the code:
EventLoopScheduler m_scheduler = new EventLoopScheduler();
. . .
m_receivedMessageHandler.StatusReceived.ObserveOn(m_scheduler) .Subscribe(p_unit => sendAll(m_retransmitManager, m_endPoint));
We have seen an answer to our post which indicated it is a JIT problem in 64 bit.
Copied from https://rx.codeplex.com/workitem/71
I seem to have found a potential issue with the Buffer
extension method.
I would like to Buffer items coming in on a subject but trigger the Processing of those items either when a batch size is reached, or if a time interval expires. I also need the items to be processed on an individual thread so I am using the NewThreadScheduler()
.
I have written a simple class to demonstrate the issue.
public class Processor<T>
{
private readonly HashSet<int> _threadIds = new HashSet<int>();
private readonly Subject<T> _queue;
private readonly IDisposable _observation;
private int _count;
public Processor(TimeSpan interval, int batchSize)
{
_queue = new Subject<T>();
var scheduler = new NewThreadScheduler();
_observation = _queue.Buffer(interval, batchSize, scheduler).Subscribe(Process);
}
public IEnumerable<int> ProcessThreadIds
{
get
{
return _threadIds;
}
}
public int Count
{
get
{
return _count;
}
set
{
_count = value;
}
}
public void Stop()
{
_observation.Dispose();
}
public void Add(T item)
{
_queue.OnNext(item);
}
private void Process(IList<T> item)
{
_threadIds.Add(Thread.CurrentThread.ManagedThreadId);
Count += item.Count;
}
}
I have run this using the following code:
public static void Main(string[] args)
{
Console.WriteLine("Main Thread Id: {0}", Thread.CurrentThread.ManagedThreadId);
var processor = new Processor<int>(new TimeSpan(0, 0,0, 0, 500), 100);
var tasks = new List<Task>();
var random = new Random(7);
for (int i = 0; i < 10; i++)
{
var task = new Task(
() =>
{
for (int j = 0; j < 1000; j++)
{
processor.Add(j);
Thread.Sleep(random.Next(5, 10));
}
});
tasks.Add(task);
task.Start();
}
Task.WaitAll(tasks.ToArray());
while (processor.Count < 10000)
{
Thread.Sleep(100);
}
Console.WriteLine("Recorded the following Process thread ids: {0}", string.Join(",", processor.ProcessThreadIds.Distinct()));
processor.Stop();
Console.ReadKey();
}
I would have expect that _threadIds only ever contains 1 thread id. This is the case if I provide just an interval or just a batch size ... but if I include both, then _threadIds contains multiple thread ids.
Interestingly if I use ObserveOn
to provide the same scheduler, I still see the same issue (this may actually be more likely to be an issue).
_queue.ObserveOn(scheduler).Buffer(interval, batchSize).Subscribe(Process);
Can anyone explain what is happening? And if it is by design, how I may be able to achieve what I need?
Thanks
David
Copied from https://rx.codeplex.com/workitem/48
Hi,
I just ran into a problem related to the unexpected introduction of concurrency when using Task-based asynchrony in conjunction with Rx in a Windows 8 app. I've had this problem before and thought that I was doing something wrong or that it's the correct behavior of some kind of new multi-threaded UI model in WinRT, yet now I realize that it seems to be a bug in Rx.
If you search the entire Rx codebase for ".ContinueWith(" you'll see that every usage doesn't specify the ExecuteSynchronously flag, thus allowing the TPL to schedule the continuation on the thread pool, and it always does.
According to the documentation:
Creates a continuation that executes asynchronously when the target Task completes.
But we don't want asynchronous execution, we want synchronous execution. I'd imagine this is true for all uses of ContinueWith throughout Rx, though I haven't looked at the finer details yet.
Executing ContinueWith asynchronously is a problem in general because the concurrency being introduced isn't related to the specified IScheduler, if one is supplied by the user, and it's entirely unexpected when using particular operators that should never introduce concurrency by default, as shown below. As a result, the threading context of notifications is incorrect and we have little control over it.
There are two situations in which I've run into this problem in my code:
ConfigureAwait(false)
, yet it's lost with ToObservable. My expectation has always been that the observable returned by ToObservable simply generates a notification within the completion context of the converted Task, without introducing concurrency unnecessarily. (NOTE: I'm not actually sure whether forcing Rx's usage of ContinueWith to execute synchronously would solve this problem, though I expect that it will.)private async void _Loaded(object sender, RoutedEventArgs e)
{
Contract.Assert(Dispatcher.HasThreadAccess); // true
await DoSomethingAsync();
Contract.Assert(Dispatcher.HasThreadAccess); // true
DoSomethingAsync().ToObservable().Subscribe(_ =>
{
Contract.Assert(Dispatcher.HasThreadAccess); // false!
});
}
private async Task DoSomethingAsync()
{
await Task.Delay(TimeSpan.FromSeconds(1));
Contract.Assert(Dispatcher.HasThreadAccess); // true
}
Can anyone else confirm that this behavior is indeed a bug?
I'd be happy to fix it and have it ready in a day or two, before the next planned release hopefully, though it's also possibly a breaking change (for the better, I guess) so I'd like to get a green light from the Rx team before doing the work.
Thanks,
Dave
Hi,
raised this on CodePlex too some months ago, to no avail.
Before
https://github.com/Reactive-Extensions/Rx.NET/blob/a2a03840ac8417fec14e53cfc31adfd7fef1635f/Ix.NET/Source/System.Interactive.Async/AsyncEnumerable.Single.cs#L268
it is crucial that the inner enumerator instance is disposed. We've run into leaks repeatedly. The inner enumerator also leaks whenever the outer is disposed and the inner has not completed.
Thanks
Daniel
Copied from https://rx.codeplex.com/workitem/75
Method : System.Reactive.Linq.QueryLanguage.LastOrDefaultInternal
The ManualResetEvent
used to wait for the end of the sequence should be disposed after the call to WaitOne
Copied from https://rx.codeplex.com/workitem/37
EventLoopScheduler.Schedule may throw SemaphoreFullException. It appears that it may be due to a couple of race conditions in the Run method:
Related discussion:
http://social.msdn.microsoft.com/Forums/en-US/rx/thread/223f2447-4cb9-47ce-be21-6fd0e5b45e30/?prof=required
Copied from https://rx.codeplex.com/workitem/17
When System.Reactive.Concurrency.LocalScheduler
tries to schedule next long term event which is further than maximum interval (48 days or so, I believe), the underlying System.Threading.Timer
throws ArgumentOutOfRangeException
.
Our production system subscribes to rare economic events, sometimes two months into the future which causes this problem. This issue caused me earlier to re-implement own real-time event queue, for version 1.0 of Rx, but with your improvements in handling time drift and time adjustments in 2.0 it becomes less and less reasonable to keep doing so.
Since, I'd assume actual user troubles are inconsequential, I'd ask you to still consider resolving this for the sake of making sure the sample from Bart's article (http://blogs.msdn.com/b/rxteam/archive/2012/06/20/reactive-extensions-v2-0-release-candidate-available-now.aspx ) still works, once updated to his next birthday:
var birthday = new DateTime(2014, 2, 11, 0, 0, 0);
Observable.Timer(birthday).Subscribe(_ =>
{
var now = DateTime.Now;
var onTime = Math.Abs((now - birthday).TotalMilliseconds);
Console.WriteLine("Happy birthday! {0}", onTime < 1000 ? ":-)" : ":-(");
});
Copied from https://rx.codeplex.com/workitem/44
Several operators in Rx accept static parameters such as Int32 and TimeSpan. Given that Rx is a reactive framework, wouldn't it make sense to provide observable variants as well?
In many cases, it seems that this would help users avoid having to do complex things related to re-subscribing, publishing, scheduling and/or buffering when writing dynamic queries, if these operators were to offer reactive overloads that avoid re-subscription altogether.
In at least one known case it seems that it's absolutely required to prevent data loss: http://rx.codeplex.com/workitem/43
The new signatures in general would simply replace the static parameters of their corresponding overloads with IObservable<T>
, though perhaps in some cases due to overload conflicts a new naming convention for operators may be called for.
Note that this was already done for some overloads of Window, Buffer, Throttle, Sample, Delay and Timeout in Rx 2.0. Although, instead of replacing TimeSpan
parameters with IObservable<TimeSpan>
, an IObservable<TOther>
parameter was added instead because observable notifications carry time.
Furthermore, instead of overloading Skip and Take in Rx 1.0, different operators were introduced: SkipUntil and TakeUntil. It seems that the same basic principles apply.
These operators are potential candicates for overloading as of Rx 2.0:
Buffer<TSource>(this IObservable<TSource>, int);
Buffer<TSource>(this IObservable<TSource>, int, int);
Buffer<TSource>(this IObservable<TSource>, TimeSpan, int);
Buffer<TSource>(this IObservable<TSource>, TimeSpan, int, IScheduler);
Buffer<TSource>(this IObservable<TSource>, TimeSpan, TimeSpan);
Buffer<TSource>(this IObservable<TSource>, TimeSpan, TimeSpan, IScheduler)
DistinctUntilChanged<TSource>(this IObservable<TSource>, IEqualityComparer<TSource>)
Merge<TSource>(this IEnumerable<IObservable<TSource>>, int)
Merge<TSource>(this IEnumerable<IObservable<TSource>>, int, IScheduler)
Merge<TSource>(this IObservable<IObservable<TSource>>, int) (See the related work item above.)
ObserveOn<TSource>(this IObservable<TSource>, IScheduler)
Repeat<TSource>(this IObservable<TSource>, int)
Replay<TSource, TResult>(this IObservable<TSource>, Func<IObservable<TSource>,IObservable<TResult>>, int)
Replay<TSource, TResult>(this IObservable<TSource>, Func<IObservable<TSource>,IObservable<TResult>>, int, IScheduler)
Replay<TSource, TResult>(this IObservable<TSource>, Func<IObservable<TSource>,IObservable<TResult>>, int, TimeSpan)
Replay<TSource, TResult>(this IObservable<TSource>, Func<IObservable<TSource>,IObservable<TResult>>, int, TimeSpan, IScheduler)
Replay<TSource, TResult>(this IObservable<TSource>, Func<IObservable<TSource>,IObservable<TResult>>, TimeSpan)
Replay<TSource, TResult>(this IObservable<TSource>, Func<IObservable<TSource>,IObservable<TResult>>, TimeSpan, IScheduler)
Replay<TSource>(this IObservable<TSource>, int)
Replay<TSource>(this IObservable<TSource>, int, IScheduler)
Replay<TSource>(this IObservable<TSource>, int, TimeSpan)
Replay<TSource>(this IObservable<TSource>, int, TimeSpan, IScheduler)
Replay<TSource>(this IObservable<TSource>, TimeSpan)
Replay<TSource>(this IObservable<TSource>, TimeSpan, IScheduler)
Retry<TSource>(this IObservable<TSource>, int)
Skip<TSource>(this IObservable<TSource>, int)
SkipLast<TSource>(this IObservable<TSource>, int)
SkipLast<TSource>(this IObservable<TSource>, TimeSpan)
SkipLast<TSource>(this IObservable<TSource>, TimeSpan, IScheduler)
Take<TSource>(this IObservable<TSource>, int)
Take<TSource>(this IObservable<TSource>, int, IScheduler)
TakeLast<TSource>(this System.IObservable<TSource>, int)
TakeLast<TSource>(this IObservable<TSource>, int, IScheduler)
TakeLast<TSource>(this IObservable<TSource>, TimeSpan)
TakeLast<TSource>(this IObservable<TSource>, TimeSpan, IScheduler)
TakeLast<TSource>(this IObservable<TSource>, TimeSpan, IScheduler, IScheduler)
TakeLastBuffer<TSource>(this IObservable<TSource>, int)
TakeLastBuffer<TSource>(this IObservable<TSource>, TimeSpan)
TakeLastBuffer<TSource>(this IObservable<TSource>, TimeSpan, IScheduler)
Window<TSource>(this IObservable<TSource>, int)
Window<TSource>(this IObservable<TSource>, int, int)
Window<TSource>(this IObservable<TSource>, TimeSpan, int)
Window<TSource>(this IObservable<TSource>, TimeSpan, int, IScheduler)
Window<TSource>(this IObservable<TSource>, TimeSpan, TimeSpan)
Window<TSource>(this IObservable<TSource>, TimeSpan, TimeSpan, IScheduler)
Without giving much thought to semantics or to potential overload collisions, here are their coresponding reactive overloads:
Note: ObserveOn is the only operator that should have an observable IScheduler parameter?
Buffer<TSource>(this IObservable<TSource>, IObservable<int>);
Buffer<TSource>(this IObservable<TSource>, IObservable<int>, IObservable<int>);
Buffer<TSource, TDuration>(this IObservable<TSource>, IObservable<TDuration>, IObservable<int>);
Buffer<TSource, TDuration>(this IObservable<TSource>, IObservable<TDuration>, IObservable<int>, IScheduler);
Buffer<TSource, TDuration, TShift>(this IObservable<TSource>, IObservable<TDuration>, IObservable<TShift>);
Buffer<TSource, TDuration, TShift>(this IObservable<TSource>, IObservable<TDuration>, IObservable<TShift>, IScheduler)
DistinctUntilChanged<TSource>(this IObservable<TSource>, IObservable<IEqualityComparer<TSource>>)
Merge<TSource>(this IEnumerable<IObservable<TSource>>, IObservable<int>)
Merge<TSource>(this IEnumerable<IObservable<TSource>>, IObservable<int>, IScheduler)
Merge<TSource>(this IObservable<IObservable<TSource>>, IObservable<int>) (See the related work item above.)
ObserveOn<TSource>(this IObservable<TSource>, IObservable<IScheduler>)
Repeat<TSource>(this IObservable<TSource>, IObservable<int>)
Replay<TSource, TResult>(this IObservable<TSource>, Func<IObservable<TSource>,IObservable<TResult>>, IObservable<int>)
Replay<TSource, TResult>(this IObservable<TSource>, Func<IObservable<TSource>,IObservable<TResult>>, IObservable<int>, IScheduler)
Replay<TSource, TDuration, TResult>(this IObservable<TSource>, Func<IObservable<TSource>,IObservable<TResult>>, IObservable<int>, IObservable<TDuration>)
Replay<TSource, TDuration, TResult>(this IObservable<TSource>, Func<IObservable<TSource>,IObservable<TResult>>, IObservable<int>, IObservable<TDuration>, IScheduler)
Replay<TSource, TDuration, TResult>(this IObservable<TSource>, Func<IObservable<TSource>,IObservable<TResult>>, IObservable<TDuration>)
Replay<TSource, TDuration, TResult>(this IObservable<TSource>, Func<IObservable<TSource>,IObservable<TResult>>, IObservable<TDuration>, IScheduler)
Replay<TSource>(this IObservable<TSource>, IObservable<int>)
Replay<TSource>(this IObservable<TSource>, IObservable<int>, IScheduler)
Replay<TSource, TDuration>(this IObservable<TSource>, IObservable<int>, IObservable<TDuration>)
Replay<TSource, TDuration>(this IObservable<TSource>, IObservable<int>, IObservable<TDuration>, IScheduler)
Replay<TSource, TDuration>(this IObservable<TSource>, IObservable<TDuration>)
Replay<TSource, TDuration>(this IObservable<TSource>, IObservable<TDuration>, IScheduler)
Retry<TSource>(this IObservable<TSource>, IObservable<int>)
Skip<TSource>(this IObservable<TSource>, IObservable<int>)
SkipLast<TSource>(this IObservable<TSource>, IObservable<int>)
SkipLast<TSource, TDuration>(this IObservable<TSource>, IObservable<TDuration>)
SkipLast<TSource, TDuration>(this IObservable<TSource>, IObservable<TDuration>, IScheduler)
Take<TSource>(this IObservable<TSource>, IObservable<int>)
Take<TSource>(this IObservable<TSource>, IObservable<int>, IScheduler)
TakeLast<TSource>(this IObservable<TSource>, IObservable<int>)
TakeLast<TSource>(this IObservable<TSource>, IObservable<int>, IScheduler)
TakeLast<TSource, TDuration>(this IObservable<TSource>, IObservable<TDuration>)
TakeLast<TSource, TDuration>(this IObservable<TSource>, IObservable<TDuration>, IScheduler)
TakeLast<TSource, TDuration>(this IObservable<TSource>, IObservable<TDuration>, IScheduler, IScheduler)
TakeLastBuffer<TSource>(this IObservable<TSource>, IObservable<int>)
TakeLastBuffer<TSource, TDuration>(this IObservable<TSource>, IObservable<TDuration>)
TakeLastBuffer<TSource, TDuration>(this IObservable<TSource>, IObservable<TDuration>, IScheduler)
Window<TSource>(this IObservable<TSource>, IObservable<int>)
Window<TSource>(this IObservable<TSource>, IObservable<int>, IObservable<int>)
Window<TSource, TDuration>(this IObservable<TSource>, IObservable<TDuration>, IObservable<int>)
Window<TSource, TDuration>(this IObservable<TSource>, IObservable<TDuration>, IObservable<int>, IScheduler)
Window<TSource, TDuration>(this IObservable<TSource>, IObservable<TDuration>, IObservable<TDuration>)
Window<TSource, TDuration>(this IObservable<TSource>, IObservable<TDuration>, IObservable<TDuration>, IScheduler)
Copied from https://rx.codeplex.com/workitem/59
Is this behavior by design ?
Copied from https://rx.codeplex.com/workitem/54
Versions: 2.1.30214 and 2.2.1 beta.
The following code never terminates:
static void Main(string[] args)
{
Observable.Return(1).Repeat().Take(100).Subscribe(Console.WriteLine);
}
However, this works:
Observable
.Return(1, Scheduler.Default)
.Repeat()
.Take(100)
.Subscribe(Console.WriteLine);
The underlying issue is that the Repeat
(or Concat
) re-subscribes to the source Return(1)
before even able to return a disposable so the Take
can stop it. One possible solution is to sidestep the subscription in Concat
into another thread so it can return a disposable, something like:
Observable
.Return(1)
.SubscribeOn(NewThreadScheduler.Default)
.Repeat()
.Take(100)
.Subscribe(Console.WriteLine);
Copied from https://rx.codeplex.com/workitem/36
Implement an overloaded operator named AsObservablesWithPrecedence that accepts an ISubject and yields a list of observables with the specified subscription precedence, indicating the expected order of observations among the hot observables for a given notification.
Example implementation:
http://social.msdn.microsoft.com/Forums/en-US/rx/thread/a97a7e97-cb3d-4b42-afcb-8cb2c88c1778
Precedence is only guaranteed if all of the following are true:
ISubject
implementation natively ensures subscription precedence.ISubject
's callers obey the Rx Grammar and serialize notifications; e.g., calls to OnNext never overlap.ISubject
are never made concurrently. (Though subscriptions to the generated observables can be made concurrently.)Related discussion:
http://social.msdn.microsoft.com/Forums/en-US/rx/thread/ac721f91-4dbc-40b8-a2b2-19f00998239f
Copied from https://rx.codeplex.com/workitem/77
When TimeSpan.Zero
or TimeSpan
with a value < 1ms is passed to Observable.Generate
a System.StackOverflowException
is thrown. Looks like there is an infinite loop or a recursion happening.
Copied from https://rx.codeplex.com/workitem/72
We are using Rx "2.1.30214.0" with "net40" and got a NullReferenceException on a production machine - all we know is, that the NRE was caught but our recovery attempt failed...
---> (Interne Ausnahme #0) System.NullReferenceException: Der Objektverweis wurde nicht auf eine Objektinstanz festgelegt.
bei System.Reactive.Concurrency.VirtualTimeScheduler2.GetNext()
bei System.Reactive.Concurrency.VirtualTimeSchedulerBase2.Start()
bei ...OurCustomScheduler.Worker() in d:\Build...\OurCustomScheduler.cs:Zeile 70.
bei System.Threading.Tasks.Task.InnerInvoke()
bei System.Threading.Tasks.Task.Execute()<---
Our scheduler is derived from VirtualTimeScheduler<,> ...
class OurCustomScheduler : VirtualTimeScheduler<DateTime, ulong>, IDisposable { ... }
...and its main purpose is to run a Worker with
Task.Factory.StartNew(Worker, TaskCreationOptions.LongRunning);
...that will advance the (virtual) time eventually which will result in a call to VirtualTimeScheduler.GetNext()
to execute the next item(s). It is used like that:
subscription = Observable
.Interval(cycle, scheduler) // OurCustomScheduler
.Subscribe(...);
However, a NRE is thrown inside GetNext() and when we try a restart to recover
// recovery:
log.Message("Restarting scheduler.").Info();
Stop();
Start();
...the call to VirtualTimeSchedulerBase.Start()
calls GetNext()
again resulting in another NRE as the problem persisted.
The GetNext()
method does not contain much code - any ideas what caused the NullReferenceException?
(The queue is readonly...)
protected override IScheduledItem<TAbsolute> GetNext()
{
while (this.queue.Count > 0)
{
ScheduledItem<TAbsolute> scheduledItem = this.queue.Peek();
if (!scheduledItem.IsCanceled)
return (IScheduledItem<TAbsolute>) scheduledItem;
this.queue.Dequeue();
}
return (IScheduledItem<TAbsolute>) null;
}
Hello! I would expect the following code to be valid.
var source = Observable.Empty<BaseClass>();
var destination = source
.Where(x => x is DerivedClass)
.Cast<DerivedClass>();
However, Cast
takes an IObservable<object>
, and therefore the types are not compatible. Calling the method directly doesn't work either...
var source = Observable.Empty<BaseClass>();
var filtered = source
.Where(x => x is DerivedClass);
var destination = Observable.Cast<DerivedClass>(filtered);
In both cases, the following compilation error is produced.
Argument 1: cannot convert from 'System.IObservable<T>' to 'System.IObservable<object>'
Is there a reason to avoid doing something like this?
I discovered a bug in the LocalScheduler which is affecting one of our applications. The app makes use of ThreadPoolScheduler and DispatcherScheduler, which both inherit from the LocalScheduler base class. I have attached a minimal source file which duplicates the problem. If everything worked as expected, the program should print "GUI" and "THREADPOOL" forever. Instead, it will eventually print the "FAIL!" message.
The "FAIL!" case arises when LocalScheduler.ExecuteNextShortTermWorkItem runs on the ThreadPoolScheduler, but 'next' (the result of s_shortTerm.Dequeue) is a DispatcherScheduler WorkItem. In other words, the running thread is a ThreadPool thread (and 'scheduler' is a ThreadPoolScheduler), but 'next'._scheduler is a DispatcherScheduler. The appearance of the bug is also sensitive to the size of the application's delays relative to to the size of LocalScheduler.RETRYSHORT (which is currently 50 ms).
I'm not sure what is the best way to fix this. One way is to convert s_shortTerm and the related data structures into instance variables rather than static variables. That will keep the various workItems from mixing together. Another possibility is to store a separate short term set per subclass (one for DispatcherScheduler, one for ThreadPoolScheduler, etc). A third possibility is to make ExecuteNextShortTermWorkItem use 'next'._scheduler to run the work, but I'm not sure if that causes other problems (excessive delays, ordering issues, etc).
Program.cs
using System;
using System.Diagnostics;
using System.Reactive.Concurrency;
using System.Windows;
using System.Windows.Threading;
namespace RxSandbox4 {
public class ModelObject : DependencyObject {
public void Start() {
Schedule();
}
private void Schedule() {
var now=DateTimeOffset.Now;
const int dispatcherDelay=55;
const int threadPoolDelay=60;
new DispatcherScheduler(Dispatcher, DispatcherPriority.SystemIdle)
.Schedule(now.Add(TimeSpan.FromMilliseconds(dispatcherDelay)), RunMeOnGui);
ThreadPoolScheduler.Instance
.Schedule(now.Add(TimeSpan.FromMilliseconds(threadPoolDelay)), RunMeOnThreadPool);
}
private void RunMeOnThreadPool() {
Debug.Print("THREADPOOl");
}
private void RunMeOnGui() {
if(!CheckAccess()) {
Debug.Print("FAIL!");
return;
}
Debug.Print("GUI");
Schedule();
}
}
internal class Program {
private static void Main(string[] args) {
Dispatcher.CurrentDispatcher.InvokeAsync(() => {
var model=new ModelObject();
model.Start();
});
Dispatcher.Run();
}
}
}
Please let me know if you need any more information. Thanks.
Add support for Ix for supporting Portable Library format, Win4.5, WP8, WP81, Win8, Win81
Since Rx does not use WCF, the portable target should be updated to portable-win8+net45+wp8+wpa81
(profile259).
Although Rx.NET 2.2.4 adds Universal App support the PCL does not yet include it, or at least not according to the folder in the NuGet package.
Profile259 is the new hotness as discussed in reactiveui/splat#55, reactiveui/ReactiveUI#552 and MvvmCross/MvvmCross#660
/cc @paulcbetts @dsplaisted
Copied from https://rx.codeplex.com/workitem/76
If you build a Windows Phone 8.1 store app that uses Rx, and you run the Release build on an actual device (not the emulator), you will encounter MissingManifestResourceExceptions in scenarios in which Rx should be calling OnError
on its subscribers.
This does not happen in Debug builds. And it only happens in Release builds if you run the code on a real phone - the problem does not occur on the phone emulator.
To repro this, I just created a new 8.1 Phone Store App in Visual Studio 2013 (Update 2), and then added a NuGet reference to Rx-Main (2.2.4). Then in the MainPage.xaml.cs codebehind, I added this to the constructor:
var src = new Subject<int>();
IObservable<int> shouldError = src.FirstAsync();
shouldError.Subscribe(
i => Debug.WriteLine(i),
x => Debug.WriteLine("Error: " + x));
src.OnCompleted();
Since I'm using FirstAsync
on a source that completes without producing items, the expected outcome here is that my error handler should run. And if I run the Debug build (on either the emulator or a real phone) that's exactly what happens - I see this in my Debug output in VS:
Error: System.InvalidOperationException: Sequence contains no elements.
If I run the Release build on the emulator, there's no debug output of course, because the Debug.WriteLine
gets compiled out, but it runs without error.
But if I run the Release build on a real device (a Nokia Lumia 620) the app crashes - an unhandled MissingManifestResourceException emerges from that call to OnCompleted. Here's the detail:
System.Resources.MissingManifestResourceException occurred
_HResult=-2146233038
message=[MissingManifestResource_NoPRIresources]
Arguments:
Debugging resource strings are unavailable. Often the key and arguments provide sufficient information to diagnose the problem. See http://go.microsoft.com/fwlink/?linkid=106663&Version=4.0.30227.0&File=mscorlib.dll&Key=MissingManifestResource_NoPRIresources
HResult=-2146233038
IsTransient=false
Message=[MissingManifestResource_NoPRIresources]
Arguments:
Debugging resource strings are unavailable. Often the key and arguments provide sufficient information to diagnose the problem. See http://go.microsoft.com/fwlink/?linkid=106663&Version=4.0.30227.0&File=mscorlib.dll&Key=MissingManifestResource_NoPRIresources
Source=mscorlib
StackTrace:
at System.Resources.ResourceManager.GetString(String name, CultureInfo culture)
InnerException:
Here's the stack trace:
mscorlib.ni.dll!System.Resources.ResourceManager.GetString(string name, System.Globalization.CultureInfo culture)
System.Reactive.Linq.ni.DLL!System.Reactive.Strings_Linq.NO_ELEMENTS.get()
System.Reactive.Linq.ni.DLL!System.Reactive.Linq.ObservableImpl.FirstAsync..OnCompleted()
System.Reactive.Linq.ni.DLL!System.Reactive.Subjects.Subject.OnCompleted()
As far as I can tell, Rx is attempting to look up a localizable version of that "Sequence contains no elements." error message in order to construct the InvalidOperationException before passing that onto the subscriber's OnError. But because the resource lookup failed, the subscriber never sees anything, and instead the code calling Subject.OnCompleted receives this resource lookup exception.
Copied from https://rx.codeplex.com/workitem/14
Right now I am taking the RX Contract Assemblies Dave has made over in the RXX project and manually creating new NuGet packages to put in my company NuGet repo so we can get code contracts working with Rx. I'll have to do this everytime I upgrade versions of RX.
Is there any chance you can start including contract assemblies in the NuGet releases?
Copied from https://rx.codeplex.com/workitem/20
With Ix-Main 0.9.0.2 from the NuGet package, trying to use Yielder as in this example ( https://gist.github.com/mausch/5241729 ) throws a TypeLoadException
:
Unhandled Exception: System.TypeLoadException: Inheritance security rules violated while overriding member: 'System.Linq.Yielder`1<T>.UnsafeOnCompleted(System.Action)'. Security accessibility of the overriding method must match the security accessibility of the method being overriden.
at System.Linq.EnumerableEx.<Create>d__a6`1.MoveNext()
at System.Linq.Enumerable.ElementAt[TSource](IEnumerable`1 source, Int32 index)
at ConsoleApplication21.Program.Main(String[] args) in g:\prg\ConsoleApplication21\ConsoleApplication21\Program.cs:line 15
Earlier I copied the code from this diff ( http://rx.codeplex.com/SourceControl/changeset/8f1e73bca96e ) instead of using the NuGet package and the same example worked fine.
Copied from https://rx.codeplex.com/workitem/56
The following code causes deadlock. I think there is race condition when group is immediately closed by another thread.
using System;
using System.Reactive.Concurrency;
using System.Reactive.Linq;
using System.Reactive.Subjects;
namespace RxGroupByUntilDeadlock
{
class Program
{
static void Main(string[] args)
{
var writer = new Subject<int>();
var grouped = writer.GroupByUntil(
i => i % 5,
g => g.Take(1).Delay(i => Observable.Timer(DateTimeOffset.Now.AddMinutes(i * -1), TaskPoolScheduler.Default)));
using (grouped.Subscribe())
{
for (int i = 0; i < 100000; i++)
{
Console.WriteLine("Writing {0}", i);
writer.OnNext(i);
}
Console.WriteLine("Done");
Console.ReadLine();
}
}
}
}
In the current implementation of DefaultExceptionServices
the exception is thrown, this means a new stack trace is created and we only keep the actual message. e.g.
internal class DefaultExceptionServices : IExceptionServices
{
public void Rethrow(Exception exception)
{
throw exception;
}
}
To keep the stack trace it would be better to wrap the exception. e.g.
internal class DefaultExceptionServices : IExceptionServices
{
public void Rethrow(Exception exception)
{
throw new Exception(exception.Message, exception);
}
}
Copied from https://rx.codeplex.com/workitem/18
There are a number of help messages for obsolete methods and properties which have help messages which end with "See http://go.microsoft.com/fwlink/?LinkID=260866 for more information." However, this URL resolves to http://www.bing.com which doesn't really help.
Can these messages be updated to refer to some more specific content, or removed?
Locations:
◾Rx\NET\Source\System.Reactive.Core\Reactive\Internal\Constants.cs(11): public const string OBSOLETE_SCHEDULER_NEWTHREAD = OBSOLETE_REFACTORING + " Please add a reference to the System.Reactive.PlatformServices assembly for your target platform and use NewThreadScheduler.Default to obtain an instance of this scheduler type. See http://go.microsoft.com/fwlink/?LinkID=260866 for more information.";
◾Rx\NET\Source\System.Reactive.Core\Reactive\Internal\Constants.cs(12): public const string OBSOLETE_SCHEDULER_TASKPOOL = OBSOLETE_REFACTORING + " Please add a reference to the System.Reactive.PlatformServices assembly for your target platform and use TaskPoolScheduler.Default to obtain an instance of this scheduler type. See http://go.microsoft.com/fwlink/?LinkID=260866 for more information.";
◾Rx\NET\Source\System.Reactive.Core\Reactive\Internal\Constants.cs(13): public const string OBSOLETE_SCHEDULER_THREADPOOL = OBSOLETE_REFACTORING + " Consider using Scheduler.Default to obtain the platform's most appropriate pool-based scheduler. In order to access a specific pool-based scheduler, please add a reference to the System.Reactive.PlatformServices assembly for your target platform and use the appropriate scheduler in the System.Reactive.Concurrency namespace. See http://go.microsoft.com/fwlink/?LinkID=260866 for more information.";
◾Rx\NET\Source\System.Reactive.Core\Reactive\Internal\Constants.cs(15): public const string OBSOLETE_SCHEDULEREQUIRED = "This instance property is no longer supported. Use CurrentThreadScheduler.IsScheduleRequired instead. See http://go.microsoft.com/fwlink/?LinkID=260866 for more information.";
◾Rx\NET\Source\System.Reactive.Linq\Reactive\Internal\Constants.cs(14): public const string USE_ASYNC = "This blocking operation is no longer supported. Instead, use the async version in combination with C# and Visual Basic async/await support. In case you need a blocking operation, use Wait or convert the resulting observable sequence to a Task object and block. See http://go.microsoft.com/fwlink/?LinkID=260866 for more information.";
◾Rx\NET\Source\System.Reactive.Linq\Reactive\Internal\Constants.cs(15): public const string USE_TASK_FROMASYNCPATTERN = "This conversion is no longer supported. Replace use of the Begin/End asynchronous method pair with a new Task-based async method, and convert the result using ToObservable. If no Task-based async method is available, use Task.Factory.FromAsync to obtain a Task object. See http://go.microsoft.com/fwlink/?LinkID=260866 for more information.";
◾Rx\NET\Source\System.Reactive.Providers\Reactive\Internal\Constants.cs(14): public const string USE_ASYNC = "This blocking operation is no longer supported. Instead, use the async version in combination with C# and Visual Basic async/await support. In case you need a blocking operation, use Wait or convert the resulting observable sequence to a Task object and block. See http://go.microsoft.com/fwlink/?LinkID=260866 for more information.";
◾Rx\NET\Source\System.Reactive.Providers\Reactive\Internal\Constants.cs(15): public const string USE_TASK_FROMASYNCPATTERN = "This conversion is no longer supported. Replace use of the Begin/End asynchronous method pair with a new Task-based async method, and convert the result using ToObservable. If no Task-based async method is available, use Task.Factory.FromAsync to obtain a Task object. See http://go.microsoft.com/fwlink/?LinkID=260866 for more information.";
◾Rx\NET\Source\System.Reactive.Windows.Threading\Reactive\Internal\Constants.cs(10): public const string OBSOLETE_INSTANCE_PROPERTY = "Use the Current property to retrieve the DispatcherScheduler instance for the current thread's Dispatcher object. See http://go.microsoft.com/fwlink/?LinkID=260866 for more information.";
Copied from https://rx.codeplex.com/workitem/68
This has come up in RxJava development: ReactiveX/RxJava#831
Basically, when left and right emit their values in a way that, for example, the right's emission falls between L112 - L129 of Join.cs, both parties will emit left and right values. Is this expected or a bug?
If I add Rx-Linq (and consequently Rx-Core and Rx-Interfaces) to a Xamarin.Android project, NuGet adds the dlls from "packages\Rx-xyz.2.2.1-beta\lib\portable-windows8+net45+wp8". I guess that denotes Profile 78 and Xamarin.Android is supported by that profile.
However, the build fails: Exception while loading assemblies: System.IO.FileNotFoundException: Could not load assembly 'System.Runtime.InteropServices.WindowsRuntime, Version=4.0.0.0, Culture=neutral, PublicKeyToken=b03f5f7f11d50a3a'. Perhaps it doesn't exist in the Mono for Android profile?
The reference to System.Runtime.InteropServices.WindowsRuntime is due to System.Reactive.Linq. If it's really profile 78, should it reference that library?
On the other hand: If I add another PCL-project with profile 78 to the solution and add Rx-Linq to it, the same System.Reactive.Liq.dll from the same location is used - it builds just fine.
So this might as well be a problem with the Xamarin Linker, but should the reference to System.Runtime.InteropServices.WindowsRuntime not be there in the first place ?
Copied from https://rx.codeplex.com/workitem/55
This synchronous code throws an Exception, as I would expect:
Observable.Return(1).Subscribe(
_ => { throw new Exception(); });
However, I was hoping to find the following code would also throw an Exception but I was very surprised to find that it does not!
Observable.Return(1).ObserveOn(TaskPoolScheduler.Default).Subscribe(
_ => { throw new Exception(); });
Certainly switching to the ThreadPoolScheduler gives the expected Exception:
Observable.Return(1).ObserveOn(ThreadPoolScheduler.Instance).Subscribe(
_ => { throw new Exception(); });
Perhaps the Rx team can comment. It seems the behaviour is inconsistent?
This came up while I was looking at this question on StackOverflow
The OP there has a different experience yet again.
Copied from https://rx.codeplex.com/workitem/61
I noticed that BehaviorSubject doesn't hold a lock while it sends notifications. I assume this is to avoid some subtle deadlocks, but now there's no mechanism present to prevent notifications from being re-ordered.
I always assumed that subscribing to BehaviorSubject was eventually consistent with respect to the latest value. That the latest notification was the latest value. So...
◾Is that not the case?
◾Is that intentional?
◾What about enqueueing things to notify (while holding the lock) then draining the queue (outside the lock)?
Copied from https://rx.codeplex.com/workitem/63
Is it possible to have an Aggregate / Scan function overloads which pass in more than 1 item of history for input and output.
This would be very useful for Filters and Signal processing.
Copied from https://rx.codeplex.com/workitem/78
Compiled with VS 2012, with project type WP 8.0 the following code will fail if debugger is not attached.
Somehow, if debugger not attached, compiler optimizations ruins the code inside Crash() - see comments in code.
Tested on Lumia 1520 (8.1), Lumia 630 (8.0).
public partial class MainPage : PhoneApplicationPage
{
public MainPage()
{
InitializeComponent();
Button.Tap += (sender, args) => new A<B, string>(new B(), "string").Crash();
}
}
public class B
{
public void Foo<T>(T val) { }
}
public class A<T1, T2> where T1 : B
{
private T1 _t1;
private T2 _t2;
public A(T1 t1, T2 t2)
{
_t2 = t2;
_t1 = t1;
}
public void Crash()
{
var obs = Observable.Return(_t2);
obs.Subscribe(result =>
{
//CLR is expecting T2 to be System.String here,
//but somehow, after passing through Observable
//T2 here is not a string, it's A<T1, T2>
new List<T2>().Add(result);
});
//Will run normally if commented
_t1.Foo(new object());
}
}
Copied from https://rx.codeplex.com/workitem/42
Request to have the Timeout operators to accept an additional message parameter. This would be added as the TimeoutException message e.g. new TimeoutException(message).
Copied from https://rx.codeplex.com/workitem/43
Rx defines an overload of the Merge operator with a maxConcurrent parameter that is useful for controlling the level of concurrency through the number of active subscriptions to inner observables; however, the maxConcurrent parameter is static and in some cases it's impossible to write a query that changes the parameter dynamically. Unsubscribing and re-subscribing to Merge won't work for a hot source due to the fact that Merge uses an internal buffer. Unsubscribing will lose all of the inner observables that have already been buffered but aren't yet active, and since the source is hot, this causes permanent data loss.
Consider adding a Merge overload such as the one shown in the following discussion:
It has the following signature:
IObservable<TSource> Merge<TSource>(this IObservable<IObservable<TSource>> source, IObservable<int> maxConcurrent)
Copied from https://rx.codeplex.com/workitem/15
UPDATE: looking on other issues I found out that it is equals to Issue #55
when you put a Timer into the GroupJoin duration the process will crash.
The following snippet will reproduce it.
Notice that while changing the GetDuration method to use a custom timer it doesn't crash
static void Main(string[] args)
{
Console.WriteLine("Start");
IObservable<long> ls = Observable.Timer(TimeSpan.Zero, TimeSpan.FromSeconds(3));
IObservable<long> rs = Observable.Timer(TimeSpan.Zero, TimeSpan.FromSeconds(2));
var join = ls.GroupJoin(rs,
item => GetDuration(),
item => Observable.Empty<Unit>(),
(l, rStream) => Tuple.Create(l, rStream));
int i = 4;
join.Subscribe(js =>
{
int tmp = Interlocked.Increment(ref i);
var gStream = js.Item2;
gStream.Subscribe(v =>
{
Console.WriteLine("{0}[{1},{2}]", new string('\t', tmp * 2), js.Item1, v);
});
});
Console.ReadKey();
}
private static IObservable<long> GetDuration()
{
TimeSpan delay = TimeSpan.FromSeconds(5);
return Observable.Timer(delay, TimeSpan.Zero);
//return Observable.Create<long>(async o =>
// {
// await Task.Delay(delay);
// return Disposable.Empty;
// });
}
See this Rx discussion thread for the full details.
But generally speaking, observables ExecutionContext from the notification source down to the observer. Observer callbacks fire within the notification source's execution context.
For cold observables, the 2 contexts are usually the same. But for hot observables, the 2 contexts can be different and it becomes unpredictable to know what context your callbacks will run in. And this flow direction tends to be counter-intuitive and doesn't really align with the way Tasks flow the execution context.
Should this behavior be changed?
Copied from https://rx.codeplex.com/workitem/40
Looks like the group by implementation is not thread safe. Internally it uses a Dictionary.Add, in the onnext of the Sink, without using proper locking mechanisms.
if (!_map.TryGetValue(key, out writer)) {
writer = new Subject<TElement>();
_map.Add(key, writer);
fireNewMapEntry = true;
}
I noticed by subscribing to an event pattern that is invoked by multiple threads at the same time. (In my case 24 threads that potentially call TransportMessageReceived concurrently)
Observable.FromEventPattern<TransportMessageReceivedEventArgs>(
h => _bus.Transport.TransportMessageReceived += h,
h => _bus.Transport.TransportMessageReceived -= h);
And simply subscribe to it with a group by expression
var messagesReceived = from e in observable
group e by "something" into c
select c ;
subscription = messagesReceived.Subscribe(r => Console.WriteLine("Triggered") );
This will occasionally throw a null reference in Dictionary.Insert
... as that implementation is not thread safe by default.
If you build a Windows Phone 8.1 store app that uses Rx, and you run the Release build on an actual device (not the emulator), you will encounter MissingManifestResourceExceptions
in scenarios in which Rx should be calling OnError
on its subscribers.
This does not happen in Debug builds. And it only happens in Release builds if you run the code on a real phone - the problem does not occur on the phone emulator.
To repro this, I just created a new 8.1 Phone Store App in Visual Studio 2013 (Update 2), and then added a NuGet reference to Rx-Main (2.2.4). Then in the MainPage.xaml.cs
codebehind, I added this to the constructor:
var src = new Subject<int>();
IObservable<int> shouldError = src.FirstAsync();
shouldError.Subscribe(
i => Debug.WriteLine(i),
x => Debug.WriteLine("Error: " + x));
src.OnCompleted();
Since I'm using FirstAsync
on a source that completes without producing items, the expected outcome here is that my error handler should run. And if I run the Debug build (on either the emulator or a real phone) that's exactly what happens - I see this in my Debug output in VS:
Error: System.InvalidOperationException: Sequence contains no elements.
If I run the Release build on the emulator, there's no debug output of course, because the Debug.WriteLine
gets compiled out, but it runs without error.
But if I run the Release build on a real device (a Nokia Lumia 620) the app crashes - an unhandled MissingManifestResourceException
emerges from that call to OnCompleted
. Here's the detail:
System.Resources.MissingManifestResourceException occurred
_HResult=-2146233038
_message=[MissingManifestResource_NoPRIresources]
Arguments:
Debugging resource strings are unavailable. Often the key and arguments provide sufficient information to diagnose the problem. See http://go.microsoft.com/fwlink/?linkid=106663&Version=4.0.30227.0&File=mscorlib.dll&Key=MissingManifestResource_NoPRIresources
HResult=-2146233038
IsTransient=false
Message=[MissingManifestResource_NoPRIresources]
Arguments:
Debugging resource strings are unavailable. Often the key and arguments provide sufficient information to diagnose the problem. See http://go.microsoft.com/fwlink/?linkid=106663&Version=4.0.30227.0&File=mscorlib.dll&Key=MissingManifestResource_NoPRIresources
Source=mscorlib
StackTrace:
at System.Resources.ResourceManager.GetString(String name, CultureInfo culture)
InnerException:
Here's the stack trace:
mscorlib.ni.dll!System.Resources.ResourceManager.GetString(string name, System.Globalization.CultureInfo culture)
System.Reactive.Linq.ni.DLL!System.Reactive.Strings_Linq.NO_ELEMENTS.get()
System.Reactive.Linq.ni.DLL!System.Reactive.Linq.ObservableImpl.FirstAsync<int>._.OnCompleted()
System.Reactive.Linq.ni.DLL!System.Reactive.Subjects.Subject<int>.OnCompleted()
As far as I can tell, Rx is attempting to look up a localizable version of that "Sequence contains no elements." error message in order to construct the InvalidOperationException
before passing that onto the subscriber's OnError
. But because the resource lookup failed, the subscriber never sees anything, and instead the code calling Subject.OnCompleted
receives this resource lookup exception.
Copied from https://rx.codeplex.com/workitem/51
Consider the following C# code:
var result = Observable
.Interval(TimeSpan.FromSeconds(1))
.Do(x => Console.WriteLine("Interval.OnNext({0})", x))
.Take(30)
.Window(TimeSpan.FromSeconds(10))
.Do(x => Console.WriteLine("Window.OnNext"))
.Take(1)
// COMMENTED OUT // .Delay(TimeSpan.FromSeconds(4.5))
.Merge();
result.Subscribe(
onNext: x => Console.WriteLine("OnNext: {0} (time = {1})", x, DateTime.Now.TimeOfDay),
onCompleted: () => Console.WriteLine("OnCompleted"),
onError: e => Console.WriteLine("OnError: {0}", e));
Console.ReadLine();
This code (with Delay commented out) results in the following output:
Window.OnNext
Interval.OnNext(0)
OnNext: 0 (time = 12:58:21.3746995)
Interval.OnNext(1)
OnNext: 1 (time = 12:58:22.3713666)
Interval.OnNext(2)
OnNext: 2 (time = 12:58:23.3730377)
Interval.OnNext(3)
OnNext: 3 (time = 12:58:24.3747082)
Interval.OnNext(4)
OnNext: 4 (time = 12:58:25.3753784)
Interval.OnNext(5)
OnNext: 5 (time = 12:58:26.3760485)
Interval.OnNext(6)
OnNext: 6 (time = 12:58:27.3767189)
Interval.OnNext(7)
OnNext: 7 (time = 12:58:28.3773891)
Interval.OnNext(8)
OnNext: 8 (time = 12:58:29.3780589)
OnCompleted
Now uncomment the code to delay subscription to windows by 4.5 seconds. I would expect the modification to simply drop items in the first 4.5 seconds of a window, leaving the output of 5,6,7,8,… intact. Is my expectation reasonable? FWIW the actual behavior of Rx.NET is to emit nothing at all.
Copied from https://rx.codeplex.com/workitem/69
(ConsoleApplication1.zip attachment not included)
I haven't come up with a repro for this outside of our full system, but I'll be sure to update it when I do. In the mean time, I'll do my best to describe the system and how we get to the point of seeing this exception and attach a project that has the relevant code but nothing that generates data.
We're using Rx inside our autoscaling service to window and aggregate the utilization statistics the other workloads in the system are publishing. Workloads publish statistics using the pub/sub capabilities of Redis and we use a pattern subscription to monitor all workloads from the central autoscaler.
The attached project has been simplified somewhat from our production code, but I believe everything relevant should still be there. There are two simplifications that I wouldn't expect to be relevant but I want to mention them in case they're useful in diagnosing the issue.
One last note, we normally expect events to come at 5 second intervals so we're configuring a 2 minute buffer with 5 seconds between buffers. However, when we used time-based buffering instead, the problem seemed to occur more frequently.
The following exception is thrown intermittently and when it occurs I seem to lose my subscriptions entirely.
Object reference not set to an instance of an object.
at System.Collections.Generic.Dictionary`2.Insert(TKey key, TValue value, Boolean add)
at System.Reactive.Linq.ObservableImpl.GroupBy`3._.OnNext(TSource value)
--- End of stack trace from previous location where exception was thrown ---
at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
at System.Reactive.AnonymousSafeObserver`1.OnError(Exception error)
at System.Reactive.Linq.ObservableImpl.GroupBy`3._.Error(Exception exception)
at System.Reactive.Linq.ObservableImpl.GroupBy`3._.OnNext(TSource value)
at System.Reactive.Linq.ObservableImpl.Where`1._.OnNext(TSource value)
at Apprenda.SaaSGrid.Topology.Utility.KernelPubSubExtensions.<>c__DisplayClass8`1.<PatternSubscribe>b__7(String sourcePath, T data) in c:\code\trunk\SaaSGrid\SaaSGrid.Topology\Apprenda\SaaSGrid\Topology\Utility\KernelPubSubExtensions.cs:line 39
at Apprenda.Caching.Redis.RedisPubSub.<>c__DisplayClass5`1.<PatternSubscribe>b__3(String recievedTopic, Byte[] bytes) in c:\code\trunk\SaaSGrid\SaaSGrid.Caching\Apprenda\Caching\Redis\RedisPubSub.cs:line 63
at BookSleeve.RedisSubscriberConnection.RaiseEvent(Action`2 handler, String key, RedisResult value) in c:\Dev\booksleeve\BookSleeve\RedisSubscriberConnection.cs:line 92
--- End of stack trace from previous location where exception was thrown ---
at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
at System.Reactive.AnonymousSafeObserver`1.OnError(Exception error)
at System.Reactive.Linq.ObservableImpl.Select`2._.OnError(Exception error)
at System.Reactive.Linq.ObservableImpl.Buffer`1._.OnError(Exception error)
at System.Reactive.Linq.ObservableImpl.Select`2._.OnError(Exception error)
at System.Reactive.AutoDetachObserver`1.OnErrorCore(Exception exception)
at System.Reactive.Subjects.Subject`1.Subscribe(IObserver`1 observer)
at System.Reactive.Linq.GroupedObservable`2.SubscribeCore(IObserver`1 observer)
at System.Reactive.ObservableBase`1.Subscribe(IObserver`1 observer)
at System.ObservableExtensions.SubscribeSafe[T](IObservable`1 source, IObserver`1 observer)
at System.Reactive.Producer`1.SubscribeRaw(IObserver`1 observer, Boolean enableSafeguard)
at System.ObservableExtensions.SubscribeSafe[T](IObservable`1 source, IObserver`1 observer)
at System.Reactive.Linq.ObservableImpl.Buffer`1.Run(IObserver`1 observer, IDisposable cancel, Action`1 setSink)
at System.Reactive.Producer`1.SubscribeRaw(IObserver`1 observer, Boolean enableSafeguard)
at System.ObservableExtensions.SubscribeSafe[T](IObservable`1 source, IObserver`1 observer)
at System.Reactive.Producer`1.Run(IScheduler _, State x)
at System.Reactive.Concurrency.ScheduledItem`2.InvokeCore()
at System.Reactive.Concurrency.ScheduledItem`1.Invoke()
at System.Reactive.Concurrency.CurrentThreadScheduler.Trampoline.Run(SchedulerQueue`1 queue)
at System.Reactive.Concurrency.CurrentThreadScheduler.Schedule[TState](TState state, TimeSpan dueTime, Func`3 action)
at System.Reactive.Concurrency.LocalScheduler.Schedule[TState](TState state, Func`3 action)
at System.Reactive.Producer`1.SubscribeRaw(IObserver`1 observer, Boolean enableSafeguard)
at Apprenda.Monitoring.Scaling.Autoscaler.ObserveStatistics() in c:\code\trunk\ApprendaCore\Apprenda.Monitoring\Scaling\Autoscaler.cs:line 66
at Apprenda.Monitoring.Scaling.Autoscaler..ctor(Component component, IObservable`1 deploymentEventsByComponent, AutoscalingController controller, IObservable`1 statisticsStream) in c:\code\trunk\ApprendaCore\Apprenda.Monitoring\Scaling\Autoscaler.cs:line 50
at Apprenda.Monitoring.Scaling.AutoscalingController.<>c__DisplayClass14.<HandleNewComponentStream>b__e(Guid _) in c:\code\trunk\ApprendaCore\Apprenda.Monitoring\Scaling\AutoscalingController.cs:line 70
at System.Collections.Concurrent.ConcurrentDictionary`2.GetOrAdd(TKey key, Func`2 valueFactory)
at Apprenda.Monitoring.Scaling.AutoscalingController.HandleNewComponentStream(Guid componentId, IObservable`1 componentStream) in c:\code\trunk\ApprendaCore\Apprenda.Monitoring\Scaling\AutoscalingController.cs:line 67
--- End of stack trace from previous location where exception was thrown ---
at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
at System.Reactive.AnonymousSafeObserver`1.OnError(Exception error)
at System.Reactive.Linq.ObservableImpl.Select`2._.OnError(Exception error)
at System.Reactive.Linq.ObservableImpl.Buffer`1._.OnError(Exception error)
at System.Reactive.Linq.ObservableImpl.Select`2._.OnError(Exception error)
at System.Reactive.AutoDetachObserver`1.OnErrorCore(Exception exception)
at System.Reactive.Subjects.Subject`1.Subscribe(IObserver`1 observer)
at System.Reactive.Linq.GroupedObservable`2.SubscribeCore(IObserver`1 observer)
at System.Reactive.ObservableBase`1.Subscribe(IObserver`1 observer)
at System.ObservableExtensions.SubscribeSafe[T](IObservable`1 source, IObserver`1 observer)
at System.Reactive.Producer`1.SubscribeRaw(IObserver`1 observer, Boolean enableSafeguard)
at System.ObservableExtensions.SubscribeSafe[T](IObservable`1 source, IObserver`1 observer)
at System.Reactive.Linq.ObservableImpl.Buffer`1.Run(IObserver`1 observer, IDisposable cancel, Action`1 setSink)
at System.Reactive.Producer`1.SubscribeRaw(IObserver`1 observer, Boolean enableSafeguard)
at System.ObservableExtensions.SubscribeSafe[T](IObservable`1 source, IObserver`1 observer)
at System.Reactive.Producer`1.Run(IScheduler _, State x)
at System.Reactive.Concurrency.ScheduledItem`2.InvokeCore()
at System.Reactive.Concurrency.ScheduledItem`1.Invoke()
at System.Reactive.Concurrency.CurrentThreadScheduler.Trampoline.Run(SchedulerQueue`1 queue)
at System.Reactive.Concurrency.CurrentThreadScheduler.Schedule[TState](TState state, TimeSpan dueTime, Func`3 action)
at System.Reactive.Concurrency.LocalScheduler.Schedule[TState](TState state, Func`3 action)
at System.Reactive.Producer`1.SubscribeRaw(IObserver`1 observer, Boolean enableSafeguard)
at Apprenda.Monitoring.Scaling.Autoscaler.ObserveStatistics() in c:\code\trunk\ApprendaCore\Apprenda.Monitoring\Scaling\Autoscaler.cs:line 66
at Apprenda.Monitoring.Scaling.Autoscaler.ResetStatisticsWindow() in c:\code\trunk\ApprendaCore\Apprenda.Monitoring\Scaling\Autoscaler.cs:line 238
--- End of stack trace from previous location where exception was thrown ---
at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
at System.Reactive.AnonymousSafeObserver`1.OnError(Exception error)
at System.Reactive.Linq.ObservableImpl.Select`2._.OnError(Exception error)
at System.Reactive.Linq.ObservableImpl.Buffer`1._.OnError(Exception error)
at System.Reactive.Linq.ObservableImpl.Select`2._.OnError(Exception error)
at System.Reactive.AutoDetachObserver`1.OnErrorCore(Exception exception)
at System.Reactive.Subjects.Subject`1.Subscribe(IObserver`1 observer)
at System.Reactive.Linq.GroupedObservable`2.SubscribeCore(IObserver`1 observer)
at System.Reactive.ObservableBase`1.Subscribe(IObserver`1 observer)
at System.ObservableExtensions.SubscribeSafe[T](IObservable`1 source, IObserver`1 observer)
at System.Reactive.Producer`1.SubscribeRaw(IObserver`1 observer, Boolean enableSafeguard)
at System.ObservableExtensions.SubscribeSafe[T](IObservable`1 source, IObserver`1 observer)
at System.Reactive.Linq.ObservableImpl.Buffer`1.Run(IObserver`1 observer, IDisposable cancel, Action`1 setSink)
at System.Reactive.Producer`1.SubscribeRaw(IObserver`1 observer, Boolean enableSafeguard)
at System.ObservableExtensions.SubscribeSafe[T](IObservable`1 source, IObserver`1 observer)
at System.Reactive.Producer`1.Run(IScheduler _, State x)
at System.Reactive.Concurrency.ScheduledItem`2.InvokeCore()
at System.Reactive.Concurrency.ScheduledItem`1.Invoke()
at System.Reactive.Concurrency.CurrentThreadScheduler.Trampoline.Run(SchedulerQueue`1 queue)
at System.Reactive.Concurrency.CurrentThreadScheduler.Schedule[TState](TState state, TimeSpan dueTime, Func`3 action)
at System.Reactive.Concurrency.LocalScheduler.Schedule[TState](TState state, Func`3 action)
at System.Reactive.Producer`1.SubscribeRaw(IObserver`1 observer, Boolean enableSafeguard)
at Apprenda.Monitoring.Scaling.Autoscaler.ObserveStatistics() in c:\code\trunk\ApprendaCore\Apprenda.Monitoring\Scaling\Autoscaler.cs:line 66
Copied from https://rx.codeplex.com/workitem/34
Silverlight 5 should support all (or almost all) of the portable library API, please add it to the list of supported platforms for the portable library.
By not supporting Silverlight 5 you prevent everyone else from building portable libraries compatible with SL5 from using Rx. If you wan't async support just utilize the microsoft.bcl.async package since it add async support
Copied from https://rx.codeplex.com/workitem/60
When using an timer in Rx (Net), if the user changes the system clock back somewhere over a month an exception (ArgumentOutOfRangeException
) is raised:
Time-out interval must be less than 2^32-2.
Parameter name: dueTm
Stack trace:
at System.Threading.Timer..ctor(TimerCallback callback, Object state, TimeSpan dueTime, TimeSpan period)
at System.Reactive.Concurrency.ConcurrencyAbstractionLayerImpl.Timer..ctor(Action`1 action, Object state, TimeSpan dueTime)
at System.Reactive.Concurrency.ConcurrencyAbstractionLayerImpl.StartTimer(Action`1 action, Object state, TimeSpan dueTime)
at System.Reactive.Concurrency.LocalScheduler.UpdateLongTermProcessingTimer()
at System.Reactive.Concurrency.LocalScheduler.EvaluateLongTermQueue(Object state)
at System.Reactive.Concurrency.LocalScheduler.SystemClockChanged(Object sender, SystemClockChangedEventArgs args)
at System.Reactive.PlatformServices.SystemClock.OnSystemClockChanged(Object sender, SystemClockChangedEventArgs e)
at System.Reactive.PlatformServices.PeriodicTimerSystemClockMonitor.TimeChanged()
at System.Reactive.Concurrency.ConcurrencyAbstractionLayerImpl.PeriodicTimer.Tick(Object state)
at System.Threading.TimerQueueTimer.CallCallbackInContext(Object state)
at System.Threading.ExecutionContext.RunInternal(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean preserveSyncCtx)
at System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean preserveSyncCtx)
at System.Threading.TimerQueueTimer.CallCallback()
at System.Threading.TimerQueueTimer.Fire()
at System.Threading.TimerQueue.FireNextTimers()
at System.Threading.TimerQueue.AppDomainTimerCallback()
The exception is unhandled and causes the host application to crash.
A related issue was reported on the Rx MSDN Forums back in 2011:
◾ BufferWithTime
exception when user changes system clock
Copied from https://rx.codeplex.com/workitem/62
(Rx Design Guidelines.pdf attachment not included)
I thought it would be good to add the Rx Design Guidelines document to source control, and maintain it on an ongoing basis. It's proved extremely useful to me on several occasions.
Current found on MSDN at:
http://go.microsoft.com/fwlink/?LinkID=205219
And attached.
In the code below, I would expect the result to be always 1, but when using async/await keywords with IObservable (in function AsyncWithObservable), it somehow ignores the first value produced and prints 2. Furthermore, if I remove the 100ms delay in ConnectAndProduceFirstValue(), then AsyncWithTaskRun() and WithTaskRun() never return. I am not entirely sure if those results are to be expected, but that behavior is surprising.
using System;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Reactive.Threading.Tasks;
using System.Threading;
using System.Threading.Tasks;
namespace ConsoleApplication1
{
class Program
{
static void Main(string[] args)
{
var s = new Subject<int>();
Task<int> t;
for (int i = 0; i < 1000; i++)
{
t = AsyncWithObservable(s);
s.OnNext(2);
t.Wait();
Console.WriteLine("AsyncWithObservable: " + t.Result); // prints 2
t = AsyncWithTask(s);
s.OnNext(2);
t.Wait();
Console.WriteLine("AsyncWithTask: " + t.Result); // prints 1
t = AsyncWithTaskRun(s);
s.OnNext(2);
t.Wait();
Console.WriteLine("AsyncWithTaskRun: " + t.Result); // prints 1, blocks if Thread.Sleep is removed
t = WithTask(s);
s.OnNext(2);
t.Wait();
Console.WriteLine("WithTask: " + t.Result); // prints 1
t = WithTaskRun(s);
s.OnNext(2);
t.Wait();
Console.WriteLine("WithTaskRun: " + t.Result); // prints 1, blocks if Thread.Sleep is removed
}
Console.ReadLine();
}
static void ConnectAndProduceFirstValue(Subject<int> s)
{
Thread.Sleep(100);
s.OnNext(1);
}
static async Task<int> AsyncWithObservable(Subject<int> s)
{
var t = s.FirstAsync();
ConnectAndProduceFirstValue(s);
return await t;
}
static async Task<int> AsyncWithTask(Subject<int> s)
{
var t = s.FirstAsync().ToTask();
ConnectAndProduceFirstValue(s);
return await t;
}
static async Task<int> AsyncWithTaskRun(Subject<int> s)
{
var t = Task.Run(() => s.First());
ConnectAndProduceFirstValue(s);
return await t;
}
static Task<int> WithTask(Subject<int> s)
{
var t = s.FirstAsync().ToTask();
ConnectAndProduceFirstValue(s);
return t;
}
static Task<int> WithTaskRun(Subject<int> s)
{
var t = Task.Run(() => s.First());
ConnectAndProduceFirstValue(s);
return t;
}
}
}
Copied from https://rx.codeplex.com/workitem/70
The Rx, Ix, and other related packages in NuGet (such as http://www.nuget.org/packages/Rx-Main/)
has the link to the license page.
However it refers to http://go.microsoft.com/fwlink/?LinkID=261272, not Apache License.
According to the license page in this project site (https://rx.codeplex.com/license), all resources by
this project are licensed under the Apache License 2.0, and I had found no exception clause.
I think that the link to the license page of NuGet packages might be a mistake and should be fixed, or
if this is correct, the notice of special licensing for NuGet packages should be mentioned in this project site.
Copied from https://rx.codeplex.com/workitem/82
The problems of backpressure (fast producer, slow consumer) and unbounded queues in native operators (e.g., Zip, GroupBy, Merge) appear to be solved in a general way in RxJava by incorporating a Producer interface as a first-class citizen.
https://speakerdeck.com/benjchristensen/reactive-streams-with-rx-at-javaone-2014?slide=67
https://github.com/reactive-streams/reactive-streams
It's worth considering whether it makes sense for - and how - it could be implemented in Rx.NET.
However, my initial impression is that it may not be an optimal approach. For example, how does an observer know what number to pass to the producer's request method? What happens when it's wrong?
It seems to me that simply adding overloads to problematic operators to accept a maximum queue size avoids the unbounded problem in a simple and effective way. Furthermore, accepting an IObservable instead would make it reactive (similar to this work item). Ensuring that the supplied observable pushes notifications in the same context as the producer allows for synchronized change to the queue size, presumably similar to how RxJava's Producer works. For example:
IO<R> Zip<F,S,R>(IO<F> f, IO<S> s, Func<F,S,R> sel, IObservable<int> maxQueueSize)
As for backpressure, in Rxx, I've implemented a pattern that I call Introspection to solve a similar problem with backpressure, though it does not solve the problem of unbounded queues in a general way.
IObservable<IList<TSource>> WindowIntrospective<TSource>(this IObservable<TSource> source, IScheduler scheduler)
The WindowIntrospective operator (pooly named) in Rxx works close to the metal in Rx to dynamically produce heterogeneously-sized buffers, to accommodate slow observers, as follows:
*i.e., we presume it's faster than the observer, but it need not be. In fact, the observer's speed can vary dynamically. It's one of the benefits over having an explicit Producer.request(#) method, from which an observer must pull. The introspective operators push heterogeneously-sized buffers that vary dynamically based on the dynamically varying performance of the observer.
** Often the next operator applied is ObserveOnDispatcher anyway, which adds an unbounded queue. The introspective operator has the same semantics without an unbounded queue -- it only assumes that the observer will not block OnNext indefinitely.
Copied from https://rx.codeplex.com/workitem/74
Please, observe the following unit test:
using System;
using System.Reactive.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
namespace UnitTests
{
[TestClass]
public class TestRx
{
public const int UNIT_TEST_TIMEOUT = 5000;
private static IObservable<int> GetObservable(int count = 100, int msWait = 10)
{
return Observable.Create<int>(async (obs, cancellationToken) =>
{
for (int i = 0; i < count && !cancellationToken.IsCancellationRequested; ++i)
{
int value = i;
obs.OnNext(await Task.Factory.StartNew(() =>
{
Thread.Sleep(msWait);
return value;
}));
}
});
}
[TestMethod, TestCategory("CI"), Timeout(UNIT_TEST_TIMEOUT)]
public void Subscribe()
{
var tcs = new TaskCompletionSource<object>();
int i = 0;
GetObservable().Subscribe(n =>
{
Assert.AreEqual(i, n);
++i;
}, e => Assert.Fail(), () =>
{
Assert.AreEqual(100, i);
tcs.TrySetResult(null);
});
tcs.Task.Wait();
}
[TestMethod, TestCategory("CI"), Timeout(UNIT_TEST_TIMEOUT)]
public void SubscribeCancel()
{
var tcs = new TaskCompletionSource<object>();
var cts = new CancellationTokenSource();
int i = 0;
GetObservable().Subscribe(n =>
{
Assert.AreEqual(i, n);
++i;
if (i == 5)
{
cts.Cancel();
}
}, e =>
{
Assert.IsTrue(i < 100);
tcs.TrySetResult(null);
}, () =>
{
Assert.IsTrue(i < 100);
tcs.TrySetResult(null);
}, cts.Token);
tcs.Task.Wait();
}
[TestMethod, TestCategory("CI"), Timeout(UNIT_TEST_TIMEOUT)]
public void SubscribeThrow()
{
var tcs = new TaskCompletionSource<object>();
int i = 0;
GetObservable().Subscribe(n =>
{
Assert.AreEqual(i, n);
++i;
if (i == 5)
{
throw new Exception("xo-xo");
}
}, e =>
{
Assert.AreEqual("xo-xo", e.Message);
tcs.TrySetResult(null);
}, Assert.Fail);
tcs.Task.Wait();
}
}
}
The Subscribe unit test passes, but SubscribeCancel
and SubscribeThrow
time out, because OnError
is never called.
A declarative, efficient, and flexible JavaScript library for building user interfaces.
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google ❤️ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.